---
title: "Reliability & governance helpers"
description: "The optional, stdlib-only reliability and governance helpers in the Python core — idempotency (wrap + seen-set store), the transactional outbox, DLQ redrive + replay-bypass, GDPR field encryption (with the [gdpr] extra for the reference cipher), and OpenTelemetry traceparent (the [otel] extra). Each composes with the runtime, adds nothing to the frozen envelope, and mirrors the cross-SDK spec."
source: https://babelqueue.com/docs/babelqueue-python/1.x/reliability-and-governance/
updated: 2026-06-21T00:00:00.000Z
---

# 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`):

```python
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](/docs/spec/1.x/idempotency/) and the
[store deep-dive](/docs/spec/1.x/idempotency-stores/).

## 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.

```python
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](/docs/spec/1.x/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:

```python
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](/docs/spec/1.x/redrive-and-replay/).

## 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.

```python
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](/docs/spec/1.x/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.

```python
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](/docs/spec/1.x/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](/docs/spec/1.x/schema-validation/).