Skip to main content
Version: 1.0.4

API Reference

Flat reference for every public symbol in django_rmq. Signatures and parameter descriptions are taken directly from the source code.


Producer​

from django_rmq.producer import Producer

Publishes messages to RabbitMQ through a thread-local blocking channel.

Producer.__init__​

def __init__(
self,
exchange: str = '',
queue: QueueConfig | str = '',
using: str | None = None,
) -> None
ParameterTypeDefaultDescription
exchangestr''Exchange name. Empty string means the default (direct) exchange.
queueQueueConfig | str''Queue configuration or name. Used as default routing key; declared lazily on first publish. Empty string disables queue declaration (exchange-only mode).
usingstr | NoneNoneConnection alias from RABBITMQ_CONNECTIONS. May be omitted when exactly one alias is configured.

Producer.publish​

def publish(
self,
body: str | bytes,
routing_key: str | None = None,
properties: BasicProperties | None = None,
) -> None
ParameterTypeDefaultDescription
bodystr | bytesrequiredMessage body. Strings are encoded to bytes (UTF-8).
routing_keystr | NoneNoneRouting key. Defaults to self.queue (the queue name) when not provided.
propertiesBasicProperties | NoneNoneAMQP message properties. When omitted, created with content_type='application/json'. delivery_mode is always forced to 2 (persistent).

Returns: None

Raises:

  • pika.exceptions.UnroutableError — broker returned the message because no queue matched the routing key (requires publisher confirms + mandatory=True, both always enabled).
  • pika.exceptions.NackError — broker nacked the message.
  • Any exception from _RECONNECTABLE_ERRORS if both the first attempt and the single retry fail.

Producer.__call__​

Allows a Producer instance to be used as a function decorator. The decorated function must return str, bytes, or None. The return value is published automatically; None skips publishing.

def __call__(
self,
func: Callable[..., str | bytes | None],
) -> Callable[..., str | bytes | None]

Raises: TypeError if the decorated function returns a type other than str, bytes, or None.

Example:

from django_rmq.producer import Producer

producer: Producer = Producer(queue='notifications')


@producer
def build_notification(user_id: int) -> str:
return f'{{"user_id": {user_id}}}'

Consumer​

from django_rmq.consumer import Consumer

Consumes messages from a single RabbitMQ queue with one registered handler. Reconnects on transient AMQP errors with exponential backoff.

Consumer.__init__​

def __init__(
self,
queue: QueueConfig | str,
prefetch_count: int = 1,
reconnect_initial_backoff: float | None = None,
reconnect_max_backoff: float | None = None,
using: str | None = None,
) -> None
ParameterTypeDefaultDescription
queueQueueConfig | strrequiredQueue to consume from. QueueConfig triggers an active declare with arguments; a plain string triggers a durable declare.
prefetch_countint1Maximum unacknowledged messages delivered at once (basic_qos).
reconnect_initial_backofffloat | NoneNoneInitial reconnect delay in seconds. Falls back to the alias config value when None.
reconnect_max_backofffloat | NoneNoneMaximum reconnect delay in seconds (exponential backoff cap). Falls back to alias config when None.
usingstr | NoneNoneConnection alias. May be omitted when exactly one alias is configured.

Consumer.handler​

def handler(self, func: MessageCallback) -> MessageCallback

Registers a callback for incoming messages. Can be used as a decorator.

Raises: RuntimeError if a handler is already registered on this consumer.

Consumer.__call__​

Shorthand for @consumer.handler. Identical behavior.

def __call__(self, func: MessageCallback) -> MessageCallback

Consumer.consume​

def consume(self, stop_event: threading.Event | None = None) -> None

Starts the consume loop. Reconnects on transient errors. Exits when stop_event is set or an unrecoverable error is raised.

ParameterTypeDefaultDescription
stop_eventthreading.Event | NoneNoneEvent that signals graceful shutdown. When None, an internal event (never set) is created — the consumer runs until an error.

Consumer properties​

PropertyTypeDescription
prefetch_countintMaximum unacknowledged messages delivered at once.
usingstr | NoneConnection alias, or None when the single alias is used implicitly.
handler_namestrName of the registered handler function, or 'unregistered' if none.

