warren.pubsub.rabbitmq.aio_pika
aio_pika-based implementation of the RabbitMQ pubsub layer.
The protocol-level configs in config describe what
to set up; the modules here use aio_pika to actually do it.
aio_pika is an optional dependency: importing this sub-package
without it installed raises
OptionalDependencyError pointing at the
warren[rmq] extra.
Classes
RMQConnectionManager
classRMQConnectionManager(config, *, name=None)paramconfigRMQConnectionConfigparamnamestr | None= NoneattributeconnectionAbstractRobustConnection | NoneThe robust connection, or None before setup().
methodsetupasync () -> Nonemethodteardownasync () -> Nonemethodcreate_channelasync () -> AbstractChannelRMQConsumerManager
classRMQConsumerManager(config, connection_manager, consumer, *, data_publisher=None, control_publisher=None, observer_publisher=None, retry_config=None, extract_identity_func=None, publish_hard_failures=True)RabbitMQ consumer manager that bridges message queue operations with application workers.
This class handles all RabbitMQ-specific concerns (connection management, message acknowledgment, retry logic) while delegating business logic to a worker function that remains agnostic of the underlying message transport.
The consumer receives plain dictionaries and returns optional response dictionaries—it never interacts with RabbitMQ directly.
paramconfigRMQConsumerManagerConfigparamconnection_managerRMQConnectionManagerparamconsumerMessageConsumerInterfaceparamdata_publisherPublisherInterface | None= Noneparamcontrol_publisherPublisherInterface | None= Noneparamobserver_publisherPublisherInterface | None= Noneparamretry_configRetryConfig | None= Noneparamextract_identity_funcExtractMessageIdentityFunc | None= Noneparampublish_hard_failuresbool= Truemethodsetupasync () -> NoneInitialize RabbitMQ resources for consumption.
- Sets up all publishers.
- Creates a channel from the shared connection manager.
- Declares the necessary exchange and queue, and binds the queue to the exchange.
- Must be called before start_consuming().
methodstart_consumingasync () -> NoneBegin consuming messages from the configured queue.
Each delivery spawns a task; handler concurrency is bounded separately from the channel prefetch.
methodstop_consumingasync () -> NoneGracefully stop message consumption and close RabbitMQ resources.
- Signals shutdown (stops accepting new work)
- Cancels the consumer (no new messages delivered)
- Waits for in-flight messages to complete (with timeout)
- Closes channel and tears down publishers
methodhealthasync (*, probe_timeout=1.0) -> ConsumerHealthObserve transport state without touching the broker.
Reads public aio-pika surface, plus consumers on the underlying
aiormq channel — a plain attribute that is not on aiormq's ABC, so
its absence reads as unknown, never as lost.
paramprobe_timeoutfloat= 1.0RMQPublisher
classRMQPublisher(connection_manager, exchange_config, *, route=None, route_func=None, delivery_mode=2, content_type='application/json', name=None)RabbitMQ publisher that owns its own routing decisions.
Publishes to exchange_config. A static route gives one routing
key for every message. A route_func computes the routes for each
message.
- Fanout: pass neither
routenorroute_func, or pass aroute_func. A staticrouteraises ValueError. - Topic and direct: pass exactly one of
routeandroute_func. Neither raises ValueError. - Both
routeandroute_funcalways raise ValueError.
paramconnection_managerRMQConnectionManagerparamexchange_configRMQExchangeConfigparamrouteRoute | None= Noneparamroute_funcRouteFunc | None= Noneparamdelivery_modeint= 2paramcontent_typestr= 'application/json'paramnamestr | None= NoneRaises:
ValueError: For a combination ofroute,route_funcand exchange type that the rules above do not allow.
methodsetupasync () -> NoneOpen a channel and declare the publisher's exchange.
On partial failure the caller is responsible for cleanup via
teardown() (idempotent, best-effort).
Raises:
PubSubSetupError: if the channel or exchange cannot be set up.
methodteardownasync () -> Nonemethod__call__async (message) -> Noneparammessagedict[str, Any]Functions
declare_exchange
functiondeclare_exchangeasync (channel, config) -> AbstractExchangeIdempotently declare the exchange described by config on channel.
paramchannelAbstractChannelparamconfigRMQExchangeConfigRaises:
PubSubSetupError: if the broker rejects the declaration.
declare_queue
functiondeclare_queueasync (channel, exchange, config, *, exchange_type) -> AbstractQueueIdempotently declare the queue described by config and bind it to exchange.
paramchannelAbstractChannelparamexchangeAbstractExchangeparamconfigRMQQueueConfigparamexchange_typestrRaises:
PubSubSetupError: if the broker rejects the declaration or bind.
purge_queues
functionpurge_queuesasync (*, connection_manager, queue_names, exchange_name='') -> NoneDelete the listed queues and optionally the exchange.
Best-effort: each deletion runs on its own channel, so failing to open a channel or to delete one entity is logged and the remaining entities are still attempted.
paramconnection_managerRMQConnectionManageran already-setup RMQConnectionManager.
paramqueue_namesSequence[str]queues to delete.
paramexchange_namestr= ''exchange to delete. Skipped when empty.