Skip to content

wrap_callable_to_fastapi_compatible

faststream.broker.fastapi.route.wrap_callable_to_fastapi_compatible #

wrap_callable_to_fastapi_compatible(user_callable, *, provider_factory, dependencies, response_model, response_model_include, response_model_exclude, response_model_by_alias, response_model_exclude_unset, response_model_exclude_defaults, response_model_exclude_none)
Source code in faststream/broker/fastapi/route.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def wrap_callable_to_fastapi_compatible(
    user_callable: Callable[P_HandlerParams, T_HandlerReturn],
    *,
    provider_factory: Callable[[], Any],
    dependencies: Iterable["params.Depends"],
    response_model: Any,
    response_model_include: Optional["IncEx"],
    response_model_exclude: Optional["IncEx"],
    response_model_by_alias: bool,
    response_model_exclude_unset: bool,
    response_model_exclude_defaults: bool,
    response_model_exclude_none: bool,
) -> Callable[["NativeMessage[Any]"], Awaitable[Any]]:
    __magic_attr = "__faststream_consumer__"

    if getattr(user_callable, __magic_attr, False):
        return user_callable  # type: ignore[return-value]

    if response_model:
        response_field = create_response_field(
            name="ResponseModel",
            type_=response_model,
            mode="serialization",
        )
    else:
        response_field = None

    parsed_callable = build_faststream_to_fastapi_parser(
        dependent=get_fastapi_native_dependant(user_callable, list(dependencies)),
        provider_factory=provider_factory,
        response_field=response_field,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
    )

    setattr(parsed_callable, __magic_attr, True)
    return wraps(user_callable)(parsed_callable)