MessageCallback​

from django_rmq.consumer import MessageCallback

Type alias for the handler callable signature:

MessageCallback = Callable[
[BlockingChannel, Basic.Deliver, BasicProperties, bytes],
None,
]
ArgumentTypeDescription
chBlockingChannelChannel the message was delivered on. Used to call basic_ack / basic_nack.
methodBasic.DeliverDelivery metadata, including delivery_tag.
propsBasicPropertiesAMQP message properties.
bodybytesRaw message body.

QueueConfig​

from django_rmq.queues.queue_config import QueueConfig

Frozen dataclass for declarative queue configuration.

@dataclass(frozen=True)
class QueueConfig:
name: str
durable: bool = True
dead_letter_exchange: str | None = None
dead_letter_routing_key: str | None = None
FieldTypeDefaultDescription
namestrrequiredQueue name. Also used as str(queue_config).
durableboolTrueQueue survives broker restarts.
dead_letter_exchangestr | NoneNoneExchange dead-lettered messages are routed to (x-dead-letter-exchange).
dead_letter_routing_keystr | NoneNoneRouting key for dead-lettered messages (x-dead-letter-routing-key).

QueueConfig.arguments property​

@property
def arguments(self) -> dict[str, Any] | None

Builds the AMQP arguments dict for queue declaration from the dead-letter fields. Returns None when neither field is set.


RabbitMQConfig​

from django_rmq.dto.rabbitmq_config import RabbitMQConfig

Frozen dataclass holding the resolved configuration for one connection alias. Created internally by RabbitMQAppConfig.ready() from the RABBITMQ_CONNECTIONS setting. Exposed on RabbitMQConnectionManager.config.

@dataclass(frozen=True)
class RabbitMQConfig:
host: str
port: int
virtual_host: str
user: str
password: str
heartbeat: int
blocked_connection_timeout: int
reconnect_initial_backoff: float
reconnect_max_backoff: float
FieldTypeDescription
hoststrBroker hostname or IP.
portintAMQP port.
virtual_hoststrVirtual host to connect to.
userstrUsername for PlainCredentials.
passwordstrPassword for PlainCredentials.
heartbeatintHeartbeat interval in seconds.
blocked_connection_timeoutintSeconds to wait while the connection is blocked by the broker.
reconnect_initial_backofffloatInitial consumer reconnect delay in seconds.
reconnect_max_backofffloatMaximum consumer reconnect delay (exponential backoff cap).

RabbitMQConnectionManager​

from django_rmq.connections import RabbitMQConnectionManager

Manages thread-local connections for producer and consumer roles. Instantiated once per alias by RabbitMQAppConfig.ready().

RabbitMQConnectionManager.__init__​

def __init__(self, config: RabbitMQConfig) -> None

RabbitMQConnectionManager.get_producer_connection​

def get_producer_connection(self) -> BlockingConnection

Returns the thread-local BlockingConnection used by producers on the current thread. Creates it on first call; reuses the cached instance while it is open.

RabbitMQConnectionManager.get_consumer_connection​

def get_consumer_connection(self) -> BlockingConnection

Returns the thread-local BlockingConnection used by consumers on the current thread. Kept separate from the producer connection to allow safe publishing from inside a handler.

RabbitMQConnectionManager.get_producer_channel​

def get_producer_channel(self) -> BlockingChannel

Returns the thread-local producer channel with publisher confirms enabled (confirm_delivery()). Creates it on first call; reuses while open.

RabbitMQConnectionManager.reset_producer_channel​

def reset_producer_channel(self) -> None

Closes and discards the cached producer channel and connection. Called automatically by Producer.publish after a reconnectable error so the next publish opens a fresh channel. Safe to call even when no channel is cached.


get_connection_manager​

from django_rmq.connections import get_connection_manager
def get_connection_manager(using: str | None = None) -> RabbitMQConnectionManager

Returns the RabbitMQConnectionManager for the given alias.

ParameterTypeDefaultDescription
usingstr | NoneNoneAlias to resolve. May be omitted when exactly one alias is configured.

Raises: ImproperlyConfigured — alias not found, ambiguous (multiple aliases, using omitted), or django_rmq not initialized.


ConsumersRegistry​

