feat: Celery + Django adapters (v0.5.0)

Author: muhammetsafak

.github/workflows/ci.yml

@@ -24,10 +24,10 @@ jobs:
         with:
           python-version: ${{ matrix.python }}
 
-      - name: Install
+      - name: Install (with Celery + Django adapters)
         run: |
           python -m pip install --upgrade pip
-          pip install -e ".[dev]"
+          pip install -e ".[dev,celery,django]"
 
       - name: Run tests
         run: pytest

CHANGELOG.md

@@ -9,6 +9,19 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [0.5.0] - 2026-06-06
+
+### Added
+- **Celery adapter** (`babelqueue.celery`, `[celery]` extra) — `from_celery(app)`
+  builds a `BabelQueue` runtime on a Celery app's broker, and `install_worker(app)`
+  registers a Celery worker bootstep that drains URN-routed polyglot messages in a
+  background thread alongside Celery's own consumer.
+- **Django adapter** (`babelqueue.django`, `[django]` extra) — settings-driven
+  `BABELQUEUE` config, `get_app()` / `publish()` shortcuts, and a
+  `manage.py babelqueue_worker` management command. Add `"babelqueue.django"` to
+  `INSTALLED_APPS`.
+- Both adapters lazy-import their framework, so the core stays dependency-free.
+
 ## [0.4.0] - 2026-06-06
 
 ### Added
@@ -55,7 +68,8 @@ The envelope wire format is versioned separately by `meta.schema_version`
 - Pre-1.0: the public API may change before the `1.0.0` tag.
 - The core has **zero runtime dependencies** (standard library only); Python `>=3.9`.
 
-[Unreleased]: https://github.com/BabelQueue/babelqueue-python/compare/v0.4.0...HEAD
+[Unreleased]: https://github.com/BabelQueue/babelqueue-python/compare/v0.5.0...HEAD
+[0.5.0]: https://github.com/BabelQueue/babelqueue-python/compare/v0.4.0...v0.5.0
 [0.4.0]: https://github.com/BabelQueue/babelqueue-python/compare/v0.3.0...v0.4.0
 [0.3.0]: https://github.com/BabelQueue/babelqueue-python/compare/v0.2.0...v0.3.0
 [0.2.0]: https://github.com/BabelQueue/babelqueue-python/compare/v0.1.0...v0.2.0

README.md

@@ -118,13 +118,51 @@ app.run()                               # consume forever (Ctrl-C to stop)
   `pika`, with the contract AMQP properties) and `memory://` (in-process, great for
   tests/local). Bring your own by passing `transport=...`.
 
