Reliability & governance helpers

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, wrap, InMemoryStore

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

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.