django-channels-spectacular 0.1.0

AsyncAPI 3.0 documentation generator for Django Channels WebSocket consumers

django-channels-spectacular

AsyncAPI 3.0 documentation generator for Django Channels WebSocket consumers.

Think of it as drf-spectacular for Django Channels: annotate your consumer's action handlers once and the package generates and serves the spec automatically. No more hand-maintaining YAML that slowly drifts out of sync with your actual implementation.

---

Features

• Decorator-based annotation: @document_action / @document_event on consumer methods
• Multi-consumer specs: merge several consumers into one spec or serve them separately with a built-in switcher dropdown
• Hand-written YAML support: render existing AsyncAPI templates via manage.py export_asyncapi --template
• Interactive try-it-out panel: connect, send, and observe messages directly in the docs browser
• DRF / Pydantic / dataclass payload introspection: no hand-written schemas needed
• AsyncAPI 3.0: compliant spec rendered via the official @asyncapi/react-component

---

Installation

pip install django-channels-spectacular
# or
uv add django-channels-spectacular

With DRF serializer support:

pip install "django-channels-spectacular[drf]"
uv add "django-channels-spectacular[drf]"

Add to INSTALLED_APPS:

INSTALLED_APPS = [
    ...
    "channels_spectacular",
]

---

Quick start

1. Annotate your consumer

# myapp/consumers.py
from channels_spectacular import document_action, document_event

class DispatchConsumer(AsyncJsonWebsocketConsumer):

    @document_action(
        summary="Request a ride",
        payload=RequestRideSerializer,   # DRF, Pydantic, dataclass, or dict
        responses={"ride.requested": {"ride_id": "uuid"}},
        tags=["rides"],
        examples=[
            {
                "name": "Cash payment",
                "summary": "Rider pays with cash",
                "payload": {
                    "action": "request_ride",
                    "pickup_lat": 6.5244,
                    "pickup_lng": 3.3792,
                    "fare": "1500.00",
                    "payment_method": "cash",
                },
            },
        ],
    )
    async def handle_request_ride(self, content): ...

    @document_action(summary="Accept an offer", payload=AcceptOfferSerializer)
    async def handle_accept_offer(self, content): ...

    @document_event(
        "ride.offer",
        summary="Offer pushed to driver",
        payload=RideOfferSerializer,
        examples=[
            {
                "name": "Standard offer",
                "payload": {
                    "type": "ride.offer",
                    "ride_id": "d290f1ee-6c54-4b01-90e6-d701748f0851",
                    "fare": "1500.00",
                    "driver_name": "Emeka",
                },
            },
        ],
    )
    async def ride_offer(self, event): ...

    @document_event("ride.accepted", summary="Ride accepted by driver")
    async def ride_accepted(self, event): ...

Action name inference — @document_action strips the handle_ prefix automatically: handle_request_ride → "request_ride". Pass action= to override.

Event type is always explicit on @document_event because the method name (ride_offer) doesn't encode the full dotted type ("ride.offer").

2. Wire up the views

# myapp/urls.py
from django.urls import path
from channels_spectacular.views import AsyncAPIDocView, AsyncAPISpecView
from myapp.consumers import DispatchConsumer

urlpatterns = [
    path("ws-docs/", AsyncAPIDocView.as_view()),
    path(
        "ws-docs/asyncapi.yaml",
        AsyncAPISpecView.as_view(consumer=DispatchConsumer),
    ),
]

Visit /ws-docs/ for the interactive HTML viewer. Visit /ws-docs/asyncapi.yaml for the raw spec.

---

Full example — Dispatch API

A realistic consumer showing all decorator features and all three payload formats (dataclass, DRF serializer, Pydantic model):

# dispatch/consumers.py
from dataclasses import dataclass
from decimal import Decimal
from typing import Optional
from uuid import UUID

from channels.generic.websocket import AsyncJsonWebsocketConsumer
from channels_spectacular import document_action, document_event

# --------------------------------------------------------------------------
# Payload option 1 — Python dataclass (no extra dependency)
# --------------------------------------------------------------------------
@dataclass
class RequestRidePayload:
    pickup_address: str
    fare: Decimal
    passenger_count: int
    note: Optional[str] = None          # Optional[T] → not required in schema


