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 | |
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
|
broker instance-attribute
context instance-attribute
description instance-attribute
description = description
external_docs instance-attribute
external_docs = external_docs
identifier instance-attribute
license instance-attribute
logger instance-attribute
terms_of_service instance-attribute
terms_of_service = terms_of_service
version instance-attribute
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
|