API Reference

Protocols

QueueProtocol

source

Structural protocol for message queue / bus backends.

Implementations include in-memory, Redis Pub/Sub, RabbitMQ, Kafka, and SQS. connect() / close() manage the connection lifecycle; publish() and subscribe() handle message delivery.

connect

async def connect() -> None

source

Establish connection to the broker.

close

async def close() -> None

source

Close the connection and release resources.

publish

async def publish(
    topic: str,
    message: BusMessage
) -> None

source

Publish a message to a topic.

Parameters

Parameter	Type	Description
`topic`	str	Destination topic or queue name.
`message`	BusMessage	The message to publish.

Raises

Exception	Description
QueueError	For expected publish failures (quota, auth, etc.).

subscribe

async def subscribe(
    topic: str,
    handler: Callable[[BusMessage], Coroutine[Any, Any, None]]
) -> None

source

Subscribe a handler to a topic.

The handler is called for each message received. The backend is responsible for acknowledging or rejecting messages.

Parameters

Parameter	Type	Description
`topic`	str	Topic or queue name to subscribe to.
`handler`	Callable[[BusMessage], Coroutine[Any, Any, None]]	Async callable invoked per message.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check broker connectivity.

---

Classes

BatchedPublisher

source

Batch in-memory publish calls for a single atomic flush within one process.

Messages staged via stage are held in memory only. flush() publishes every unpublished entry concurrently; entries whose publish fails stay unpublished and are retried on the next flush().

Note

This class is purely in-memory — staged entries are lost on process restart or when the instance is discarded without a flush(). It provides no crash-safety or cross-restart durability guarantee. For durable, crash-safe publishing use the store-based outbox (OutboxStoreProtocol in oridecon.contracts.data.outbox with oridecon.sql.outbox.SQLOutboxStore and oridecon.sql.outbox.OutboxPublisher) inside the caller’s database transaction.

Example

publisher = BatchedPublisher(queue)
publisher.stage("orders.created", BusMessage(topic="orders.created", payload=data))
await publisher.flush()

__init__

def __init__(queue: QueueProtocol) -> None

source

Initialize publisher.

Parameters

Parameter	Type	Description
`queue`	QueueProtocol	Queue protocol to publish messages to.

stage

def stage(
    topic: str,
    message: BusMessage
) -> None

source

Stage a message locally for the next flush.

Parameters

Parameter	Type	Description
`topic`	str	Destination topic.
`message`	BusMessage	Message to stage.

flush

async def flush() -> None

source

Publish all staged messages that haven’t been published yet.

Failed publishes are logged and remain unpublished so the next flush() retries them. Publish failures never raise.

clear

def clear() -> None

source

Clear all staged entries.

---

BusMessage

source

A message transported through the queue backend.

id defaults to a new UUID4. timestamp defaults to the current wall-clock time. Both are set via field defaults so callers do not need to supply them.

Attributes: id: Unique message identifier. topic: Destination topic or queue name. payload: Arbitrary message body; must be serialisable by the backend. headers: Opaque string key-value headers passed to the broker. timestamp: Unix epoch timestamp when the message was created. ttl: Time-to-live in seconds; None uses the backend default. priority: Message priority (higher = more urgent); backend-dependent. delivery_guarantee: Delivery semantics for this message. retry_count: Number of times this message has been retried. max_retries: Maximum retries before routing to DLQ.

is_expired

def is_expired() -> bool

source

Return True if the message TTL has elapsed.

should_retry

def should_retry() -> bool

source

Return True if the message may be retried.

---

ConsumerRegisteredEvent

source

Consumer was registered to a queue.

Consumed by: consumer tracking, lifecycle management, monitoring.

---

DeadLetterQueue

source

In-memory dead letter queue for storing failed messages.

__init__

def __init__(max_size: int = 1000) -> None

source

Initialize DLQ.

Parameters

Parameter	Type	Description
`max_size`	int	Maximum number of entries to store.

push

async def push(
    message: BusMessage,
    error: str
) -> None

source

Add a failed message to the DLQ.

Parameters

Parameter	Type	Description
`message`	BusMessage	The message that failed.
`error`	str	Error description.

drain

async def drain() -> list[DeadLetterEntry]

source

Remove and return all entries from the DLQ.

Returns

Type	Description
list[DeadLetterEntry]	List of dead letter entries.

size

property size() -> int

source

Return current number of entries in DLQ.

---

MessageConsumedEvent

source

Message was successfully consumed from the queue.

Consumed by: message tracking, audit logging, metrics collection.

---

MessageConsumedHook

source

Payload fired when a message is successfully consumed by a handler.

Attributes: queue_name: Name of the queue or topic the message came from. message_type: Type label of the consumed message.

---

MessageConsumer

source

Base class for message consumer workers.

Subclass this and implement handle() to create a worker that subscribes to a topic and processes incoming messages.

__init__

def __init__(queue: QueueProtocol) -> None

source

Initialize consumer.

Parameters

Parameter	Type	Description
`queue`	QueueProtocol	Queue protocol instance for subscribing.

start

async def start() -> None

source

Start consuming messages from the subscribed topic.

stop

async def stop() -> None

source

Stop consuming messages.

handle

async def handle(message: BusMessage) -> None

source

Handle a received message. Subclasses must implement this.

Parameters

Parameter	Type	Description
`message`	BusMessage	Message to handle.

---

MessageDeadLetteredEvent

source

Message failed and was moved to dead-letter queue.

Consumed by: error handling, retry management, incident tracking.

---

MessagePipeline

source

