warren

Async consumer with identity — runs on the main event loop.

Heartbeat safety: this consumer runs on the same event loop as the RabbitMQ connection. The event loop must not be blocked, or heartbeats will fail and the connection will be dropped.

Rules:

• Pure async I/O (HTTP clients, DB queries) is safe — every await yields control for heartbeats.
• CPU-bound or blocking I/O work MUST be offloaded to a thread or process pool using asyncio.get_running_loop().run_in_executor().
• If your consumer is primarily CPU-bound or uses sync libraries, use SyncMessageConsumerInterface instead — the framework handles threading automatically.

import asyncio

import httpx


def cpu_heavy_parse(data: dict) -> dict:
    return data  # placeholder for CPU-bound work


class FetchAndParse:
    name = "fetch-and-parse-1"
    type = "fetch_and_parse"

    async def __call__(self, message: dict) -> dict | None:
        async with httpx.AsyncClient() as client:
            resp = await client.get("https://api.example.com/data")
            data = resp.json()

        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(None, cpu_heavy_parse, data)
        return {"parsed": result}

Identity for a message consumer — name and type.

Base protocol bundling identity with the consumer. Both async and sync consumer interfaces inherit from this, ensuring that any callable consumer also carries its identity.

Sync consumer with identity — framework runs this in a thread pool.

The framework automatically executes this consumer in a ThreadPoolExecutor via run_in_executor(), keeping the main event loop free for RabbitMQ heartbeats.

Rules:

• CPU-bound work and sync library calls (requests, file I/O, sync DB drivers) work naturally — no special handling needed.
• The consumer may spawn its own ProcessPoolExecutor or multiprocessing.Pool internally for CPU parallelism.
• Do NOT use asyncio.run() or asyncio.get_event_loop() — async objects from the main loop cannot be used from a different thread due to event loop affinity.
• If you need async I/O (aiohttp, async DB clients), use AsyncMessageConsumerInterface instead.

import httpx


def heavy_parse(payload: dict) -> dict:
    return payload  # placeholder for CPU-bound work


class ParseAndPost:
    name = "parse-and-post-1"
    type = "parse_and_post"

    def __call__(self, message: dict) -> dict | None:
        parsed = heavy_parse(message["data"])

        resp = httpx.post("https://api.example.com", json=parsed)
        resp.raise_for_status()
        return resp.json()

Union type accepted by the consumer manager — either sync or async.

Value for origin.type on messages that a publisher puts into the pipeline.

RoutingPlanRouter routes a message whose origin is not a plan node to the plan entry. JobStatusWorker stores the value as origin_type in the job results record. No other framework code reads this value.

Permanent failure — message is dropped. No retry.

Any other exception from a worker, except SoftFailureException and PublishFailureException, also counts as a hard failure. A hard failure publishes a hard-failure envelope and rejects the message without requeue.

Raised when an optional dependency is not installed.

Retryable failure — message is retried with back-off.

Workers raise this to signal a transient failure. The consumer manager catches it, publishes a soft-failure envelope on the observer exchange, and acks the original message.

• A running RetryWorker is required: it stores the envelope and republishes the message after the delay. Without a retry worker, the message is not retried.
• Without a control publisher, the consumer manager requeues the message after RetryConfig.fallback_requeue_delay.
• After retry_max attempts, the message is a hard failure.
• A value of 0 for retry_after, retry_max or backoff_base means "not set": the default of the consumer manager applies.
• RetryConfig.max_delay_cap and RetryConfig.max_retries_cap limit the resolved delay and the resolved maximum.

Base exception for the distributed processing framework.