BabelQueue for Python

Polyglot Queues, Simplified. Read and write the canonical BabelQueue message envelope from Python — so your Python services (AI/ML, data processing, …) exchange messages with Laravel, Symfony, Go, .NET and Node over one strict JSON format, on the broker you already run.

This is the framework-agnostic Python core: the wire-envelope codec, contracts, and dead-letter helpers — zero runtime dependencies (standard library only). The full standard is documented at babelqueue.com.

Installation

pip install babelqueue

Requires Python >=3.9.

Usage

from babelqueue import EnvelopeCodec

# Produce — build the canonical envelope and publish the JSON to your broker.
envelope = EnvelopeCodec.make("urn:babel:orders:created", {"order_id": 1042})
body = EnvelopeCodec.encode(envelope)        # -> UTF-8 JSON string
# redis.rpush("queues:orders", body)  /  channel.basic_publish(body=body, ...)

# Consume — decode a message produced by ANY BabelQueue SDK.
incoming = EnvelopeCodec.decode(body)
urn      = incoming["job"]          # "urn:babel:orders:created"
data     = incoming["data"]         # {"order_id": 1042}
trace_id = incoming["trace_id"]     # correlate across services

The envelope is identical to every other SDK's:

{
  "job": "urn:babel:orders:created",
  "trace_id": "…",
  "data": { "order_id": 1042 },
  "meta": { "id": "…", "queue": "default", "lang": "python", "schema_version": 1, "created_at": 1749132727000 },
  "attempts": 0
}

Extension metadata, forbidden keys and the urn alias (1.20.0, ADR-0031)

# Producer meta: extra keys next to the core meta keys (recommended shape: meta.ext.<vendor>).
envelope = EnvelopeCodec.make(
    "urn:babel:orders:created", {"order_id": 1042},
    queue="orders", meta={"ext": {"acme": {"tier": "gold"}}},
)

• meta= is keyword-only. A key that collides with a core meta key (id, queue, lang, schema_version, created_at) or with a forbidden key (max_retries, attempts, source, ts) raises EnvelopeKeyError (a BabelQueueError; its pointer names the key, e.g. /meta/ts).
• Unknown top-level and meta keys of an inbound message survive a decode -> attempts + 1 -> encode hop (retry, redrive, dead-letter), so a vendor meta.ext block is never lost.
• EnvelopeCodec.encode rejects an envelope that carries a forbidden key (top-level timestamp; meta.max_retries, meta.attempts, meta.source, meta.ts) or the inbound-only top-level urn alias: it raises EnvelopeKeyError instead of dropping the key with a warning. decode is unchanged (it still warns and drops), so a decoded envelope can always be re-encoded.
• decode folds urn into job: a job that is missing, empty or only ASCII whitespace (space, \t, \n, \v, \f, \r) falls back to urn; when both carry a value, job wins and urn is dropped with a warning naming /urn. The decoded envelope never keeps a urn key.

Typed messages (optional)

from babelqueue import EnvelopeCodec, PolyglotMessage

class OrderCreated:                      # structurally a PolyglotMessage
    def __init__(self, order_id: int):
        self.order_id = order_id
    def get_babel_urn(self) -> str:
        return "urn:babel:orders:created"
    def to_payload(self) -> dict:
        return {"order_id": self.order_id}

envelope = EnvelopeCodec.from_message(OrderCreated(1042), queue="orders")

Continue an existing trace by adding get_babel_trace_id(self) -> str | None (see HasTraceId), or pass trace_id= to EnvelopeCodec.make.

Dead-letter

from babelqueue import dead_letter

dlq = dead_letter.annotate(envelope, "failed", "orders", attempts=3, error="boom")
# publish `EnvelopeCodec.encode(dlq)` to the "orders.dlq" queue

Runtime — produce & consume

For an end-to-end app, use BabelQueue with a broker. Broker clients come via extras:

pip install "babelqueue[redis]"   # redis://
pip install "babelqueue[amqp]"    # amqp:// (RabbitMQ)

from babelqueue import BabelQueue

app = BabelQueue("redis://localhost:6379/0", queue="orders")
# or: BabelQueue("amqp://guest:guest@localhost:5672/", queue="orders")