# --------------------------------------------------------------------------
# Payload option 2 — DRF Serializer (pip install djangorestframework)
# --------------------------------------------------------------------------
from rest_framework import serializers

class AcceptOfferSerializer(serializers.Serializer):
    ride_id   = serializers.UUIDField()
    driver_id = serializers.UUIDField()


# --------------------------------------------------------------------------
# Payload option 3 — Pydantic model (pip install pydantic)
# --------------------------------------------------------------------------
from pydantic import BaseModel

class RideOfferPayload(BaseModel):
    ride_id:     UUID
    driver_name: str
    eta_minutes: int


# --------------------------------------------------------------------------
# Consumer
# --------------------------------------------------------------------------
class DispatchConsumer(AsyncJsonWebsocketConsumer):

    async def connect(self):
        if not self.scope["user"].is_authenticated:
            await self.close(code=4401)
            return
        await self.accept()

    # ---- client → server actions ----------------------------------------

    @document_action(
        summary="Request a ride",
        description="Rider sends pickup details to start dispatch.",
        payload=RequestRidePayload,       # ← dataclass
        responses={"ride.requested": {"ride_id": "uuid"}},
        tags=["rides"],
        examples=[
            {
                "name": "Cash ride",
                "summary": "Standard cash pickup",
                "payload": {
                    "action": "request_ride",
                    "pickup_address": "23 Marina Rd, Lagos",
                    "fare": "1500.00",
                    "passenger_count": 1,
                },
            }
        ],
    )
    async def handle_request_ride(self, content):
        ...

    @document_action(
        summary="Accept a driver offer",
        payload=AcceptOfferSerializer,    # ← DRF serializer
        tags=["rides"],
    )
    async def handle_accept_offer(self, content):
        ...

    @document_action(
        action="ping",
        summary="Health check — expects a pong event in return",
        # payload=None → discriminator-only schema (no extra fields)
    )
    async def handle_ping(self, content):
        ...

    @document_action(summary="Cancel an active ride request", deprecated=True)
    async def handle_cancel_ride(self, content):
        ...

    # ---- server → client events -----------------------------------------

    @document_event(
        "ride.offer",
        summary="Server pushes a driver offer to the rider",
        payload=RideOfferPayload,         # ← Pydantic model
        tags=["rides"],
        examples=[
            {
                "name": "Standard offer",
                "payload": {
                    "type": "ride.offer",
                    "ride_id": "d290f1ee-6c54-4b01-90e6-d701748f0851",
                    "driver_name": "Emeka",
                    "eta_minutes": 4,
                },
            }
        ],
    )
    async def ride_offer(self, event):
        ...

    @document_event(
        "ride.accepted",
        summary="Ride accepted by a driver",
        payload={"type": "object", "properties": {"driver_name": {"type": "string"}}},
    )
    async def ride_accepted(self, event):
        ...

All four payload formats produce a JSON Schema fragment in the spec; the table in Payload formats shows the mapping in detail.

---

Authentication

WebSocket handshakes are HTTP upgrade requests, so Django's normal auth mechanisms apply — they just need to run in the ASGI middleware layer.

Cookie / session auth (browser clients)

The browser sends the sessionid cookie automatically on same-origin connections. Wrap your router with AuthMiddlewareStack:

# myproject/asgi.py
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from dispatch.routing import websocket_urlpatterns

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(websocket_urlpatterns)
    ),
})

AuthMiddlewareStack reads the sessionid cookie from the handshake headers and populates scope["user"].

Query parameter auth (API / mobile clients)

Apps that can't set cookies attach a short-lived JWT as a URL query param: wss://api.example.com/ws/?token=<jwt>. Write a small ASGI middleware that reads it from scope["query_string"]:

# dispatch/middleware.py
from urllib.parse import parse_qs
from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
import jwt

@database_sync_to_async
def get_user_from_token(token):
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        from django.contrib.auth import get_user_model
        return get_user_model().objects.get(pk=payload["user_id"])
    except Exception:
        return AnonymousUser()

class QueryTokenAuthMiddleware:
    def __init__(self, inner):
        self.inner = inner

    async def __call__(self, scope, receive, send):
        if scope["type"] == "websocket":
            qs = parse_qs(scope["query_string"].decode())
            token = qs.get("token", [None])[0]
            scope["user"] = (
                await get_user_from_token(token) if token else AnonymousUser()
            )
        return await self.inner(scope, receive, send)