Ordered chain of middleware around a terminal handler.

__init__

def __init__() -> None

source

Initialize pipeline.

add

def add(middleware: MiddlewareBase) -> None

source

Add middleware to the pipeline.

Parameters

Parameter	Type	Description
`middleware`	MiddlewareBase	Middleware instance to add.

execute

async def execute(
    message: BusMessage,
    handler: Any
) -> None

source

Execute the middleware chain and terminal handler.

Parameters

Parameter	Type	Description
`message`	BusMessage	The message to process.
`handler`	Any	Terminal async callable to invoke after all middleware.

---

MessagePublishedHook

source

Payload fired when a message is published to a queue or topic.

Attributes: queue_name: Name of the queue or topic the message was published to. message_type: Type label of the published message.

---

MiddlewareBase

source

Base class for message pipeline middleware.

process

async def process(
    message: BusMessage,
    next_handler: Any
) -> None

source

Process a message and pass to the next handler.

Parameters

Parameter	Type	Description
`message`	BusMessage	The message to process.
`next_handler`	Any	Async callable to invoke the next middleware or terminal handler.

---

QueueDrainedHook

source

Payload fired when a queue is fully drained (all pending messages processed).

Attributes: queue_name: Name of the queue that was drained.

---

QueueModule

source

Message queue/bus integration with Named DI multi-backend support.

Registers QueueProtocol for constructor injection.

Usage

from oridecon.queue.config import QueueConfig, NamedQueueConfig
from oridecon.queue.module import QueueModule

@module(
    imports=[QueueModule.configure(QueueConfig(backends=[...]))]
)
class AppModule(Module):
    pass

from oridecon.queue.config import QueueConfig, NamedQueueConfig
from oridecon.queue.module import QueueModule

@module(
    imports=[QueueModule.configure(QueueConfig(backends=[...]))]
)
class AppModule(Module):
    pass

Scope consumers into a feature module

@module(
    imports=[
        QueueModule.configure(QueueConfig(...)),
        QueueModule.scope(OrderConsumer, PaymentConsumer),
    ]
)
class OrdersFeatureModule(Module):
    pass

@module(
    imports=[
        QueueModule.configure(QueueConfig(...)),
        QueueModule.scope(OrderConsumer, PaymentConsumer),
    ]
)
class OrdersFeatureModule(Module):
    pass

Named injection

class MyService:
    def __init__(
        self,
        queue: QueueProtocol,                                   # primary
        events: Annotated[QueueProtocol, Named("events")],     # named
    ) -> None: ...

class MyService:
    def __init__(
        self,
        queue: QueueProtocol,                                   # primary
        events: Annotated[QueueProtocol, Named("events")],     # named
    ) -> None: ...

configure

def configure(
    cls,
    config: QueueConfig | Any | None = None
) -> DynamicModule

source

Create a QueueModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	QueueConfig | Any | None	QueueConfig or ``None`` to use defaults (reads from environment variables).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

scope

def scope(
    cls,
    *consumers: type,
    **kwargs: Any
) -> DynamicModule

source

Scope queue consumer classes into a feature module.

Registers the given consumer classes as providers so they are discovered and wired by the queue backend. The parent module graph must already include QueueModule.configure — this does not create a new queue connection.

Uses the anonymous token pattern so both configure() and scope() can coexist in the same compiled graph without a ModuleDuplicateError.

Example

@module(
    imports=[
        QueueModule.configure(QueueConfig(...)),
        QueueModule.scope(OrderConsumer, PaymentConsumer),
    ]
)
class OrdersFeatureModule(Module):
    pass

@module(
    imports=[
        QueueModule.configure(QueueConfig(...)),
        QueueModule.scope(OrderConsumer, PaymentConsumer),
    ]
)
class OrdersFeatureModule(Module):
    pass

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	A DynamicModule scoped to this feature.

stub

def stub(
    cls,
    config: QueueConfig | None = None
) -> DynamicModule

source

Return an in-memory QueueModule for unit testing.

Uses a memory backend so no external broker is required.

Parameters

Parameter	Type	Description
`config`	QueueConfig | None	Optional config override; defaults to empty QueueConfig (which uses memory backend).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

Exceptions

AzureServiceBusQueueError

source

Azure Service Bus queue failure.

__init__

def __init__(
    message: str = 'Azure Service Bus error',
    **kwargs: Any
) -> None

source

---

GCPPubSubQueueError

source

GCP Pub/Sub queue failure.

__init__

def __init__(
    message: str = 'GCP Pub/Sub error',
    **kwargs: Any
) -> None

source

---

KafkaQueueError

source

Kafka queue failure.

__init__

def __init__(
    message: str = 'Kafka error',
    **kwargs: Any
) -> None

source

---

QueueError

source

Base for expected, recoverable queue / bus failures.

Attributes: backend: Name of the queue backend (e.g. “redis”, “kafka”). topic: Topic or queue name where the failure occurred.

__init__

def __init__(
    message: str = 'Queue error',
    *,
    backend: str = 'unknown',
    topic: str = 'unknown',
    **kwargs: Any
) -> None

source

---

RabbitMQQueueError

source

RabbitMQ queue failure.

__init__

def __init__(
    message: str = 'RabbitMQ error',
    **kwargs: Any
) -> None

source

---

RedisQueueError

source

Redis queue failure.

__init__

def __init__(
    message: str = 'Redis queue error',
    **kwargs: Any
) -> None

source

---

SQSQueueError

source

SQS queue failure.

__init__

def __init__(
    message: str = 'SQS error',
    **kwargs: Any
) -> None

source

---