flask_restx_marshmallow.util

utils of flask_restx_marshmallow.

Request

Bases: Request

The request object that is used by default in this project.

Response

Bases: Response

The response object that is used by default in this project.

classproperty

classproperty(func: Callable[..., T])

Class property.

Initialize class property.

Parameters:

Name	Type	Description	Default
func	Callable[P, T]	function get	required

Source code in flask_restx_marshmallow/util.py

def __init__(self: Self, func: Callable[..., T]) -> None:
    """Initialize class property.

    Args:
        func (Callable[P, T]): function get
    """
    self.fget = func
    self.__doc__ = func.__doc__
    self.__name__ = func.__name__
    self.slot_name = f"_class_{self.__name__}"
    self.__module__ = func.__module__

__get__

__get__(instance: None, owner: type[object]) -> T

__get__(instance: object, owner: type[object]) -> T

__get__(instance, owner) -> T

Get the value.

Parameters:

Name	Type	Description	Default
instance	object	instance	required
owner	type[object]	owner	required

Returns:

Name	Type	Description
T	T	value

Source code in flask_restx_marshmallow/util.py

def __get__(self: Self, instance, owner) -> T:
    """Get the value.

    Args:
        instance (object): instance
        owner (type[object]): owner

    Returns:
        T: value
    """
    return classmethod(self.fget).__get__(None, owner)()

authorization_helper

authorization_helper(
    authorization: Mapping[str, Mapping] | None,
) -> dict[
    str,
    ApiKeySecurityScheme
    | HttpSecurityScheme
    | OAuth2SecurityScheme
    | OpenIdConnectSecurityScheme
    | MutualTLSSecurityScheme,
]

Authorization helper.

Parameters:

Name	Type	Description	Default
authorization	Mapping	authorization	required

Raises:

Type	Description
TypeError	if authorization is not Mapping
ValueError	if required fields are missing

Returns:

Type	Description
dict[str, ApiKeySecurityScheme | HttpSecurityScheme | OAuth2SecurityScheme | OpenIdConnectSecurityScheme | MutualTLSSecurityScheme]	dict[str, ApiKeySecurityScheme | HttpSecurityScheme | OAuth2SecurityScheme | OpenIdConnectSecurityScheme | MutualTLSSecurityScheme]: security scheme

Source code in flask_restx_marshmallow/util.py

def authorization_helper(  # noqa: C901, PLR0912
    authorization: Mapping[str, Mapping] | None,
) -> dict[
    str,
    ApiKeySecurityScheme
    | HttpSecurityScheme
    | OAuth2SecurityScheme
    | OpenIdConnectSecurityScheme
    | MutualTLSSecurityScheme,
]:
    """Authorization helper.

    Args:
        authorization (Mapping): authorization

    Raises:
        TypeError: if authorization is not Mapping
        ValueError: if required fields are missing

    Returns:
        dict[str, ApiKeySecurityScheme | HttpSecurityScheme | OAuth2SecurityScheme | OpenIdConnectSecurityScheme | MutualTLSSecurityScheme]: security scheme
    """  # noqa: E501
    if authorization is None:
        return {}
    if not isinstance(authorization, Mapping) or not all(
        isinstance(s, Mapping) for s in authorization.values()
    ):
        msg = "authorization must be Mapping[str, Mapping]"
        raise TypeError(msg)
    authorization_ = dict[
        str,
        ApiKeySecurityScheme
        | HttpSecurityScheme
        | OAuth2SecurityScheme
        | OpenIdConnectSecurityScheme
        | MutualTLSSecurityScheme,
    ]()
    for name, security_scheme in authorization.items():
        match security_scheme.get("type"):
            case "apiKey":
                authorization_[name] = {
                    "type": "apiKey",
                    "in": security_scheme.get("in", "header"),
                    "name": security_scheme.get("name", "Authorization"),
                }
                if (description := security_scheme.get("description")) and isinstance(
                    description,
                    str,
                ):
                    authorization_[name]["description"] = description
            case "http":
                http_ = HttpSecurityScheme(
                    type="http",
                    scheme=security_scheme.get("scheme", "bearer"),
                )
                if (description := security_scheme.get("description")) and isinstance(
                    description,
                    str,
                ):
                    http_["description"] = description
                if (
                    bearer_format := security_scheme.get("bearerFormat")
                ) and isinstance(bearer_format, str):
                    http_["bearerFormat"] = bearer_format
                authorization_[name] = http_
            case "oauth2":
                if (flows := security_scheme.get("flows")) is None:
                    msg = "flows is required."
                    raise ValueError(msg)
                authorization_[name] = OAuth2SecurityScheme(type="oauth2", flows=flows)
                if (description := security_scheme.get("description")) and isinstance(
                    description,
                    str,
                ):
                    authorization_[name]["description"] = description
            case "openIdConnect":
                if (url := security_scheme.get("openIdConnectUrl")) is None:
                    msg = "openIdConnectUrl is required."
                    raise ValueError(msg)
                authorization_[name] = OpenIdConnectSecurityScheme(
                    type="openIdConnect",
                    openIdConnectUrl=url,
                )
                if (description := security_scheme.get("description")) and isinstance(
                    description,
                    str,
                ):
                    authorization_[name]["description"] = description
            case "mutualTLS":
                authorization_[name] = MutualTLSSecurityScheme(type="mutualTLS")
                if (description := security_scheme.get("description")) and isinstance(
                    description,
                    str,
                ):
                    authorization_[name]["description"] = description
    return authorization_

