Reliability & governance helpers

Last updated: October 13, 2026

Beyond produce/consume, the Python core ships optional, opt-in helpers that mechanise the reliability and governance contracts the wire spec defines. The core stays stdlib-only; the crypto and OTel references live behind extras. None touch the frozen envelope (schema_version: 1); each is a tooling layer over the codec. They are the Python face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

wrap(store, handler) makes a handler run at most once per meta.id, even on at-least-once redelivery. It keeps the handler’s signature (via functools.wraps):

from babelqueue import BabelQueue, InMemoryStore
from babelqueue.idempotency import wrap

app = BabelQueue("redis://localhost:6379/0", queue="orders")
store = InMemoryStore()  # tests / single process
app.register("urn:babel:orders:created", wrap(store, on_order_created))

The in-memory InMemoryStore is the reference; for a fleet, implement the IdempotencyStore protocol (seen / remember / forget) over a shared backend (no persistent store ships in core yet — bring your own). See Idempotency and the store deep-dive.

Transactional outbox

Outbox / OutboxRelay remove the producer dual write: persist the encoded envelope in the same DB transaction as your business row, then a relay publishes the durable rows verbatim.

from babelqueue import Outbox, OutboxRelay, InMemoryOutboxStore

store = InMemoryOutboxStore()   # production: a DB-backed OutboxStore
outbox = Outbox(store)

# inside YOUR DB transaction, beside the business write — no commit of its own:
row_id = outbox.write(envelope) # encodes via the codec, calls store.save()

# later, a relay drains the durable rows to the broker:
result = OutboxRelay(app.transport, store).drain()  # result.published / result.failed

OutboxStore is the four-method protocol (save / fetch_unpublished / mark_published / mark_failed); the relay forwards through the runtime’s publish-only Transport (publish(queue, body)). The transaction boundary is yours, and the relay backoff is in seconds. See Transactional Outbox.

DLQ redrive & replay-bypass

redrive(transport, dlq, ...) moves dead-lettered messages back onto a queue — dead_letter removed, attempts reset to 0, everything else preserved — with dry_run, select, to_queue (sandbox) and bypass. The replay guard skips effects that already fired:

from babelqueue import is_replay, bypass_external_effects
from babelqueue.redrive import redrive

# inspect / sandbox-replay, stamping bq-replay-bypass
res = redrive(app.transport, "orders.dlq", to_queue="sandbox", bypass=True)

@app.handler("urn:babel:orders:created")
def on_order_created(data, meta):
    save_order(data)                                  # idempotent core — always runs
    bypass_external_effects(lambda: send_email(data)) # external effect — skipped on replay

On Kafka, redrive keeps the DLQ record’s key (ReceivedMessage.key) on the write back to the work topic as well as on every restore write back to the DLQ: a null key stays null and an empty key stays empty.

is_replay() reads the per-message replay flag the runtime sets (a contextvars flag), so it takes no argument. See DLQ redrive & replay-bypass.

GDPR field encryption

protect / unprotect encrypt only the data leaves a schema marks x-gdpr-sensitive, in place. Cipher is a caller-bound protocol; the core ships no concrete cipher (Python’s stdlib has no AES-GCM) — the AES-256-GCM reference rides the optional babelqueue[gdpr] extra (cryptography). Validate cleartext — protect after validation on produce, unprotect before validation on consume.

from babelqueue import protect, unprotect, Cipher, DecryptError
from babelqueue.schema import sensitive_paths

protect(data, schema, cipher)      # producer: encrypt marked leaves (cipher implements Cipher)
try:
    unprotect(data, schema, cipher)  # consumer: inverse
except DecryptError:
    raise                            # wrong key / tampered → retry / dead-letter

See GDPR field encryption.

OpenTelemetry (traceparent)

The babelqueue.otel module (the babelqueue[otel] extra, so importing the core stays dependency-free) emits publish <urn> / process <urn> spans and, on a header-carrying transport, injects/extracts the W3C traceparent so a consumer span is a true child of the producer span — degrading to v0.1 trace_id correlation otherwise. In-memory, Redis, AMQP and SQS carry it.

from babelqueue import otel
from opentelemetry import trace

tracer = trace.get_tracer("orders")
app.register("urn:babel:orders:created", otel.wrap_handler(tracer, on_order_created))
otel.publish(tracer, app, "urn:babel:orders:created", {"order_id": 1042})

See Observability.

Per-URN schema validation

validate(provider, urn, data) (producer guard) / wrap(provider, urn, handler) (consumer safety net) validate a message’s data against the JSON Schema registered for its URN, bridged from a babelqueue-registry manifest via DirProvider. See Per-URN schema validation.