@app.handler("urn:babel:orders:created")
def on_order_created(data, meta):       # AI/ML, data processing, anything
    print("order", data["order_id"])

# producer (any service, any language) …
app.publish("urn:babel:orders:created", {"order_id": 1042})

# worker
app.run()                               # consume forever (Ctrl-C to stop)

• Routing is by URN; the wire format is the canonical envelope, so this consumes messages produced by any BabelQueue SDK.

• Handlers receive (data, meta), or (data, meta, message) to get the full envelope (incl. trace_id).

• Retry & dead-letter: failures are retried up to max_attempts (bumping the envelope's attempts); enable dead_letter=True to quarantine exhausted messages on <queue>.dlq. on_unknown_urn = fail | delete | release | dead_letter.

• Transports: redis:// (reliable-queue pattern; add ?laravel=1 to share a Redis queue with a Laravel BabelQueue worker using its reserved-set semantics — see below), kafka:// (opt-in retry tiers, see below), amqp:// (RabbitMQ via pika, with the contract AMQP properties) and memory:// (in-process, great for tests/local). Bring your own by passing transport=....

• Graceful shutdown: SIGTERM/SIGINT finish the in-flight handler, or release the message unchanged once shutdown_timeout expires. Delivery is at-least-once — keep handlers idempotent (see the idempotency helper).

Unsupported schema_version (SQS, Azure Service Bus, Kafka, Artemis, Pulsar)

These transports read the binding's schema-version property (bq-schema-version; Artemis bq_schema_version) before decoding the body. A missing or blank value (empty or only ASCII whitespace) follows the normal path; any other value must be exactly "1" (compared untrimmed; Azure Service Bus and Artemis also accept the integer 1). A message carrying anything else is not decoded: it reaches the runtime on the reserved route babelqueue:unsupported-schema-version:<v> (the version percent-escaped) and is never retried — it goes straight to the dead-letter queue (the incoming body, byte-identical, with no dead_letter block) when dead_letter is enabled, and is dropped otherwise, whatever max_attempts is. It does not depend on a broker RedrivePolicy. If the dead-letter write fails the message is left unacknowledged (an error is logged), is never retried and never reaches a handler; the dead-letter write is attempted again when the broker brings it back, which depends on the broker:

• SQS — redelivered when the visibility timeout expires.
• Azure Service Bus — redelivered when the message lock expires.
• Artemis — an unsettled AMQP 1.0 delivery is returned only when the link or connection closes, so it stays unsettled until then.
• Kafka — offsets are committed cumulatively per partition, so merely skipping the commit would lose the record once a later one is committed. The transport therefore implements the optional Rewinder protocol: it seeks the partition back to the failed record's offset and pauses _rewind_pause (1 s by default, so an unavailable DLQ is not hammered) before the record is polled again; the gate rejects it again and the dead-letter write is retried. While the dead-letter write keeps failing (for example the .dlq topic is not writable) the partition stalls on that record (head-of-line blocking, consumer lag grows) and the worker waits the pause on every attempt, which also delays its other partitions and queues. The commit does not move past the record as long as the seek succeeds; if the seek itself fails, an error naming the topic, partition and offset is logged and a later commit on that partition may skip the record, so recover it by hand from that offset. If the partition was lost to a rebalance, the new owner resumes from the last committed offset and the record is not lost.
• Pulsar — acknowledgement is per message; the unacknowledged message is redelivered by the broker.

A warning naming the version (cut at 64 characters plus ...) and queue is logged.

SQS release and poison messages

On a handler failure, unknown-URN release or shutdown release, the SQS transport always releases with ChangeMessageVisibility (broker-bindings §3.5) — it never re-sends a copy. The broker's ApproximateReceiveCount is the attempt counter, so a shutdown release also consumes an attempt. The default delay is 0 s (retry_backoff / unknown_urn_release_delay), i.e. the message is visible again immediately; delays outside 0..43200 s are clamped with a warning. A per-message delay at publish time is SqsTransport.publish(queue, body, delay_seconds=n) (DelaySeconds, 0..900 on a standard queue; a larger value, or any per-message delay on a FIFO queue, raises BabelQueueError instead of being truncated or dropped).

Poison-loop risk: a message that always fails is redelivered at once until max_attempts is reached. If the receive count is unavailable the SDK cannot advance attempts at all. Configure a native RedrivePolicy (maxReceiveCount ≥ max_attempts, pointing at <queue>.dlq) on every SQS queue as the broker-side backstop, and set a non-zero retry_backoff if the handler's dependencies need time to recover.

Retry back-off

BabelQueue(..., retry_backoff=, retry_backoff_multiplier=, retry_backoff_max=, retry_jitter=) set the worker-side retry delay (ADR-0034 section 9): after the n-th failure the message is released with d(n) = min(retry_backoff_max, retry_backoff * retry_backoff_multiplier ** (n - 1)) seconds (retry_backoff_max=0 means no cap) and retry_jitter="full" draws uniform(0, d(n)). The defaults (retry_backoff=0, multiplier 1, jitter "none") keep the previous behaviour: an immediate release. How the delay is honoured depends on the transport: SQS uses it as the visibility timeout, the Laravel-compatible Redis layout as the :delayed score and Kafka picks a retry tier (see below).

Kafka retry tiers and the re-inject consumer

Kafka has no per-message delay, so delayed delivery and retry back-off are SDK-owned retry topics (broker-bindings section 6.4, ADR-0034). They are opt-in: pass retry_tiers_ms to the transport (the recommended set is DEFAULT_RETRY_TIERS_MS = 5 s, 30 s, 5 min, 30 min):

from babelqueue import BabelQueue
from babelqueue.kafka_transport import KafkaTransport, DEFAULT_RETRY_TIERS_MS
from babelqueue.kafka_retry import RetryReinjector

transport = KafkaTransport("kafka://localhost:9092", retry_tiers_ms=DEFAULT_RETRY_TIERS_MS)
app = BabelQueue(transport=transport, queue="orders", max_attempts=5, dead_letter=True)

# Run one next to the workers, otherwise a parked message is never delivered again:
RetryReinjector("kafka://localhost:9092", work_topics=["orders"],
                retry_tiers_ms=DEFAULT_RETRY_TIERS_MS).run()

A failed message is republished (attempts + 1) to <queue>.retry.<n>, where n = min(attempts, N - 1) + 1; publish_delayed picks the smallest tier that covers the requested delay. Create the <queue>.retry.<n> topics up front. The RetryReinjector copies each record's value and key byte for byte back to its work topic once the tier delay has elapsed (headers bq-delay / bq-original-topic are dropped) and commits only after the broker confirmed the write, so delivery stays at-least-once. It holds every record it has accepted per partition (oldest first, a record never overtakes the one before it) and a failed write rewinds the partition to the failed record. Without retry_tiers_ms, publish_delayed and a retry raise BabelQueueError and nothing is produced. The worker does not stop on that error: it logs it, leaves the record uncommitted (the consumer seeks back to it and the loop pauses briefly, so the record is redelivered and never lost) and keeps consuming. A worker started without tiers while max_attempts is above 1 logs a warning at start-up: a failing handler cannot be retried (the record keeps failing in place) until tiers are configured. The re-inject delay is measured from the record timestamp: set message.timestamp.type=LogAppendTime on the .retry.<n> topics so that timestamp is the broker's, not a producer clock's.

Records without an identity are poison. A Kafka record whose JSON object has neither a valid string job nor a valid string urn (ADR-0034 section 10) is poison, not an unknown URN. With the DLQ on (dead_letter or on_unknown_urn=DEAD_LETTER) it is dead-lettered on its first delivery (wrapped with dead_letter.reason: "poison", key preserved) and never retried: the record is written to <topic>.dlq (which must exist) and its offset is committed once that write is confirmed; a failed write leaves it uncommitted and rewound. With the DLQ off nothing is written: on_unknown_urn=DELETE drops it, any other strategy retries it through the retry tiers up to max_attempts and then drops it. Without retry tiers that bounded retry has no tier to use, so (with the DLQ off and on_unknown_urn not DELETE) the poison record is committed, dropped and logged instead of blocking the partition (section 6.5 degrade-to-delete, ADR-0036 decision 5); enable dead_letter to keep it in <topic>.dlq. A failing handler without tiers still cannot be retried, as above.

Poison messages on every other transport

A message body that does not decode into an envelope (not a JSON object, malformed, a wrongly typed member, or bytes that are not valid UTF-8, even inside a JSON string) or that has neither a valid string job nor a valid string urn is poison on every transport (ADR-0036 decisions 2, 4 and 6; conformance identity_verdict). It is never handed to a handler and never takes the unknown-URN strategies, which are for a valid envelope whose URN has no handler:

• DLQ on (dead_letter or on_unknown_urn=DEAD_LETTER): written to <queue>.dlq at once, with no retry, as the ADR-0035 record (dead_letter.reason: "poison", the original body preserved, base64 plus raw_encoding when it is not UTF-8), then acknowledged. A failed write leaves it unacknowledged.
• on_unknown_urn=DELETE: acknowledged and dropped (logged).
• DLQ off: the bounded FAIL path (RELEASE is replaced by it): retried up to max_attempts, then dropped and logged. The count comes from the broker where it has one (SQS receive count, Pulsar bq-attempts / redelivery count), otherwise from a bare {"attempts": n} body the retry leaves.

On Redis (both layouts) an identity-less body is dead-lettered as poison in every configuration, including on_unknown_urn=DELETE and dead-lettering off (broker-bindings section 1.1). On the legacy (<queue>:processing) layout a body that Go's typed decode rejects is dead-lettered raw on its first delivery and removed from :processing.

Sharing a Redis queue with Laravel

By default the Redis transport owns its queue end-to-end (RPUSH to produce; BLMOVE into a <queue>:processing list to reserve; LREM on ack). Each reservation also writes a <token>:<raw> member (32 lowercase hex characters of token) into the <queue>:processing:deadlines sorted set, and before every reserve a sweep moves the lapsed reservations back to the end of the list, so a kill -9 no longer strands a message in :processing; a late ack of a swept reservation touches nothing of the newer one. retry_after (default 90 seconds; None / ?retry_after=null disables recovery) sets that deadline, and migration_batch_size (-1 = all, otherwise >= 1) bounds how many due members one sweep or migration moves. To consume a shared Laravel BabelQueue Redis queue instead, enable Laravel-compatible mode:

app = BabelQueue("redis://localhost:6379/0?laravel=1", queue="orders")
# or: RedisTransport("redis://localhost:6379/0", laravel_compat=True)

This replicates Laravel's stock Redis reservation exactly (§1 of the broker-bindings contract): the ready list is queues:<name>, reservations move into the queues:<name>:reserved sorted set scored by a retry_after deadline (default 60s), with a queues:<name>:delayed set and a queues:<name>:notify wake-up list. Reserve, ack and release run the byte-for-byte same Lua scripts Laravel uses, so the reserved member a Python worker writes is identical to a Laravel worker's — either side can ack (ZREM) the other's reservation, so a Python worker and a Laravel worker share one queue without losing or double-processing messages. Before each pop, expired reserved/delayed jobs migrate back to the ready list, so a crashed worker's in-flight job is re-reserved. Options: ?prefix=queues: and ?retry_after=90 (or the key_prefix / retry_after / migration_batch_size constructor args). A job whose body does not decode, or whose attempts is not a number, is reserved raw, dead-lettered byte for byte and acked, so it never lingers in :reserved. That holds when the body cannot be re-encoded either (for example a number such as 1e400, which real Redis' cjson refuses to serialise), and it holds under on_unknown_urn="delete" too: such a message is dead-lettered, never deleted. Without a dead-letter queue it is dropped, and the drop is logged as a warning. Dead-lettering a reserved message is one atomic script (ZREM the reservation, then RPUSH the dead-letter record and its notify entry); if the reservation expired first, nothing is written and a warning says so, because the job was already migrated back and will be delivered again.

Limit (A3): the Laravel-compatible layout stores the bare envelope (Laravel's reserve script has to decode it), so out-of-band transport headers such as the W3C traceparent cannot be carried there. They are dropped without an error; the trace_id inside the envelope is still preserved. The default (legacy) layout carries them.

Framework adapters — Celery & Django

Celery (pip install "babelqueue[celery]") — reuse your Celery app's broker for polyglot interop, and consume inbound messages as a Celery worker bootstep:

from babelqueue.celery import from_celery, install_worker

bq = from_celery(celery_app, queue="orders")    # runtime on Celery's broker
bq.publish("urn:babel:orders:created", {"order_id": 1042})

@bq.handler("urn:babel:orders:created")
def on_created(data, meta): ...

install_worker(celery_app, bq)                   # `celery worker` also drains URN messages

Django (pip install "babelqueue[django]") — add "babelqueue.django" to INSTALLED_APPS and configure a BABELQUEUE dict:

# settings.py
BABELQUEUE = {"broker_url": "redis://localhost:6379/0", "queue": "orders", "dead_letter": True}

from babelqueue.django import publish, get_app

publish("urn:babel:orders:created", {"order_id": 1042})   # in a view / signal

@get_app().handler("urn:babel:orders:created")            # register handlers at startup
def on_created(data, meta): ...

python manage.py babelqueue_worker --queue orders          # run the consumer

Transactional outbox (optional)

The babelqueue.outbox helper (ADR-0029) removes the producer dual write: "commit the business row" and "publish to the broker" are two systems that can disagree on a crash. Instead the message is persisted into your database, in the same transaction as the business data — so it commits or rolls back atomically with it — and a separate relay publishes the durable rows afterwards. No distributed transaction; exactly-once handoff into the broker, then at-least-once on the wire (the consumer dedupes on meta.id — see the idempotency helper, the mirror of this).

The core stays stdlib-only: OutboxStore is an abstract Protocol you bind to your own DB (the core ships no driver). The stored value is the EnvelopeCodec-encoded envelope byte-for-byte unchanged (frozen, schema_version: 1); the relay publishes those exact bytes — it never decodes, rebuilds or re-encodes — so trace_id is preserved end-to-end.

from babelqueue import BabelQueue, EnvelopeCodec
from babelqueue.outbox import Outbox, OutboxRelay, InMemoryOutboxStore

store = InMemoryOutboxStore()          # production: your own OutboxStore adapter, DB-backed
outbox = Outbox(store)

# write side — YOU own the transaction boundary (this is the whole point):
with db.transaction():                 # your own open transaction
    db.insert_order(order)             # the business write
    envelope = EnvelopeCodec.make("urn:babel:orders:created", {"order_id": 1042}, queue="orders")
    outbox.write(envelope)             # same connection, same tx — both, or neither

# read/publish side — run on a short interval, after the business tx commits:
app = BabelQueue("redis://localhost:6379/0", queue="orders")
relay = OutboxRelay(app.transport, store)
relay.drain()                          # publish all pending rows; flush() does one batch

Outbox.write only encodes and calls OutboxStore.save — it does not begin or commit anything. A save runs inside the transaction you already opened; you commit both together. OutboxRelay marks a row published only after the transport accepts it; a publish that raises is recorded via mark_failed (with a bounded, injectable-sleeper backoff) and left pending for a later pass, so one poison row never blocks the batch. Implement OutboxStore over your DB (claim rows oldest-first, ideally with SELECT … FOR UPDATE SKIP LOCKED so two relays don't double-publish); InMemoryOutboxStore is the reference for tests and single-process demos (no real transaction).

OpenTelemetry tracing (optional)

pip install "babelqueue[otel]" adds the optional babelqueue.otel module — the core never imports OpenTelemetry, so it stays zero-dependency. It emits a PRODUCER span per publish and a CONSUMER span per handled message, correlated across every hop and SDK, at two layered levels:

• trace_id correlation (v0.1): the envelope's trace_id maps 1:1 to an OTel trace id, so every hop that shares a trace_id shares one trace — with zero wire/transport change.
• W3C traceparent span linkage (v0.2): the producer also injects its active span context as a traceparent transport header (beside the frozen envelope, never in it), so the consumer starts its span as a true child of the producer span — real cross-hop parent-child linkage. With no traceparent present it falls back to the v0.1 trace_id behaviour, so enabling it is a strict, backward-compatible upgrade.

from opentelemetry import trace
from babelqueue import BabelQueue, otel

tracer = trace.get_tracer("orders")
app = BabelQueue("redis://localhost:6379/0", queue="orders")

# consumer: wrap_handler starts a CONSUMER span (child of the producer span when a
# traceparent rode along; else in the trace_id-derived trace)
app.register("urn:babel:orders:created", otel.wrap_handler(tracer, on_order_created))

# producer: otel.publish starts a PRODUCER span and carries traceparent + trace_id
otel.publish(tracer, app, "urn:babel:orders:created", {"order_id": 1042})

The traceparent rides the out-of-band transport-header seam (publish_with_headers / headers_from_context) — the same seam the replay-bypass marker uses — so the envelope stays frozen (schema_version: 1). It is carried on the in-memory, Redis (a transport-owned JSON frame, with bare-value back-compat), RabbitMQ (AMQP header table) and SQS (MessageAttributes) transports; where a transport can't carry it, propagation degrades cleanly to v0.1 trace_id correlation with no error.

GDPR field encryption (optional)

The optional babelqueue.gdpr module is the runtime, SDK-enforcement half of ADR-0030: it encrypts the data fields a registry declared x-gdpr-sensitive. The babelqueue-registry only declares and audits sensitivity (and can mask for safe logging); this module enforces it on the wire — a producer encrypts each marked leaf before publish, a consumer decrypts it after decode. It is strictly opt-in: if you never call protect/unprotect, nothing changes.

The envelope stays frozen (schema_version: 1): only the values inside data change — a sensitive leaf's value becomes a ciphertext string. data stays pure JSON, so an SDK without the key still carries the envelope (it just can't read the protected fields), and trace_id is never touched. Which fields are sensitive lives in the schema, not the message.

Because Python's standard library has no AES-GCM, the core ships no cipher and pulls no crypto dependency (GR-7). Cipher is a caller-provided Protocol you bind to a KMS / Vault / HSM / tokenisation service, or to a local AES-256-GCM via the optional cryptography library (pip install "babelqueue[gdpr]" — still not a core dependency):

import base64, os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

class AesGcmCipher:
    def __init__(self, key: bytes) -> None:       # 16/24/32-byte key — the CALLER's
        self._aead = AESGCM(key)

    def encrypt(self, plaintext: bytes) -> str:
        nonce = os.urandom(12)
        return base64.b64encode(nonce + self._aead.encrypt(nonce, plaintext, None)).decode("ascii")

    def decrypt(self, ciphertext: str) -> bytes:
        raw = base64.b64decode(ciphertext)
        return self._aead.decrypt(raw[:12], raw[12:], None)   # raises on wrong key / tamper

Wire it around the codec — validate cleartext (before protect, after unprotect), since a schema that constrains a sensitive field would reject the ciphertext string:

from babelqueue import EnvelopeCodec
from babelqueue.gdpr import protect, unprotect

# producer: encrypt marked leaves in place, then encode
protect(data, schema, cipher)
body = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:created", data, queue="orders"))

# consumer: decode, then decrypt marked leaves in place before the handler reads data
envelope = EnvelopeCodec.decode(body)
unprotect(envelope["data"], schema, cipher)

schema is the same per-URN JSON Schema the validator uses; babelqueue.schema.sensitive_paths locates every x-gdpr-sensitive leaf (nested objects, addresses[].line array items, container or root marks). An absent marked field is skipped; a re-run unprotect on cleartext is a no-op (non-string leaves are left alone); a wrong key / tampered ciphertext raises babelqueue.DecryptError so the consumer fails the message (retry / dead-letter) rather than process unreadable PII. The round-trip restores data byte-for-byte.

What's here

The codec/contracts/dead-letter (zero-dep core), the BabelQueue runtime (in-memory built in; Redis via [redis], RabbitMQ via [amqp]), and framework adapters for Celery ([celery]) and Django ([django]). Every layer speaks the one canonical envelope, so it interoperates with the PHP/Laravel, Symfony, Go, Node and .NET SDKs.

Testing

pip install -e ".[dev]"
pytest
# (or, dependency-free) python -m unittest discover -s tests

License

MIT © Muhammet Şafak. See LICENSE.