class_schema

class_schema(
    clazz: type[C],
    *,
    base: type[B],
    **kwargs: Unpack[ClassSchemaKwargs],
) -> type[B]

class_schema(
    clazz: type[C],
    *,
    base: type[B],
    **kwargs: Unpack[ClassSchemaKwargs],
) -> type[GenericParameters[C]]

class_schema(
    clazz, base, *, use_origin_load=False, **kwargs
)

Class schema.

Parameters:

Name	Type	Description	Default
clazz	type[dataclass]	dataclass type	required
base	type[Schema | Parameters]	base schema class. Defaults to None.	required
use_origin_load	bool	use origin load. Defaults to False.	False
**kwargs	Unpack[ClassSchemaKwargs]	kwargs	{}

Returns:

Type	Description
	Schema | Parameters: schema or parameters for annotations

Source code in flask_restx_marshmallow/util.py

def class_schema(clazz, base, *, use_origin_load=False, **kwargs):
    """Class schema.

    Args:
        clazz (type[dataclass]): dataclass type
        base (type[Schema | Parameters], optional): base schema class. Defaults to None.
        use_origin_load (bool, optional): use origin load. Defaults to False.
        **kwargs (Unpack[ClassSchemaKwargs]): kwargs

    Returns:
        Schema | Parameters: schema or parameters for annotations
    """
    schema = marshmallow_dataclass_class_schema(clazz, base, **kwargs)
    if not schema.__name__.endswith("Schema") and not schema.__name__.endswith(
        "Parameters",
    ):
        schema.__name__ = (
            f"{clazz.__name__}Schema"
            if issubclass(base, Schema)
            else f"{clazz.__name__}Parameters"
        )
    if use_origin_load:
        schema.load = base.load  # type: ignore[attr-defined]
    return schema

download_files

download_files(current_version: str, filename: str) -> None

Retrieve swagger ui files.

Source code in flask_restx_marshmallow/util.py

def download_files(current_version: str, filename: str) -> None:
    """Retrieve swagger ui files."""
    with Path(Path(__file__).parent, "static", filename).open("wb") as f:
        try:
            res = get(
                f"https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/{current_version}/{filename}",
                timeout=5,
            ).content
            f.write(res)
        except (ConnectionError, ReadTimeout):
            copyfile(
                Path(Path(__file__).parent, "static", f"{filename}.bak"),
                Path(Path(__file__).parent, "static", filename),
            )

get_location

get_location(field: Field) -> Location | None

Get location.

Parameters:

Name	Type	Description	Default
field	Field	field	required

Returns:

Name	Type	Description
Location	Location | None	location

Source code in flask_restx_marshmallow/util.py

