warren.pubsub
Pubsub: the publisher and consumer manager contracts, and message routing.
The transports are in the subpackages: warren.pubsub.rabbitmq,
warren.pubsub.kafka and warren.pubsub.memory.
Modules
Classes
BasePublisher
classBasePublisher(route=None, route_func=None, *, name=None)Base class for publishers. Each publisher owns its own routing logic.
paramrouteRoute | None= Noneparamroute_funcRouteFunc | None= Noneparamnamestr | None= NoneRaises:
ValueError: If bothrouteandroute_funcare set.
methodsetupasync () -> NoneSet up the publisher (channels, connections, etc.).
methodteardownasync () -> NoneTear down the publisher.
method__call__async (message) -> NonePublish a message. Routing is determined internally by the publisher.
parammessagedictMessage body as a dictionary.
ConsumerHealth
dataclassConsumerHealth(connected, blocked, channel_open, consumer_registered, detail='', handler_timeout_seconds=None, handler_started_at=())One observation of a consumer manager's transport state.
Built for two readers: a watchdog deciding whether the process should
exit, and a readiness endpoint. consumer_lost is the only condition
a restart fixes; a blocked or reconnecting connection is left to the
transport's own recovery.
paramconnectedboolThe transport is up (not closed, not reconnecting).
paramblockedboolConnected, but the broker holds this connection in Connection.Blocked; publishes and acks stall until it lifts.
paramchannel_openboolThe consume channel is initialised and open.
paramconsumer_registeredbool | NoneOur consumer tag is registered on the underlying channel; None when that could not be determined within the probe timeout (never treated as lost).
paramdetailstr= ''Why the state is what it is, for logs and the endpoint.
paramhandler_timeout_secondsfloat | None= Noneparamhandler_started_attuple[float, ...]= ()attributeliveboolattributein_flight_handlersintattributereadyboolattributeconsumer_lostboolConnection alive and unblocked, but no live consumer on it.
attributestateLiteral['ready', 'reconnecting', 'blocked', 'consumer_lost', 'unknown']ConsumerManagerBase
classConsumerManagerBase(consumer, *, publishers=None, handler_timeout_seconds=None, concurrency=1)paramconsumerMessageConsumerInterfaceparampublisherslist[PublisherInterface] | None= Noneparamhandler_timeout_secondsfloat | None= Noneparamconcurrencyint= 1methodsetupasync () -> NoneSet up consumption for the given consumer and topic.
methodstart_consumingasync () -> NoneAllow the consumer to start consuming.
methodstop_consumingasync () -> NoneStop the consumption of messages by consumer.
methodhealthasync (*, probe_timeout=1.0) -> ConsumerHealthObserve transport state; must not raise for transport reasons.
paramprobe_timeoutfloat= 1.0Upper bound, in seconds, on any wait the observation needs (a blocked connection never becomes ready).
MessageFieldRouter
classMessageFieldRouter(field=DATA_TYPE_FIELD, prefix_field=None, default_prefix=None)A RouteFunc that routes by a single field of the message body.
The value of message[field] becomes the AMQP routing key, so the broker
delivers the message only to queues whose binding matches that value.
paramfieldstr= DATA_TYPE_FIELDMessage-body key to read the routing key from. Defaults to "data_type".
paramprefix_fieldstr | None= NoneOptional field prepended to the key with a dot.
paramdefault_prefixstr | None= NoneFallback when the prefix field is missing or empty.
Raises:
ValueError: On a call, if the message has no value forfield, or no value forprefix_fieldand nodefault_prefix.
method__call__async (message) -> list[Route]parammessagedictReplayRouter
classReplayRouter(field=REPLAY_ROUTING_KEY_FIELD)A RouteFunc that replays the routing key stamped on a message.
Used by the retry worker: the consumer manager stamps the original routing
key (REPLAY_ROUTING_KEY_FIELD) onto a failed message before it goes to
retry, so re-delivery lands back in the same worker's queue regardless of
the routing scheme (data_type, addressed, …). A missing/empty key replays
with "" (correct for fanout, which ignores keys).
paramfieldstr= REPLAY_ROUTING_KEY_FIELDMessage field holding the original routing key.
method__call__async (message) -> list[Route]parammessagedictRoute
dataclassRoute(key)A transport-agnostic route to a destination, specified by the key. Can be extended to include other routing information if necessary.
paramkeystrThe routing key.
RoutingPlan
modelRoutingPlanA job-defined routing tree over worker-type (capability) ids.
Carried in job_parameters[ROUTING_PLAN_KEY] and propagated through the
message chain, so each job can take its own path through the same deployed
workers. A tree (each node has one parent): fan-out is supported (a node
with multiple successors), fan-in/join is not.
This is the addressed routing strategy (route by worker-type id), whose
natural home is a direct exchange.
Coupled to RoutingPlanRouter: the plan is inert data on its own.
It only takes effect if the pipeline's publishers use RoutingPlanRouter
as their route_func. A plan with no such router anywhere is silently
ignored.
attributeentrylist[str]Worker-type id(s) the document enters at.
attributeedgesdict[str, list[str]]producer-id -> successor-id(s); [] marks a terminal node (routes nowhere).
RoutingPlanRouter
classRoutingPlanRouter(plan_key=ROUTING_PLAN_KEY)A RouteFunc for job-defined addressed routing (direct exchange).
Reads the RoutingPlan from message['job_parameters'][plan_key],
looks up the producing node via message['origin']['type'] (the
worker-type id), and emits one Route per successor. When the origin
is not a node in the plan — e.g. the initial publish from a publisher — it
routes to the plan's entry nodes.
This router and RoutingPlan are two halves of one mechanism: wire
this as a publisher's route_func for the plan to take effect. Without a
plan in the message it raises ValueError; with a plan but no router the plan
is silently ignored. The coupling cannot be checked at deploy time because a
route_func is opaque to the framework.
paramplan_keystr= ROUTING_PLAN_KEYKey under job_parameters holding the plan.
Raises:
ValueError: On a call, if the message has no plan atjob_parameters[plan_key].
method__call__async (message) -> list[Route]parammessagedictProtocols
ConsumerManagerInterface
protocolConsumerManagerInterfacemethodsetupasync () -> NoneSet up consumption.
methodstart_consumingasync () -> NoneStart consuming.
methodstop_consumingasync () -> NoneStop the consumption.
methodhealthasync (*, probe_timeout=1.0) -> ConsumerHealthObserve transport state; must not raise for transport reasons.
paramprobe_timeoutfloat= 1.0Upper bound, in seconds, on any wait the observation needs (a blocked connection never becomes ready).
PublisherInterface
protocolPublisherInterfaceInterface for publishers. Each publisher owns its own routing logic.
methodsetupasync () -> NoneSet up the publisher.
method__call__async (message) -> NonePublish a message.
parammessagedictMessage body as a dictionary.
methodteardownasync () -> NoneTear down the publisher.
RouteFunc
protocolRouteFuncmethod__call__async (message) -> list[Route]Resolve the routes for a message. Can route to multiple destinations, for example, a processing destination and a logging destination.
parammessagedictMessage body as a dictionary.
Returns
list[Route]List of routes.
Functions
observer_binding_key
functionobserver_binding_key(exchange) -> str | NoneBinding key for a support worker observing all messages on exchange.
fanout receives everything (no key); topic binds the catch-all #.
This is only ever called with an exchange returned by observer_exchange,
which is always fanout or topic (never direct).
paramexchangeRMQExchangeConfigobserver_exchange
functionobserver_exchange(exchange) -> RMQExchangeConfigThe exchange the support workers (job-status, retry) observe.
fanout and topic can be observed in place — fanout broadcasts to
every bound queue, and topic supports a # catch-all binding — so the
data exchange is its own observer exchange.
A direct exchange routes by exact key and cannot be observed wholesale,
so the framework derives a dedicated fanout observer exchange
(<name>.observer). On a direct pipeline, workers echo their results to it
and lifecycle envelopes go to it, so observers see the whole pipeline; the
user never declares it. See warren/docs/routing.md.
paramexchangeRMQExchangeConfigobserver_route_func
functionobserver_route_func(exchange) -> RouteFunc | NoneRoute function for a support worker that publishes on exchange.
fanout ignores keys (None); topic routes by data_type via
MessageFieldRouter.
paramexchangeRMQExchangeConfigConstants
DATA_TYPE_FIELD
constantDATA_TYPE_FIELD = 'data_type'The message field used as the routing key by convention.
DELIVERY_COUNT_FIELD
constantDELIVERY_COUNT_FIELD = 'delivery_count'Message field carrying how many times the broker has delivered this message.
A requeue hands a message back unchanged, so the consumer manager republishes
a redelivered message with the count in its body (see max_deliveries).
LANE_FIELD
constantLANE_FIELD = 'lane'The message field conventionally used as a routing-key prefix (see MessageFieldRouter).
PRIORITY_FIELD
constantPRIORITY_FIELD = 'priority'The message field the RabbitMQ publisher copies to the AMQP priority property.
REPLAY_ROUTING_KEY_FIELD
constantREPLAY_ROUTING_KEY_FIELD = '_replay_routing_key'Message field the consumer manager stamps with the original routing key so a retry worker can replay the message back to the worker that failed.
ROUTING_PLAN_KEY
constantROUTING_PLAN_KEY = 'routing'Key under job_parameters where a RoutingPlan is carried.
Errors
PubSubSetupError
errorPubSubSetupErrorFailed to set up a pubsub resource (channel, exchange, queue, publisher).
Raised by the topology helpers and the publisher/consumer setup()
paths to give a transport-layer failure warren-specific context
(which exchange/queue/publisher) before it reaches the runner's
higher-level setup-phase wrapping.
PublishFailureException
errorPublishFailureExceptionFailed to publish message to downstream queue.