-> **Celery** / **Django** adapters are the next iterations.
+## 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
+```
 
 ## What's here
 
-The codec/contracts/dead-letter (zero-dep core) **and** the `BabelQueue` runtime
-above (in-memory built in; Redis via `[redis]`, RabbitMQ via `[amqp]`). For
-framework integration, the Celery and Django adapters are planned.
+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
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "babelqueue"
-version = "0.4.0"
+version = "0.5.0"
 description = "Polyglot Queues, Simplified — the Python core: the canonical BabelQueue wire-envelope codec, contracts and dead-letter helpers."
 readme = "README.md"
 requires-python = ">=3.9"
@@ -28,9 +28,11 @@ classifiers = [
 dependencies = []
 
 [project.optional-dependencies]
-# Planned runtime / adapters — standard, zero-heavy-dep drivers.
+# Optional runtime drivers + framework adapters — standard, zero-heavy-dep.
 redis = ["redis>=4"]
 amqp = ["pika>=1.3"]
+celery = ["celery>=5"]
+django = ["django>=4.2"]
 dev = ["pytest>=7"]
 
 [project.urls]

src/babelqueue/__init__.py

@@ -19,7 +19,7 @@
 from .routing import UnknownUrnStrategy
 from .transport import InMemoryTransport, ReceivedMessage, Transport
 
-__version__ = "0.4.0"
+__version__ = "0.5.0"
 
 __all__ = [
     "BabelQueue",

src/babelqueue/celery.py

@@ -0,0 +1,88 @@
+"""Celery integration. Requires the ``celery`` extra:
+
+    pip install "babelqueue[celery]"
+
+A Celery app already configures a broker (Redis/RabbitMQ). :func:`from_celery`
+builds a :class:`~babelqueue.BabelQueue` runtime on that *same* broker, so a
+Celery-based service produces and consumes the canonical polyglot envelope
+alongside its Celery tasks — interoperating with the PHP/Laravel, Go, Node, ...
+SDKs. :func:`install_worker` runs that consumer as a Celery worker *bootstep* (a
+daemon thread started on ``celery worker``), so one process handles both Celery
+tasks and inbound polyglot messages.
+
+``celery`` is imported lazily, so the core stays dependency-free.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any, Optional
+
+from .app import BabelQueue
+from .exceptions import BabelQueueError
+
+
+def broker_url(celery_app: Any) -> str:
+    """Extract the broker URL from a Celery app (supports old/new config keys)."""
+    conf = getattr(celery_app, "conf", None)
+    url = None
+    if conf is not None:
+        url = getattr(conf, "broker_url", None)
+        if not url and hasattr(conf, "get"):
+            url = conf.get("broker_url") or conf.get("BROKER_URL")
+    if not url:
+        raise BabelQueueError(
+            "The Celery app has no broker configured; set broker_url before calling from_celery()."
+        )
+    return str(url)
+
+
+def from_celery(celery_app: Any, **kwargs: Any) -> BabelQueue:
+    """Build a :class:`~babelqueue.BabelQueue` runtime on the Celery app's broker.
+
+    Extra keyword arguments are forwarded to ``BabelQueue`` (``queue``,
+    ``max_attempts``, ``dead_letter``, ``on_unknown_urn``, ...).
+    """
+    return BabelQueue(broker_url(celery_app), **kwargs)
+
+
+def install_worker(
+    celery_app: Any,
+    babel_app: Optional[BabelQueue] = None,
+    *,
+    queue: Optional[str] = None,
+    **kwargs: Any,
+) -> type:
+    """Register a Celery worker bootstep that consumes BabelQueue messages.
+
+    When a ``celery worker`` boots, the step starts a daemon thread running the
+    BabelQueue consumer loop (URN routing, retry → dead-letter). If ``babel_app``
+    is omitted it is built with :func:`from_celery`. Returns the bootstep class.
+    """
+    from celery import bootsteps  # lazy: only needed for this integration
+
+    app = babel_app if babel_app is not None else from_celery(celery_app, **kwargs)
+
+    class BabelQueueConsumerStep(bootsteps.StartStopStep):
+        """Runs the BabelQueue consumer loop alongside Celery's own consumer."""
+
+        def __init__(self, parent: Any, **options: Any) -> None:
+            super().__init__(parent, **options)
+            self._thread: Optional[threading.Thread] = None
+            self._stop = threading.Event()
+
+        def start(self, parent: Any) -> None:
+            def loop() -> None:
+                while not self._stop.is_set():
+                    app.consume(queue, max_messages=1, timeout=1.0)
+
+            self._thread = threading.Thread(
+                target=loop, name="babelqueue-consumer", daemon=True
+            )
+            self._thread.start()
+
+        def stop(self, parent: Any) -> None:
+            self._stop.set()
+
+    celery_app.steps["worker"].add(BabelQueueConsumerStep)
+    return BabelQueueConsumerStep

src/babelqueue/django/__init__.py

@@ -0,0 +1,72 @@
+"""Django integration. Requires the ``django`` extra:
+
+    pip install "babelqueue[django]"
+
+Add ``"babelqueue.django"`` to ``INSTALLED_APPS`` and configure a ``BABELQUEUE``
+settings dict::
+
+    BABELQUEUE = {
+        "broker_url": "redis://localhost:6379/0",
+        "queue": "orders",
+        "max_attempts": 3,
+        "dead_letter": True,
+    }
+
+Then publish from views/signals with :func:`publish`, register handlers on
+:func:`get_app`, and run the consumer with ``python manage.py babelqueue_worker``.
+The runtime is the shared :class:`~babelqueue.BabelQueue`, so messages interoperate
+with the PHP/Laravel, Go, Node, ... SDKs. ``django`` is imported lazily.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Mapping, Optional
+
+from ..app import BabelQueue
+
+# Keys (besides broker_url) forwarded to the BabelQueue constructor.
+_APP_KWARGS = frozenset(
+    {
+        "queue",
+        "on_unknown_urn",
+        "max_attempts",
+        "dead_letter",
+        "dead_letter_queue",
+        "dead_letter_suffix",
+        "transport",
+    }
+)
+
+_app: Optional[BabelQueue] = None
+
+
+def _build() -> BabelQueue:
+    from django.conf import settings  # lazy
+
+    raw: Mapping[str, Any] = getattr(settings, "BABELQUEUE", {}) or {}
+    kwargs: Dict[str, Any] = {k: v for k, v in raw.items() if k in _APP_KWARGS}
+    broker = raw.get("broker_url", "memory://")
+    return BabelQueue(broker, **kwargs)
+
+
+def get_app() -> BabelQueue:
+    """Return the process-wide :class:`~babelqueue.BabelQueue`, built from
+    ``settings.BABELQUEUE`` on first use."""
+    global _app
+    if _app is None:
+        _app = _build()
+    return _app
+
+
+def publish(urn: str, data: Mapping[str, Any], **kwargs: Any) -> str:
+    """Publish a message through the configured app; returns its id (``meta.id``)."""
+    return get_app().publish(urn, dict(data), **kwargs)
+
+
+def reset() -> None:
+    """Drop the cached app so the next :func:`get_app` rebuilds it (tests / settings reload)."""
+    global _app
+    _app = None
+
+
+__all__ = ["get_app", "publish", "reset"]

src/babelqueue/django/apps.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from django.apps import AppConfig
+
+
+class BabelQueueConfig(AppConfig):
+    """Django app config for the BabelQueue adapter."""
+
+    name = "babelqueue.django"
+    label = "babelqueue"
+    verbose_name = "BabelQueue"