def get_location(field: "Field") -> Location | None:
    """Get location.

    Args:
        field (Field): field

    Returns:
        Location: location
    """
    if not isinstance(field, List | Tuple):
        if isinstance(location := field.metadata.get("location"), str):
            location = Location(location)
        return location
    if isinstance(field, List):
        return get_location(cast("Field", field.inner))
    return get_location(cast("Field", field.tuple_fields[0]))

handle_invalid_json_error

handle_invalid_json_error(
    err: BadRequest | None = None,
) -> TupleResponse

Bad request.

Parameters:

Name	Type	Description	Default
err	BadRequest	bad request. Defaults to None.	None

Returns:

Name	Type	Description
TupleResponse	TupleResponse	response

Source code in flask_restx_marshmallow/util.py

def handle_invalid_json_error(err: "BadRequest | None" = None) -> TupleResponse:
    """Bad request.

    Args:
        err (BadRequest, optional): bad request. Defaults to None.

    Returns:
        TupleResponse: response
    """
    return TupleResponse(
        response=Error(message=getattr(err, "description", None)),
        status_code=HTTPStatus.BAD_REQUEST,
    )

handle_parse_validation_error

handle_parse_validation_error(
    err: ValidationError, *_args, **_kwargs
) -> NoReturn

Handles a Marshmallow ValidationError.

By attaching a standardized error response and re-raising the exception.

Parameters:

Name	Type	Description	Default
err	ValidationError	The validation error instance to handle.	required

Source code in flask_restx_marshmallow/util.py

def handle_parse_validation_error(
    err: "ValidationError",
    *_args,  # noqa: ANN002
    **_kwargs,  # noqa: ANN003
) -> "NoReturn":
    """Handles a Marshmallow ValidationError.

    By attaching a standardized error response and re-raising the exception.

    Args:
        err (ValidationError): The validation error instance to handle.
    """
    err.data = Error(
        code=ResponseCode.ERROR,
        message=next(iter(cast("dict", err.messages).values())),
    )
    raise err

handle_validation_error

handle_validation_error(
    error: UnprocessableEntity | None = None,
) -> TupleResponse

Validation error.

Returns:

Name	Type	Description
TupleResponse	TupleResponse	response

Source code in flask_restx_marshmallow/util.py

def handle_validation_error(
    error: "UnprocessableEntity | None" = None,
) -> TupleResponse:
    """Validation error.

    Returns:
        TupleResponse: response
    """
    return TupleResponse(
        response=Error(message=(error or UnprocessableEntity()).description),
        status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
    )

make_default_schema

make_default_schema(
    name: str,
    *,
    success: bool = True,
    code: HTTPStatus | int = OK,
    message: str | None = None,
) -> type[Schema]

Make standard schema.

Parameters:

Name	Type	Description	Default
name	str	name	required
success	bool	success. Defaults to True.	True
code	HTTPStatus | int	code. Defaults to HTTPStatus.OK.	OK
message	str | None	message. Defaults to None.	None

Returns:

Type	Description
type[Schema]	type[Schema]: schema

Source code in flask_restx_marshmallow/util.py

def make_default_schema(
    name: str,
    *,
    success: bool = True,
    code: HTTPStatus | int = HTTPStatus.OK,
    message: str | None = None,
) -> type["Schema"]:
    """Make standard schema.

    Args:
        name (str): name
        success (bool, optional): success. Defaults to True.
        code (HTTPStatus | int, optional): code. Defaults to HTTPStatus.OK.
        message (str | None, optional): message. Defaults to None.

    Returns:
        type[Schema]: schema
    """
    return class_schema(
        make_dataclass(
            name,
            [
                (
                    "code",
                    ResponseCode,
                    field(
                        default=ResponseCode.SUCCESS
                        if success is True
                        else ResponseCode.ERROR
                        if success is False
                        else code,
                        metadata={
                            "by_value": Integer,
                            "metadata": {
                                "description": "response code",
                                "example": ResponseCode.SUCCESS
                                if success is True
                                else ResponseCode.ERROR
                                if success is False
                                else code,
                            },
                        },
                    ),
                ),
                (
                    "message",
                    str | None,
                    field(
                        default=message
                        or (
                            code.description
                            if isinstance(code, HTTPStatus)
                            and code is not HTTPStatus.OK
                            else None
                        )
                        or ("Request success" if success is True else "Request failed"),
                        metadata={
                            "metadata": {
                                "description": "response message",
                                "example": message
                                or (
                                    code.description
                                    if isinstance(code, HTTPStatus)
                                    and code is not HTTPStatus.OK
                                    else None
                                )
                                or (
                                    "Request success"
                                    if success is True
                                    else "Request failed"
                                ),
                            },
                        },
                    ),
                ),
            ],
        ),
        base=Schema,
    )

