Metadata-Version: 2.5
Name: babelqueue
Version: 1.16.0
Summary: Polyglot Queues, Simplified — the Python core: the canonical BabelQueue wire-envelope codec, contracts and dead-letter helpers.
Project-URL: Homepage, https://babelqueue.com
Project-URL: Source, https://github.com/BabelQueue/babelqueue-python
Project-URL: Issues, https://github.com/BabelQueue/babelqueue-python/issues
Project-URL: Changelog, https://github.com/BabelQueue/babelqueue-python/blob/main/CHANGELOG.md
Author-email: Muhammet Şafak <info@muhammetsafak.com>
License-Expression: MIT
License-File: LICENSE
Keywords: envelope,json,messaging,microservices,polyglot,queue
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Typing :: Typed
Requires-Python: >=3.9
Provides-Extra: amqp
Requires-Dist: pika>=1.3; extra == 'amqp'
Provides-Extra: artemis
Requires-Dist: python-qpid-proton>=0.39; extra == 'artemis'
Provides-Extra: azureservicebus
Requires-Dist: azure-identity>=1.15; extra == 'azureservicebus'
Requires-Dist: azure-servicebus>=7.11; extra == 'azureservicebus'
Provides-Extra: celery
Requires-Dist: celery>=5; extra == 'celery'
Provides-Extra: dev
Requires-Dist: mypy>=1.8; extra == 'dev'
Requires-Dist: pytest-cov>=4; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: ruff<0.16,>=0.5; extra == 'dev'
Provides-Extra: django
Requires-Dist: django>=4.2; extra == 'django'
Provides-Extra: gdpr
Requires-Dist: cryptography>=42; extra == 'gdpr'
Provides-Extra: kafka
Requires-Dist: confluent-kafka>=2.3; extra == 'kafka'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20; extra == 'otel'
Provides-Extra: pulsar
Requires-Dist: pulsar-client>=3.4; extra == 'pulsar'
Provides-Extra: redis
Requires-Dist: redis>=4; extra == 'redis'
Provides-Extra: sqs
Requires-Dist: boto3>=1.26; extra == 'sqs'
Description-Content-Type: text/markdown

# BabelQueue for Python

[![CI](https://github.com/BabelQueue/babelqueue-python/actions/workflows/ci.yml/badge.svg)](https://github.com/BabelQueue/babelqueue-python/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/babelqueue.svg)](https://pypi.org/project/babelqueue/)
[![Python](https://img.shields.io/pypi/pyversions/babelqueue.svg)](https://pypi.org/project/babelqueue/)
[![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)

> **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](https://babelqueue.com)**.

## Installation

```bash
pip install babelqueue
```

Requires Python `>=3.9`.

## Usage

```python
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:

```json
{
  "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
}
```

### Typed messages (optional)

```python
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

```python
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:

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

```python
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), `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 (reason `unknown_urn`, the body kept
verbatim) 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 `seek`s 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.

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

### 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). To consume a
**shared** Laravel BabelQueue Redis queue instead, enable Laravel-compatible mode:

```python
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](https://babelqueue.com/docs/spec/1.x/broker-bindings#redis)):
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=60`
(or the `key_prefix` / `retry_after` constructor args).

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

```python
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:

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

```python
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): ...
```

```bash
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.

```python
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.

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

```python
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:

```python
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

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

## License

MIT © Muhammet Şafak. See [LICENSE](LICENSE).