Documenting auth in the spec

Add these settings and the generator inserts securitySchemes into the spec:

CHANNELS_SPECTACULAR_SETTINGS = {
    "AUTH_QUERY_PARAM": "token",      # adds httpApiKey query scheme
    "AUTH_COOKIE_NAME": "sessionid",  # adds httpApiKey cookie scheme
}

Try-it-out panel

The interactive viewer's auth selector (TRY_IT_OUT_ENABLED = True) supports both schemes:

• Query param — paste a token and click Apply; it appends ?token=<jwt> to the WebSocket URL before connecting.
• Session cookie (automatic) — the browser sends sessionid automatically; just confirm you are logged in on the same origin.

---

3. Configure (optional)

# settings.py
CHANNELS_SPECTACULAR_SETTINGS = {
    "TITLE": "Dispatch API",
    "VERSION": "1.0.0",
    "DESCRIPTION": "Real-time ride and delivery dispatch.",
    "CHANNEL_PATH": "/ws/dispatch/",
    # Static servers block — omit to derive from the request.
    "SERVERS": {
        "production": {"host": "api.example.com", "protocol": "wss"},
    },
    # Fallbacks when SERVERS is None:
    "WS_HOST": None,           # defaults to request.get_host()
    "WS_PROTOCOL": None,       # defaults to "wss" if HTTPS else "ws"
    # Enable the interactive try-it-out panel (dev only):
    "TRY_IT_OUT_ENABLED": False,
    "ASYNCAPI_VERSION": "3.0.0",
}

---

Multi-consumer support

Multiple consumers in one spec

Merge several consumers into a single spec by passing a consumers list to AsyncAPISpecView. The generator prefixes operations with the consumer's channel name to keep them unique.

urlpatterns = [
    path("ws-docs/", AsyncAPIDocView.as_view()),
    path(
        "ws-docs/asyncapi.yaml",
        AsyncAPISpecView.as_view(
            consumers=[
                (RideConsumer,  "/ws/rides/"),
                (NotifConsumer, "/ws/notifications/"),
            ],
        ),
    ),
]

Separate specs with a switcher dropdown

Serve each consumer under its own URL and let users switch between them with a built-in dropdown in the docs viewer:

urlpatterns = [
    path(
        "ws-docs/",
        AsyncAPIDocView.as_view(
            specs=[
                ("Dispatch  /ws/dispatch/",      "/api/v1/ws-docs/dispatch/asyncapi.yaml"),
                ("Notifications  /ws/notif/",    "/api/v1/ws-docs/notif/asyncapi.yaml"),
            ],
        ),
    ),
    path(
        "ws-docs/dispatch/asyncapi.yaml",
        AsyncAPISpecView.as_view(consumer=DispatchConsumer, channel_path="/ws/dispatch/"),
    ),
    path(
        "ws-docs/notif/asyncapi.yaml",
        AsyncAPISpecView.as_view(consumer=NotifConsumer, channel_path="/ws/notifications/"),
    ),
]

Switching consumers re-renders the full AsyncAPI React component and auto-fills the try-it-out panel's WebSocket URL by reading the selected spec's servers block — no manual URL editing needed.

---

Management command: export_asyncapi

Export the spec to a YAML file for SDK generation, CI artefacts, or committing alongside your API contracts.

Generator mode — from annotated consumers

# Single consumer
python manage.py export_asyncapi \
    --consumer myapp.consumers.DispatchConsumer:/ws/dispatch/ \
    --output docs/asyncapi.yaml

# Multiple consumers in one file
python manage.py export_asyncapi \
    --consumer myapp.consumers.RideConsumer:/ws/rides/ \
    --consumer myapp.consumers.NotifConsumer:/ws/notifications/ \
    --output docs/asyncapi.yaml

Template mode — from a hand-written YAML

For projects that maintain a hand-written AsyncAPI YAML (or a Django template with {{ WS_HOST }} / {{ WS_PROTOCOL }} placeholders), use --template:

python manage.py export_asyncapi \
    --template rides/templates/rides/asyncapi.yaml \
    --output docs/asyncapi.yaml \
    --host localhost:8000 \
    --protocol ws

The command:

1. Substitutes {{ WS_HOST }} / {{ WS_PROTOCOL }} (both Django and uppercase variants)
2. Injects title into payload schemas that lack one, so SDK generators (@asyncapi/generator, Modelina) produce readable type names instead of AnonymousSchema_N

Full reference

usage: manage.py export_asyncapi [--consumer DOTTED.PATH[:/ws/path/] | --template FILE]
                                  [--output FILE] [--host HOST] [--protocol {ws,wss}]

Flag	Default	Description
--consumer	—	Dotted import path, optionally :channel_path. Repeatable.
--template	—	Path to a hand-written AsyncAPI YAML template.
-o / --output	asyncapi.yaml	Destination file. Parent dirs created automatically.
--host	settings / localhost:8000	WS host for the servers block.
--protocol	settings / ws	ws or wss.

---

Payload formats

document_action and document_event accept four payload types:

Type	How it's converted
None	No payload schema
dict	Returned as-is (raw JSON Schema fragment)
@dataclass class	Introspects type annotations recursively
DRF Serializer subclass	Introspects get_fields()
Pydantic BaseModel subclass	model_json_schema() (v2) or schema() (v1)

Examples

Both @document_action and @document_event accept an examples list. Each entry is a dict with optional name / summary strings and a payload dict of concrete values. Examples are emitted verbatim into the AsyncAPI spec and rendered by the viewer alongside the schema.

@document_action(
    summary="Request a ride",
    payload=RequestRideSerializer,
    examples=[
        {
            "name": "Cash payment",
            "summary": "Rider pays with cash at destination",
            "payload": {
                "action": "request_ride",
                "pickup_lat": 6.5244,
                "pickup_lng": 3.3792,
                "fare": "1500.00",
                "payment_method": "cash",
            },
        },
        {
            "name": "Card payment",
            "payload": {
                "action": "request_ride",
                "pickup_lat": 6.5244,
                "pickup_lng": 3.3792,
                "fare": "1800.00",
                "payment_method": "card",
            },
        },
    ],
)
async def handle_request_ride(self, content): ...

@document_event(
    "ride.offer",
    summary="Offer pushed to driver",
    payload=RideOfferSerializer,
    examples=[
        {
            "name": "Standard offer",
            "payload": {
                "type": "ride.offer",
                "ride_id": "d290f1ee-6c54-4b01-90e6-d701748f0851",
                "fare": "1500.00",
                "driver_name": "Emeka",
                "expires_at": 1717000015.0,
            },
        },
    ],
)
async def ride_offer(self, event): ...

A discriminator const property is always injected:

• Send messages get "action": {"type": "string", "const": "<action>"}.
• Receive messages get "type": {"type": "string", "const": "<event_type>"}.

Python / dataclass → JSON Schema

Python type	JSON Schema
str	{"type": "string"}
int	{"type": "integer"}
float	{"type": "number"}
bool	{"type": "boolean"}
Decimal	{"type": "string", "format": "decimal"}
UUID	{"type": "string", "format": "uuid"}
list[T]	{"type": "array", "items": <T-schema>}
T | None / Optional[T]	<T-schema> (not in required)
Nested @dataclass	Recursive {"type": "object", ...}

DRF field → JSON Schema

CharField, UUIDField, IntegerField, FloatField, DecimalField, BooleanField, DateField, DateTimeField, ChoiceField (with enum), ListField, nested Serializer, and more are all mapped automatically.

---

Using the generator directly

from channels_spectacular import AsyncAPIGenerator
from myapp.consumers import DispatchConsumer

# Single consumer
generator = AsyncAPIGenerator(
    DispatchConsumer,
    info={"title": "Dispatch API", "version": "1.0.0"},
    servers={"prod": {"host": "api.example.com", "protocol": "wss"}},
    channel_path="/ws/dispatch/",
)

# Multiple consumers
generator = AsyncAPIGenerator(
    consumers=[
        (RideConsumer,  "/ws/rides/"),
        (NotifConsumer, "/ws/notifications/"),
    ],
    info={"title": "All Consumers", "version": "1.0.0"},
    servers={"prod": {"host": "api.example.com", "protocol": "wss"}},
)