output_json

output_json(
    data: dict | str | bytes,
    code: int | HTTPStatus,
    headers: dict | None = None,
) -> Response

Makes a Flask response with a JSON encoded body.

Parameters:

Name	Type	Description	Default
data	dict | str | bytes	data	required
code	int | HTTPStatus	code	required
headers	dict	headers. Defaults to None.	None

Returns:

Name	Type	Description
Response	Response	response

Source code in flask_restx_marshmallow/util.py

def output_json(
    data: dict | str | bytes,
    code: int | HTTPStatus,
    headers: dict | None = None,
) -> Response:
    """Makes a Flask response with a JSON encoded body.

    Args:
        data (dict | str | bytes): data
        code (int | HTTPStatus): code
        headers (dict, optional): headers. Defaults to None.

    Returns:
        Response: response
    """
    dumped = (
        dumps
        if isinstance(
            (dumps := (json.dumps(data) if isinstance(data, dict) else data)),
            str,
        )
        else cast("bytes", dumps).decode()
    ) + "\n"
    return Response(
        response=dumped,
        status=code,
        headers=headers,
        mimetype="application/json",
    )

raise_validation_error

raise_validation_error(
    error: ValidationError,
    _req: Request,
    _schema: Parameters,
    error_status_code: int | HTTPStatus | None = None,
    error_headers: dict | None = None,
) -> NoReturn

Validation error.

Parameters:

Name	Type	Description	Default
error	ValidationError	validation error	required
_req	Request	request	required
_schema	Parameters	schema	required
error_status_code	int | HTTPStatus | None	error status code. Defaults to None.	None
error_headers	dict | None	error headers. Defaults to None.	None

Raises:

Type	Description
UnprocessableEntity	validation error

Source code in flask_restx_marshmallow/util.py

def raise_validation_error(
    error: "ValidationError",
    _req: "Request",
    _schema: "Parameters",
    error_status_code: int | HTTPStatus | None = None,  # noqa: ARG001
    error_headers: dict | None = None,  # noqa: ARG001
) -> NoReturn:
    """Validation error.

    Args:
        error (ValidationError): validation error
        _req (Request): request
        _schema (Parameters): schema
        error_status_code (int | HTTPStatus | None, optional): error status code. Defaults to None.
        error_headers (dict | None, optional): error headers. Defaults to None.

    Raises:
        UnprocessableEntity: validation error
    """  # noqa: E501
    raise UnprocessableEntity(
        description=cast("dict", error.messages).get("auto", error.messages),
    )

render_swagger_json

render_swagger_json(api: Api) -> str | bytes

Render a Swagger Json for a given API.

Parameters:

Name	Type	Description	Default
api	Api	api object	required

Returns:

Name	Type	Description
str	str | bytes	render result

Source code in flask_restx_marshmallow/util.py

def render_swagger_json(api: "Api") -> str | bytes:
    """Render a Swagger Json for a given API.

    Args:
        api (Api): api object

    Returns:
        str: render result
    """
    return json.dumps(api.__schema__)

render_swagger_ui

render_swagger_ui(api: Api) -> str

Render a SwaggerUI for a given API.

Parameters:

Name	Type	Description	Default
api	Api	api object	required

Returns:

Name	Type	Description
str	str	render result

Source code in flask_restx_marshmallow/util.py