from django_rmq.registries.registry import ConsumersRegistry

Holds the consumers registered for one connection alias.

ConsumersRegistry.register​

def register(self, consumer: Consumer) -> None

Adds a consumer to the registry.

ConsumersRegistry.all​

def all(self) -> list[Consumer]

Returns a copy of all registered consumers. Mutating the returned list does not affect the registry.


get_consumers_registry​

from django_rmq.registries.registry import get_consumers_registry
def get_consumers_registry(using: str | None = None) -> ConsumersRegistry

Returns the ConsumersRegistry for the given alias.

Raises: ImproperlyConfigured — same conditions as get_connection_manager.


SetupRegistry​

from django_rmq.registries.setup_registry import SetupRegistry

Holds idempotent topology-setup functions for one connection alias.

SetupRegistry.register​

def register(self, fn: SetupFn) -> None

Adds a setup function to the registry. Functions are invoked in registration order.

SetupRegistry.run_all​

def run_all(self, channel: BlockingChannel) -> None

Runs every registered setup function on the given channel in registration order.


get_setup_registry​

from django_rmq.registries.setup_registry import get_setup_registry
def get_setup_registry(using: str | None = None) -> SetupRegistry

Returns the SetupRegistry for the given alias.

Raises: ImproperlyConfigured — same conditions as get_connection_manager.


SetupFn​

from django_rmq.registries.setup_registry import SetupFn

Type alias for a topology setup callable:

SetupFn = Callable[[BlockingChannel], None]

A SetupFn receives an open BlockingChannel and must be idempotent (safe to call multiple times with the same outcome).


RabbitMQAppConfig​

from django_rmq.apps import RabbitMQAppConfig

Django AppConfig for django_rmq. Registered automatically when 'django_rmq' is in INSTALLED_APPS.

RabbitMQAppConfig.ready​

def ready(self) -> None

Reads RABBITMQ_CONNECTIONS from Django settings and builds, per alias, one RabbitMQConnectionManager, one SetupRegistry, and one ConsumersRegistry. Stores them on the django_rmq module as connection_managers, setup_registries, and consumers_registries.

Raises: ImproperlyConfigured — RABBITMQ_CONNECTIONS is missing or empty.


Module globals (django_rmq)​

import django_rmq

django_rmq.connection_managers # dict[str, RabbitMQConnectionManager] | None
django_rmq.setup_registries # dict[str, SetupRegistry] | None
django_rmq.consumers_registries # dict[str, ConsumersRegistry] | None

All three are None until RabbitMQAppConfig.ready() runs. Use the accessor functions (get_connection_manager, get_consumers_registry, get_setup_registry) rather than accessing these dicts directly.


Management commands​

setup_rabbitmq_topology​

uv run python manage.py setup_rabbitmq_topology [--using ALIAS]

Declares all exchanges, queues, and bindings registered in SetupRegistry for one or all aliases. Idempotent — safe to run on every deploy.

OptionDescription
--using ALIASRun only for the named alias. Omit to run for every alias.

After setup it prints a report of declared exchanges, queues, and bindings, de-duplicated by name.

start_consumers​

uv run python manage.py start_consumers [--using ALIAS]

Starts every consumer registered in ConsumersRegistry for one or all aliases. Each consumer runs in its own thread sharing a single stop_event. SIGTERM and SIGINT set the event for graceful shutdown; the command joins all threads before returning.

OptionDescription
--using ALIASStart consumers for the named alias only. Omit for all aliases.

Settings reference​

KeyTypeDescription
HOSTstrBroker hostname or IP.
PORTintAMQP port (typically 5672).
VIRTUAL_HOSTstrVirtual host (e.g. '/').
USERstrAuthentication username.
PASSWORDstrAuthentication password.
HEARTBEATintHeartbeat interval in seconds.
BLOCKED_CONNECTION_TIMEOUTintSeconds before a blocked connection times out.
RECONNECT_INITIAL_BACKOFFfloatInitial consumer reconnect delay in seconds.
RECONNECT_MAX_BACKOFFfloatMaximum consumer reconnect delay (backoff cap).

All nine keys are required for every alias. Missing keys raise a KeyError during AppConfig.ready(). See Configuration for a full settings example.