Skip to content

ABCApp

faststream.app.ABCApp #

ABCApp(
    broker: Optional[BrokerAsyncUsecase[Any, Any]] = None,
    logger: Optional[logging.Logger] = logger,
    title: str = "FastStream",
    version: str = "0.1.0",
    description: str = "",
    terms_of_service: Optional[AnyHttpUrl] = None,
    license: Optional[
        Union[License, LicenseDict, AnyDict]
    ] = None,
    contact: Optional[
        Union[Contact, ContactDict, AnyDict]
    ] = None,
    identifier: Optional[str] = None,
    tags: Optional[
        Sequence[Union[Tag, TagDict, AnyDict]]
    ] = None,
    external_docs: Optional[
        Union[ExternalDocs, ExternalDocsDict, AnyDict]
    ] = None,
)

Bases: ABC

A class representing an ABC App.

METHOD DESCRIPTION
set_broker

Set the broker object

on_startup

Add a hook to be run before the broker is connected

on_shutdown

Add a hook to be run before the broker is disconnected

after_startup

Add a hook to be run after the broker is connected

after_shutdown

Add a hook to be run after the broker is disconnected

_log

Log a message at a specified

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Initialize an instance of the class.

PARAMETER DESCRIPTION
broker

An optional instance of the BrokerAsyncUsecase class.

TYPE: Optional[BrokerAsyncUsecase[Any, Any]] DEFAULT: None

logger

An optional instance of the logging.Logger class.

TYPE: Optional[Logger] DEFAULT: logger

title

A string representing the title of the AsyncAPI.

TYPE: str DEFAULT: 'FastStream'

version

A string representing the version of the AsyncAPI.

TYPE: str DEFAULT: '0.1.0'

description

A string representing the description of the AsyncAPI.

TYPE: str DEFAULT: ''

terms_of_service

An optional URL representing the terms of service of the AsyncAPI.

TYPE: Optional[AnyHttpUrl] DEFAULT: None

license

An optional instance of the License class.

TYPE: Optional[Union[License, LicenseDict, AnyDict]] DEFAULT: None

contact

An optional instance of the Contact class.

TYPE: Optional[Union[Contact, ContactDict, AnyDict]] DEFAULT: None

identifier

An optional string representing the identifier of the AsyncAPI.

TYPE: Optional[str] DEFAULT: None

tags

An optional sequence of Tag instances.

TYPE: Optional[Sequence[Union[Tag, TagDict, AnyDict]]] DEFAULT: None

external_docs

An optional instance of the ExternalDocs class.

TYPE: Optional[Union[ExternalDocs, ExternalDocsDict, AnyDict]] DEFAULT: None

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/app.py
def __init__(
    self,
    broker: Optional[BrokerAsyncUsecase[Any, Any]] = None,
    logger: Optional[logging.Logger] = logger,
    # AsyncAPI information
    title: str = "FastStream",
    version: str = "0.1.0",
    description: str = "",
    terms_of_service: Optional[AnyHttpUrl] = None,
    license: Optional[Union[License, LicenseDict, AnyDict]] = None,
    contact: Optional[Union[Contact, ContactDict, AnyDict]] = None,
    identifier: Optional[str] = None,
    tags: Optional[Sequence[Union[Tag, TagDict, AnyDict]]] = None,
    external_docs: Optional[Union[ExternalDocs, ExternalDocsDict, AnyDict]] = None,
):
    """Initialize an instance of the class.

    Args:
        broker: An optional instance of the BrokerAsyncUsecase class.
        logger: An optional instance of the logging.Logger class.
        title: A string representing the title of the AsyncAPI.
        version: A string representing the version of the AsyncAPI.
        description: A string representing the description of the AsyncAPI.
        terms_of_service: An optional URL representing the terms of service of the AsyncAPI.
        license: An optional instance of the License class.
        contact: An optional instance of the Contact class.
        identifier: An optional string representing the identifier of the AsyncAPI.
        tags: An optional sequence of Tag instances.
        external_docs: An optional instance of the ExternalDocs class.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    self.broker = broker
    self.logger = logger
    self.context = context
    context.set_global("app", self)

    self._on_startup_calling = []
    self._after_startup_calling = []
    self._on_shutdown_calling = []
    self._after_shutdown_calling = []

    # AsyncAPI information
    self.title = title
    self.version = version
    self.description = description
    self.terms_of_service = terms_of_service
    self.license = license
    self.contact = contact
    self.identifier = identifier
    self.asyncapi_tags = tags
    self.external_docs = external_docs

asyncapi_tags instance-attribute #

asyncapi_tags = tags

broker instance-attribute #

broker = broker

contact instance-attribute #

contact = contact

context instance-attribute #

context = context

description instance-attribute #

description = description

external_docs instance-attribute #

external_docs = external_docs

identifier instance-attribute #

identifier = identifier

license instance-attribute #

license = license

logger instance-attribute #

logger = logger

terms_of_service instance-attribute #

terms_of_service = terms_of_service

title instance-attribute #

title = title

version instance-attribute #

version = version

after_shutdown #

after_shutdown(
    func: Callable[P_HookParams, T_HookReturn]
) -> Callable[P_HookParams, T_HookReturn]

Add hook running AFTER broker disconnected

Source code in faststream/app.py
def after_shutdown(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running AFTER broker disconnected"""
    self._after_shutdown_calling.append(apply_types(func))
    return func

after_startup #

after_startup(
    func: Callable[P_HookParams, T_HookReturn]
) -> Callable[P_HookParams, T_HookReturn]

Add hook running AFTER broker connected

Source code in faststream/app.py
def after_startup(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running AFTER broker connected"""
    self._after_startup_calling.append(apply_types(func))
    return func

on_shutdown #

on_shutdown(
    func: Callable[P_HookParams, T_HookReturn]
) -> Callable[P_HookParams, T_HookReturn]

Add hook running BEFORE broker disconnected

Source code in faststream/app.py
def on_shutdown(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running BEFORE broker disconnected"""
    self._on_shutdown_calling.append(apply_types(func))
    return func

on_startup #

on_startup(
    func: Callable[P_HookParams, T_HookReturn]
) -> Callable[P_HookParams, T_HookReturn]

Add hook running BEFORE broker connected This hook also takes an extra CLI options as a kwargs

Source code in faststream/app.py
def on_startup(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running BEFORE broker connected
    This hook also takes an extra CLI options as a kwargs"""
    self._on_startup_calling.append(apply_types(func))
    return func

set_broker #

set_broker(broker: BrokerAsyncUsecase[Any, Any]) -> None

Set already existed App object broker Usefull then you create/init broker in on_startup hook

Source code in faststream/app.py
def set_broker(self, broker: BrokerAsyncUsecase[Any, Any]) -> None:
    """Set already existed App object broker
    Usefull then you create/init broker in `on_startup` hook"""
    self.broker = broker

Last update: 2023-11-13