fastack.decorators

enable_context(initializer=None, finalizer=None)

A decorator that activates the application context The function can be a coroutine or a normal function.

Parameters:

Name	Type	Description	Default
initializer	Callable[[Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[NoneType]]], Awaitable[NoneType]]], Any]	The function to be called before application context starts.	None
finalizer	Callable[[Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[NoneType]]], Awaitable[NoneType]], Any], Any]	The function that will be called after your function is called.	None

Notes

• The initializer will accept one argument. initializer(app) where app is the application.
• The finalizer will accept two argument. finalizer(app, rv) where app is the application and rv is the return value of the function.

Source code in fastack/decorators.py

def enable_context(
    initializer: Callable[[ASGIApp], Any] = None,
    finalizer: Callable[[ASGIApp, Any], Any] = None,
) -> Callable:
    """
    A decorator that activates the application context
    The function can be a coroutine or a normal function.

    Args:
        initializer: The function to be called before application context starts.
        finalizer: The function that will be called after your function is called.

    notes:
        - The initializer will accept one argument. ``initializer(app)`` where ``app`` is the application.
        - The finalizer will accept two argument. ``finalizer(app, rv)`` where ``app`` is the application and ``rv`` is the return value of the function.

    """

    def wrapper(func):
        @wraps(func)
        def decorator(*args, **kwargs):
            ctx = None
            for _, v in kwargs.items():
                # Finding Context in parameters
                if isinstance(v, click.core.Context):  # pragma: no cover
                    ctx = v
                    break

            app: ASGIApp
            if ctx is None:
                app = load_app()
            else:
                # This is always a Fastack object, below is just for checking.
                app = ctx.obj  # pragma: no cover
                if not isinstance(app, Fastack):  # pragma: no cover
                    app = load_app()
                    ctx.obj = app

            assert isinstance(app, Fastack), "Invalid application type"

            async def executor():
                if callable(initializer):
                    initializer(app)

                async with app.app_context():
                    if asyncio.iscoroutinefunction(func):
                        rv = await func(*args, **kwargs)
                    else:
                        rv = func(*args, **kwargs)  # pragma: no cover

                if callable(finalizer):
                    finalizer(app, rv)

                return rv

            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()

            task = loop.create_task(executor())
            return loop.run_until_complete(task)

        return decorator

    return wrapper

route(path=None, *, action=False, response_model=None, status_code=None, tags=None, dependencies=None, summary=None, description=None, response_description='Successful Response', responses=None, deprecated=None, methods=None, operation_id=None, response_model_include=None, response_model_exclude=None, response_model_by_alias=True, response_model_exclude_unset=False, response_model_exclude_defaults=False, response_model_exclude_none=False, include_in_schema=True, response_class=<fastapi.datastructures.DefaultPlaceholder object at 0x7f0df62b25d0>, name=None, route_class_override=None, callbacks=None, openapi_extra=None)

A decorator to add additional information for endpoints in OpenAPI.

:param path: The path of the endpoint. :param action: To mark this method is the responder to be included in the controller.

Source code in fastack/decorators.py

def route(
    path: Optional[str] = None,
    *,
    action: bool = False,
    response_model: Optional[Type[Any]] = None,
    status_code: Optional[int] = None,
    tags: Optional[List[str]] = None,
    dependencies: Optional[Sequence[params.Depends]] = None,
    summary: Optional[str] = None,
    description: Optional[str] = None,
    response_description: str = "Successful Response",
    responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
    deprecated: Optional[bool] = None,
    methods: Optional[Union[Set[str], List[str]]] = None,
    operation_id: Optional[str] = None,
    response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: Union[Type[Response], DefaultPlaceholder] = Default(JSONResponse),
    name: Optional[str] = None,
    route_class_override: Optional[Type[APIRouter]] = None,
    callbacks: Optional[List[BaseRoute]] = None,
    openapi_extra: Optional[Dict[str, Any]] = None,
):
    """
    A decorator to add additional information for endpoints in OpenAPI.

    :param path: The path of the endpoint.
    :param action: To mark this method is the responder to be included in the controller.
    """

    def wrapper(func):
        if asyncio.iscoroutinefunction(func):

            @wraps(func)
            async def decorated(*args, **kwds):
                return await func(*args, **kwds)

        else:

            @wraps(func)
            def decorated(*args, **kwds):
                return func(*args, **kwds)

        params = getattr(decorated, "__route_params__", {})
        params.update(
            {
                "path": path,
                "response_model": response_model,
                "status_code": status_code,
                "tags": tags,
                "dependencies": dependencies,
                "summary": summary,
                "description": description,
                "response_description": response_description,
                "responses": responses,
                "deprecated": deprecated,
                "methods": methods,
                "operation_id": operation_id,
                "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,
                "include_in_schema": include_in_schema,
                "response_class": response_class,
                "name": name,
                "route_class_override": route_class_override,
                "callbacks": callbacks,
                "openapi_extra": openapi_extra,
            }
        )
        decorated.__route_params__ = params
        decorated.__route_action__ = action
        return decorated

    return wrapper

---

Last update: January 17, 2022 13:26:42