spec_dict = generator.get_spec()   # Python dict
spec_yaml = generator.get_yaml()   # YAML string

---

Inheritance

Annotated methods are discovered by walking __mro__, so subclasses can extend a base consumer's documented actions:

class BaseConsumer(AsyncJsonWebsocketConsumer):
    @document_action(summary="Ping")
    async def handle_ping(self, content): ...

class ExtendedConsumer(BaseConsumer):
    @document_action(summary="Override ping with more detail")
    async def handle_ping(self, content): ...  # overrides parent

    @document_action(summary="Extra action")
    async def handle_extra(self, content): ...

---

Other ways to generate AsyncAPI docs

This package is not the only path to an AsyncAPI 3.0 spec. Choose based on how your codebase is structured:

Approach	When to use
@document_action / @document_event (this package)	New projects where consumers are the source of truth
--template + export_asyncapi (this package)	Projects with an existing hand-written YAML — get host injection and title auto-fixing for free
AsyncAPI CLI directly	CI validation (asyncapi validate), linting, or code generation (asyncapi generate) without Django integration
asyncapi-python	Pure-Python programmatic spec building; no Django integration
Hand-written YAML + static AsyncAPIDocView	Small teams that find annotations overkill; point AsyncAPIDocView(spec_url=...) at a static YAML file
Postman / Insomnia export	If WS flows are already in Postman Collections v2.1 — export and convert with asyncapi convert
spectral	Linting an existing spec against the AsyncAPI ruleset; combines with any generation approach

---

PyPI publishing

The package uses Hatch as the build backend.

One-time setup

pip install hatch twine

Create a PyPI API token at https://pypi.org/manage/account/token/ and store it in ~/.pypirc:

[distutils]
index-servers = pypi

[pypi]
username = __token__
password = pypi-<your-token-here>

Or export it for the session:

export TWINE_USERNAME=__token__
export TWINE_PASSWORD=pypi-<your-token-here>

Build and publish

# 1. Bump version in pyproject.toml and channels_spectacular/__init__.py
# 2. Build
hatch build                   # → dist/*.tar.gz and dist/*.whl
twine check dist/*            # validate before uploading
twine upload dist/*           # publish to PyPI

Test against TestPyPI first:

twine upload --repository testpypi dist/*
pip install --index-url https://test.pypi.org/simple/ django-channels-spectacular

Automated publishing via GitHub Actions

Create .github/workflows/publish.yml:

name: Publish to PyPI

on:
  push:
    tags:
      - "v*"

jobs:
  publish:
    runs-on: ubuntu-latest
    environment: pypi
    permissions:
      id-token: write   # OIDC trusted publishing — no stored API token needed
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install hatch
      - run: hatch build
      - uses: pypa/gh-action-pypi-publish@release/v1

Enable Trusted Publishing on PyPI (project Settings → Publishing) to skip storing an API token in GitHub Secrets.

---

ReadTheDocs

The docs/ directory contains a Sphinx project ready to publish on ReadTheDocs using the Furo theme.

Local preview

cd docs
pip install -r requirements.txt
make html
open _build/html/index.html    # macOS; use xdg-open on Linux

ReadTheDocs setup

1. Import the GitHub repo at https://readthedocs.org/dashboard/import/
2. .readthedocs.yaml at the repo root configures the build automatically — no extra configuration in the RTD dashboard is needed.
3. Set the default branch to main.

Documentation rebuilds on every push to main and on every v* tag.

---

Running the tests

cd django-channels-spectacular
pip install -e ".[dev]"
pytest -v

---

Contributing

Contributions are welcome — bug reports, feature requests, and pull requests alike.

Setting up a development environment

git clone https://github.com/ibukun-brain/django-channels-spectacular.git
cd django-channels-spectacular

pip install -e ".[dev]"
# or
uv sync --extra dev

Submitting a pull request

1. Fork the repository and create a feature branch from main.
2. Make your changes, add tests, and confirm the suite passes.
3. Open a pull request against main — one feature or fix per PR.

Reporting bugs

Open a GitHub issue with the Python/Django/Channels versions, a minimal reproduction, and the full traceback.

---

License

BSD 3-Clause