aio_microservice.amqp¶

Submodules¶

Classes¶

`AmqpExtension`	Helper class that provides a standard way to create an ABC using
`AmqpExtensionSettings`	Usage docs: https://docs.pydantic.dev/2.7/concepts/models/
`AmqpSettings`	Usage docs: https://docs.pydantic.dev/2.7/concepts/models/
`publisher`
`subscriber`

Package Contents¶

class aio_microservice.amqp.AmqpExtension(settings)[source]¶

Bases: aio_microservice.core.abc.ExtensionABC

Helper class that provides a standard way to create an ABC using inheritance.

Parameters:: settings (AmqpExtensionSettings)

__amqp_middlewares__: ClassVar[list[type[faststream.BaseMiddleware]]] = []¶

amqp_litestar_on_app_init(app_config)¶

Parameters:: app_config (litestar.config.app.AppConfig)
Return type:: litestar.config.app.AppConfig

class aio_microservice.amqp.AmqpExtensionSettings(/, **data)[source]¶

Bases: pydantic.BaseModel

Usage docs: https://docs.pydantic.dev/2.7/concepts/models/

A base class for creating Pydantic models.

Parameters:: data (Any)

__class_vars__¶: The names of classvars defined on the model.

__private_attributes__¶: Metadata about the private attributes of the model.

__signature__¶: The signature for instantiating the model.

__pydantic_complete__¶: Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__¶: The pydantic-core schema used to build the SchemaValidator and SchemaSerializer.

__pydantic_custom_init__¶: Whether the model has a custom __init__ function.

__pydantic_decorators__¶: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__¶: Metadata for generic models; contains data used for a similar purpose to __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__¶: Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__¶: The name of the post-init method for the model, if defined.

__pydantic_root_model__¶: Whether the model is a RootModel.

__pydantic_serializer__¶: The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__¶: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_extra__¶: An instance attribute with the values of extra fields from validation when model_config[‘extra’] == ‘allow’.

__pydantic_fields_set__¶: An instance attribute with the names of fields explicitly set.

__pydantic_private__¶: Instance attribute with the values of private attributes set on the model instance.

amqp: AmqpSettings¶

class aio_microservice.amqp.AmqpSettings(/, **data)[source]¶

Bases: pydantic.BaseModel

Usage docs: https://docs.pydantic.dev/2.7/concepts/models/

A base class for creating Pydantic models.

Parameters:: data (Any)

__class_vars__¶: The names of classvars defined on the model.

__private_attributes__¶: Metadata about the private attributes of the model.

__signature__¶: The signature for instantiating the model.

__pydantic_complete__¶: Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__¶: The pydantic-core schema used to build the SchemaValidator and SchemaSerializer.

__pydantic_custom_init__¶: Whether the model has a custom __init__ function.

__pydantic_decorators__¶: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__¶: Metadata for generic models; contains data used for a similar purpose to __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__¶: Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__¶: The name of the post-init method for the model, if defined.

__pydantic_root_model__¶: Whether the model is a RootModel.

__pydantic_serializer__¶: The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__¶: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_extra__¶: An instance attribute with the values of extra fields from validation when model_config[‘extra’] == ‘allow’.

__pydantic_fields_set__¶: An instance attribute with the names of fields explicitly set.

__pydantic_private__¶: Instance attribute with the values of private attributes set on the model instance.

host: str¶

port: aio_microservice.types.Port¶

username: str¶

password: pydantic.SecretStr¶

prefetch_count: int | None¶

timeout_graceful_shutdown: float | None¶

class aio_microservice.amqp.publisher[source]¶

Bases: AmqpDecorator

queue: faststream.rabbit.RabbitQueue | str = ''¶

exchange: faststream.rabbit.RabbitExchange | str | None = None¶

routing_key: str = ''¶

mandatory: bool = True¶

immediate: bool = False¶

timeout: aio_pika.abc.TimeoutType = None¶

persist: bool = False¶

reply_to: str | None = None¶

priority: int | None = None¶

middlewares: collections.abc.Iterable[faststream.broker.types.PublisherMiddleware] = ()¶

title: str | None = None¶

description: str | None = None¶

include_in_schema: bool = True¶

__call__(fn)[source]¶

Parameters:: fn (Callable[typing_extensions.Concatenate[AmqpExtensionT, P], R])
Return type:: Callable[typing_extensions.Concatenate[AmqpExtensionT, P], R]

class aio_microservice.amqp.subscriber[source]¶

Bases: AmqpDecorator

queue: faststream.rabbit.RabbitQueue | str¶

exchange: faststream.rabbit.RabbitExchange | str | None = None¶

consume_args: faststream.types.AnyDict | None = None¶

reply_config: faststream.rabbit.schemas.ReplyConfig | None = None¶

middlewares: collections.abc.Iterable[faststream.broker.types.SubscriberMiddleware[faststream.rabbit.message.RabbitMessage]] = ()¶

filter: faststream.broker.types.Filter[faststream.rabbit.message.RabbitMessage]¶

retry: bool | int = False¶

no_ack: bool = False¶

no_reply: bool = False¶

title: str | None = None¶

description: str | None = None¶

include_in_schema: bool = True¶

__call__(fn)[source]¶

Parameters:: fn (Callable[typing_extensions.Concatenate[AmqpExtensionT, P], R])
Return type:: Callable[typing_extensions.Concatenate[AmqpExtensionT, P], R]