warren.pubsub.rabbitmq
RabbitMQ pubsub layer.
The top-level module exposes only the protocol-level configs — pure-data Pydantic models that describe the RabbitMQ topology and retry policy. They have no dependency on any specific Python client library and can be loaded, serialized, and reused freely.
Concrete client implementations live in sub-packages:
aio_pika—aio_pika-based managers and helpers.
To use the aio_pika implementation, import explicitly from that
sub-package, e.g.:
from warren.pubsub.rabbitmq.aio_pika import (
RMQConnectionManager,
RMQConsumerManager,
RMQPublisher,
)Modules
Classes
RMQConnectionConfig
modelRMQConnectionConfigConnection-specific parameters for aio_pika.connect_robust().
attributehoststr= 'localhost'attributeportint= 5672attributeloginstr= 'guest'attributepasswordSecretStr= SecretStr('guest')attributevirtualhoststr= '/'attributesslbool= Falseattributessl_optionsdict[str, Any] | None= Noneattributessl_contextAny | None= Noneattributetimeoutfloat | None= Noneattributeclient_propertiesdict[str, Any] | None= Noneattributeheartbeatint | None= NoneAMQP heartbeat timeout in seconds, sent as the client's Tune-Ok value — which RabbitMQ adopts as-is, so this setting alone decides the negotiated timeout. None keeps aiormq's default of 60; 0 disables heartbeats. The bounds are aiormq's own: anything outside 0 <= v < 65535 would be read as 0 (heartbeats off), which is never what a typo means. A long timeout also delays aiormq's detection of a silent broker to (heartbeat + 1) * 3 seconds.
RMQConsumerConfig
modelRMQConsumerConfigConsumer-side settings, one set per worker process.
attributeconcurrencyint | None= NoneMaximum number of handlers that run at the same time. None means prefetch_count handlers. The channel prefetch is raised to concurrency when concurrency is larger.
attributeprefetch_countint= 1Unacked deliveries the broker pushes at once.
attributehandler_timeout_secondsfloat | None= NoneTime limit for one handler call. A handler that times out is a soft failure. /live fails when a handler runs for twice this value. None means no limit.
attributeon_shutdown_timeoutfloat= 30.0Seconds to wait for in-flight work on stop.
attributemax_deliveriesint | None= NoneDead-letter a message after this many broker deliveries (the first counts as 1). Every requeue costs one — a channel loss mid-flight included — so this is a count of chances. None keeps redelivery unbounded.
attributeredelivery_delayint= 5Seconds before a counted redelivery is replayed through the retry worker. At least 1: a retry_after of 0 reads as "unset" to the retry policy and would fall back to its default.
attributequeue_argumentsdict[str, Any] | None= NoneForwarded into RMQQueueConfig.arguments for the worker's queue; None declares the queue as today.
RMQConsumerManagerConfig
modelRMQConsumerManagerConfigattributeexchangeRMQExchangeConfigattributequeueRMQQueueConfigattributeconsumerRMQConsumerConfigRMQExchangeConfig
modelRMQExchangeConfigattributenamestrattributetypeLiteral['fanout', 'direct', 'topic']= 'topic'attributedurablebool= TrueRMQQueueConfig
modelRMQQueueConfigQueue declaration parameters.
attributenamestrattributedurablebool= Trueattributeexclusivebool= Falseattributeauto_deletebool= Falseattributerouting_keystr | None= Noneattributebinding_keystuple[str, ...]= ()attributeargumentsdict[str, Any] | None= NoneExtra x- arguments forwarded verbatim to the declaration (queue type, delivery limit, dead-letter exchange, …). Not validated here; the broker refuses to re-declare an existing queue with different arguments.
RetryConfig
modelRetryConfigRetry policy configuration for the consumer manager.
Transport-agnostic: provides defaults when the worker's
SoftFailureException does not specify values, and caps to enforce
system-level limits. Used by all pubsub backends (RabbitMQ, Kafka,
memory) and by RuntimeRetryConfig.policy.
attributeinitial_delayint= 30Initial delay in seconds before first retry when worker does not specify.
attributemax_retriesint= 5Max retry attempts when worker does not specify.
attributebackoff_basefloat= 2.0Base for exponential backoff. Delay on attempt N = initial_delay * backoff_base^(N-1).
attributejitterbool= TrueWhether to add random jitter to delays.
attributemax_delay_capint= 300Maximum delay in seconds (caps backoff).
attributemax_retries_capint= 10Maximum retries allowed (overrides worker request if exceeded).
attributefallback_requeue_delayfloat= 2.0Delay in seconds before nack+requeue when no retry publisher is configured. Prevents tight retry loops.