def render_swagger_ui(api: "Api") -> str:
    """Render a SwaggerUI for a given API.

    Args:
        api (Api): api object

    Returns:
        str: render result
    """
    if base := current_app.config.get("SWAGGER_UI_BASE_URL"):
        return render_template(
            "index.html.j2",
            base_url=base,
            csrf_token_cookie_name=current_app.config.get(
                "CSRF_COOKIE_NAME",
                "csrf_token",
            ),
            specs_url=api.specs_url,
            title=api.title,
        )
    if (
        current_version := current_app.config.get("SWAGGER_UI_VERSION")
    ) is None and current_app.config.get("SWAGGER_UI_USE_CDN") is True:
        try:
            from bs4 import BeautifulSoup  # noqa: PLC0415
            from bs4 import Tag as Tag_  # noqa: PLC0415

            if (
                page := get("https://cdnjs.com/libraries/swagger-ui", timeout=5)
            ).status_code == HTTPStatus.OK.value:
                current_version = json.loads(
                    str(
                        cast(
                            "Tag_",
                            BeautifulSoup(page.content, "html.parser").find(
                                attrs={"type": "application/ld+json"},
                            ),
                        ).contents[0],
                    ),
                )["softwareVersion"]
        except ImportError:
            logger.warning(
                "Extra package 'beautifulsoup4' not found! Use local swagger.",
            )
        except (ConnectionError, ReadTimeout):
            logger.info("Failed to get swagger-ui version. Use local swagger.")
    if (
        current_version is not None
        and current_app.config.get("SWAGGER_UI_USE_CDN") is True
    ):
        if current_app.config.get("SWAGGER_UI_DOWNLOAD_UPDATES") is True:
            download_files(current_version, "swagger-ui-bundle.min.js")
            download_files(current_version, "swagger-ui-standalone-preset.min.js")
            download_files(current_version, "swagger-ui.min.css")
        return render_template(
            "index.html.j2",
            base_url=current_app.config.get(
                "SWAGGER_UI_CDN_URL",
                f"{current_app.config.get('SWAGGER_UI_CDN_BASE')}/{current_version}",
            ),
            csrf_token_cookie_name=current_app.config.get(
                "CSRF_COOKIE_NAME",
                "csrf_token",
            ),
            specs_url=api.specs_url,
            title=api.title,
        )
    return render_template(
        "local.html.j2",
        csrf_token_cookie_name=current_app.config.get("CSRF_COOKIE_NAME", "csrf_token"),
        specs_url=api.specs_url,
        title=api.title,
    )

schema_name_resolver

schema_name_resolver(schema: type[T]) -> str

Resolve schema name.

Parameters:

Name	Type	Description	Default
schema	type[Schema]	schema class	required

Returns:

Name	Type	Description
str	str	schema name

Source code in flask_restx_marshmallow/util.py

def schema_name_resolver[T: Schema_](schema: type[T]) -> str:
    """Resolve schema name.

    Args:
        schema (type[Schema]): schema class

    Returns:
        str: schema name
    """
    return cast("type[T]", resolve_schema_cls(schema)).__name__

security_helper

security_helper(
    security: S | Sequence[S] | A,
) -> Security

Security helper.

Parameters:

Name	Type	Description	Default
security	str | Mapping[str, list[str] | str]	security name or instance.	required

Raises:

Type	Description
TypeError	if security is not str or Mapping.

Returns:

Name	Type	Description
Security	Security	security form value.

Source code in flask_restx_marshmallow/util.py

def security_helper[A, S: str | Mapping[str, list[str] | str]](
    security: S | Sequence[S] | A,  # type: ignore[valid-type]
) -> "Security":
    """Security helper.

    Args:
        security (str | Mapping[str, list[str] | str]): security name or instance.

    Raises:
        TypeError: if security is not str or Mapping.

    Returns:
        Security: security form value.
    """
    if isinstance(security, str):
        return {security: []}
    if isinstance(security, Mapping):
        security_key = next(iter(security))
        if not isinstance(
            security_value := security[security_key],
            str | list,
        ) or not all(
            isinstance(security_value_inner, str)
            for security_value_inner in security_value
        ):
            msg = "security value must be str or list[str]."
            raise TypeError(msg)
        return {
            security_key: security_value
            if isinstance(security_value, list)
            else [security_value],
        }
    msg = f"security must be str or Mapping, got {type(security).__name__}"
    raise TypeError(msg)

