Skip to content

ABCApp

faststream.app.ABCApp #

ABCApp(broker: Optional[BrokerAsyncUsecase[Any, Any]] = None, logger: Optional[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

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

Source code in faststream/app.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
106
107
108
109
110
111
112
113
114
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,
) -> 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.
    """
    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 
150
151
152
153
154
155
156
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 
142
143
144
145
146
147
148
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 
134
135
136
137
138
139
140
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 
123
124
125
126
127
128
129
130
131
132
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.

Useful then you create/init broker in on_startup hook.

Source code in faststream/app.py 
116
117
118
119
120
121
def set_broker(self, broker: BrokerAsyncUsecase[Any, Any]) -> None:
    """Set already existed App object broker.

    Useful then you create/init broker in `on_startup` hook.
    """
    self.broker = broker