set_location

set_location(field: Field, location: Location) -> None

Set location.

Parameters:

Name	Type	Description	Default
field	Field	field	required
location	Location	location	required

Source code in flask_restx_marshmallow/util.py

def set_location(field: "Field", location: Location) -> None:
    """Set location.

    Args:
        field (Field): field
        location (Location): location
    """
    if not isinstance(field, List | Tuple):
        cast("dict", field.metadata).setdefault("location", location)
    elif isinstance(field, List):
        set_location(cast("Field", field.inner), location)
    else:
        for tuple_field in cast("list[Field]", field.tuple_fields):
            set_location(cast("Field", tuple_field), location)

tag_helper

tag_helper(tag: Sequence[str] | str | dict | None) -> Tag

Tag helper.

Parameters:

Name	Type	Description	Default
tag	Sequence[str] | str | dict | None	tag	required

Raises:

Type	Description
ValueError	Unsupported tag format.

Returns:

Name	Type	Description
Tag	Tag	tag

Source code in flask_restx_marshmallow/util.py

def tag_helper(tag: Sequence[str] | str | dict | None) -> Tag:
    """Tag helper.

    Args:
        tag (Sequence[str] | str | dict | None): tag

    Raises:
        ValueError: Unsupported tag format.

    Returns:
        Tag: tag
    """
    if isinstance(tag, str):
        return Tag(name=tag)
    if isinstance(tag, dict) and "name" in tag:
        return Tag(**tag)
    if (
        isinstance(tag, Sequence)
        and isinstance(name_ := next(iter(tag)), str)
        and isinstance(description_ := next(iter(tag)), str)
    ):
        return Tag(name=name_, description=description_)
    msg = f"Unsupported tag format for {tag}"
    raise ValueError(msg)

unpack

unpack(
    response: TupleResponse
    | T
    | tuple[T]
    | tuple[T, int | HTTPStatus]
    | tuple[T, int | HTTPStatus, dict],
    default_code: HTTPStatus | int = OK,
) -> TupleResponse

Unpack a Flask standard response.

Parameters:

Name	Type	Description	Default
response	T | tuple[T] | tuple[T, int | HTTPStatus] | tuple[T, int | HTTPStatus, dict]	response	required
default_code	HTTPStatus | int	default code. Defaults to HTTPStatus.OK.	OK

Raises:

Type	Description
ValueError	if the response does not have one of the expected format

Returns:

Name	Type	Description
TupleResponse	TupleResponse	a 3-tuple ``(data, code, headers)

Source code in flask_restx_marshmallow/util.py

def unpack[T](
    response: TupleResponse
    | T
    | tuple[T]
    | tuple[T, int | HTTPStatus]
    | tuple[T, int | HTTPStatus, dict],
    default_code: HTTPStatus | int = HTTPStatus.OK,
) -> TupleResponse:
    """Unpack a Flask standard response.

    Args:
        response (T | tuple[T] | tuple[T, int | HTTPStatus] | tuple[T, int | HTTPStatus, dict]): response
        default_code (HTTPStatus | int, optional): default code. Defaults to HTTPStatus.OK.

    Raises:
        ValueError: if the response does not have one of the expected format

    Returns:
        TupleResponse: a 3-tuple ``(data, code, headers)
    """  # noqa: E501
    if isinstance(response, TupleResponse):
        return response
    if not isinstance(response, tuple):
        return TupleResponse(Success(result=response), default_code, {})
    if len(response) == 1:
        return TupleResponse(Success(result=response[0]), default_code, {})
    if len(response) == 2:  # noqa: PLR2004
        data, code = response
        return TupleResponse(Success(result=data), code, {})
    if len(response) == 3:  # noqa: PLR2004
        data, code, headers = response
        return TupleResponse(Success(result=data), code, headers)
    msg = "Too many response values"
    raise ValueError(msg)