cypartagraphqlsubscriptionstools 4.1.7

Graphene + Django GraphQL subscriptions over Django Channels (async WebSockets, bounded outbox, multi-operation registry).

CypartaGraphqlSubscriptionsTools

Version 4.1.7

Graphene + Django GraphQL subscriptions over Django Channels (async WebSockets). The package ships a production-oriented consumer that speaks:

• graphql-transport-ws — negotiate with WebSocket subprotocol graphql-transport-ws (recommended).
• Legacy graphql-ws — subprotocol graphql-ws (Apollo-style start / stop / data).

The server selects the protocol from the Sec-WebSocket-Protocol header. Your GraphQL schema, models, and routing live in your Django project — not inside this library.

Architecture overview

---

Features

• WebSocket GraphQL subscriptions — JSON messages, connection_init gating, per-operation id, GraphQL errors over the wire.
• Per-connection bounded outbox — asyncio.Queue plus one sender task so slow clients cannot grow an unbounded send backlog (CYPARTA_WS_OUTBOX_MAXSIZE).
• Multi-operation support — each subscribe / start operation has its own id; channel groups map to operation ids (_ops / _group_ops).
• Optional group permission class — dotted path to a class with has_permission(...) for per-group authorization (CYPARTA_WS_GROUP_PERMISSION_CLASS).
• Group name validation — names validated like Django Channels BaseChannelLayer.valid_group_name (validate_group_name, CYPARTA_WS_STRICT_GROUP_NAMES).
• Optional model lifecycle mixin — CypartaSubscriptionModelMixin schedules trigger_subscription on transaction.on_commit (create / update / delete hooks via django_lifecycle).
• Custom event serializer — optional CYPARTA_WS_EVENT_SERIALIZER callable for server-side trigger_subscription payloads; import path cached until reset_event_serializer_cache().
• Outbox overflow strategy — when the queue is full: drop_newest (default), drop_oldest, or close_connection (WebSocket close code 4413).

---

Installation

From PyPI:

pip install cypartagraphqlsubscriptionstools

From a git checkout (editable):

pip install -e .

Install test dependencies:

pip install -e ".[test]"

---

Quick start

1. Register the app

# settings.py
INSTALLED_APPS = [
    # ...
    "channels",
    "CypartaGraphqlSubscriptionsTools",
    "articles",  # your app
]

2. Channel layers

In-memory (development / tests):

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer",
    },
}

Redis (typical production):

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

Install channels-redis when using the Redis backend.

3. Graphene schema

Point Django Graphene at your schema object (define Query, Subscription, and schema in your project — for example articles.schema):

# settings.py
GRAPHENE = {
    "SCHEMA": "articles.schema.schema",
}

4. ASGI: HTTP + WebSocket consumer

Mount CypartaGraphqlSubscriptionsConsumer on a URL your clients use (here /ws/graphql/).

# articles/routing.py
from django.urls import re_path

from CypartaGraphqlSubscriptionsTools.consumers import CypartaGraphqlSubscriptionsConsumer

websocket_urlpatterns = [
    re_path(r"^ws/graphql/$", CypartaGraphqlSubscriptionsConsumer.as_asgi()),
]

# asgi.py
import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from articles.routing import websocket_urlpatterns

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

django_asgi_app = get_asgi_application()

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

Set ASGI_APPLICATION = "myproject.asgi.application" in settings.py.

To authenticate browsers or API clients with a DRF token on WebSockets (?token=… or Authorization: Token …), wrap the WebSocket stack with TokenAuthMiddleware as shown in WebSocket token authentication (DRF authtoken).

---

Recommended production settings

Use stricter defaults in production unless you have a deliberate reason not to:

# settings.py

CYPARTA_WS_REQUIRE_AUTH = True
CYPARTA_WS_STRICT_GROUP_NAMES = True
CYPARTA_WS_DROP_EVENT_ON_SERIALIZATION_ERROR = True

CYPARTA_WS_GROUP_PERMISSION_CLASS = "articles.permissions.SubscriptionGroupPermission"

# Dashboards / live metrics: prefer latest snapshot when client lags
CYPARTA_WS_OUTBOX_OVERFLOW_STRATEGY = "drop_oldest"

# Chat / notifications: keep older queued messages; drop only newest on overflow (default)
# CYPARTA_WS_OUTBOX_OVERFLOW_STRATEGY = "drop_newest"

# Shed pathological slow clients (disconnect with code 4413 when outbox stays full)
# CYPARTA_WS_OUTBOX_OVERFLOW_STRATEGY = "close_connection"

Settings reference

Setting	Role
CYPARTA_WS_OUTBOX_MAXSIZE	Max queued outbound ExecutionResult frames per WebSocket (default 256). When full, overflow policy applies.
CYPARTA_WS_OUTBOX_OVERFLOW_STRATEGY	drop_newest (default): refuse enqueue of the newest event. drop_oldest: remove oldest queued item then enqueue. close_connection: schedule one close per socket with code 4413.
CYPARTA_WS_REQUIRE_AUTH	If True (default), anonymous or missing scope["user"] cannot join subscription groups.
CYPARTA_WS_GROUP_PERMISSION_CLASS	Optional dotted path to a permission class (instantiated once per connection) implementing has_permission(self, user, group_name, operation_id=None, scope=None, variables=None) (sync or async).
CYPARTA_WS_STRICT_GROUP_NAMES	If True (default), invalid group strings raise a generic GraphQL error. If False, unsafe characters are normalized to _ where possible.
CYPARTA_WS_EVENT_SERIALIZER	Optional dotted path: (value, group=None, scope=None) returning the wire payload (sync, async, or awaitable).
CYPARTA_WS_DROP_EVENT_ON_SERIALIZATION_ERROR	If True, failed custom + default serialization skips group_send for that event (logged).
CYPARTA_WS_RAISE_ON_INVALID_TRIGGER_GROUP	If True, trigger_subscription raises GroupNameInvalid on bad names instead of skipping.

Outbox strategy hints

• Dashboards / live metrics — often use drop_oldest so the socket keeps newer updates when the client falls behind.
• Chat / notifications — often keep drop_newest so older queued messages are not discarded silently.
• close_connection — disconnect slow clients (code 4413) so they reconnect rather than staying in a bad state.

---

Complete example: Articles app

Below, articles is your Django app. Adjust imports and model fields to your project.

Model (articles/models.py)

from django.conf import settings
from django.db import models
from django_lifecycle import LifecycleModelMixin

from CypartaGraphqlSubscriptionsTools.mixins import CypartaSubscriptionModelMixin


class Article(LifecycleModelMixin, CypartaSubscriptionModelMixin, models.Model):
    title = models.CharField(max_length=255)
    body = models.TextField(blank=True)
    is_public = models.BooleanField(default=False)
    owner = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name="articles",
    )

    def get_subscription_group_names(self, action):
        if action == "create":
            return ["ArticleCreated"]
        if action == "update":
            return [f"ArticleUpdated.{self.pk}"]
        if action == "delete":
            return [f"ArticleDeleted.{self.pk}"]
        return []

    def get_subscription_payload(self, action):
        return self

GraphQL types and subscriptions (articles/schema.py)

import graphene
from asgiref.sync import async_to_sync
from graphene_django.types import DjangoObjectType

from .models import Article


class ArticleType(DjangoObjectType):
    class Meta:
        model = Article


class ArticleSubscription(graphene.ObjectType):
    """Subscription root fields; ``root`` in resolvers is the WebSocket consumer."""

    article_created = graphene.Field(ArticleType, subscribe=graphene.Boolean())

    def resolve_article_created(root, info, subscribe=True):
        node = info.field_nodes[0]
        selections = node.selection_set.selections if node.selection_set else ()
        requested_fields = [
            sel.name.value for sel in selections if hasattr(sel, "name")
        ]
        return async_to_sync(root.detect_register_group_status)(
            ["ArticleCreated"],
            requested_fields=requested_fields,
            variables=info.variable_values,
            subscribe=subscribe,
        )

    article_updated = graphene.Field(
        ArticleType,
        id=graphene.String(required=True),
        subscribe=graphene.Boolean(),
    )

    def resolve_article_updated(root, info, id, subscribe=True):
        node = info.field_nodes[0]
        selections = node.selection_set.selections if node.selection_set else ()
        requested_fields = [
            sel.name.value for sel in selections if hasattr(sel, "name")
        ]
        return async_to_sync(root.detect_register_group_status)(
            [f"ArticleUpdated.{id}"],
            requested_fields=requested_fields,
            variables=info.variable_values,
            subscribe=subscribe,
        )

    article_deleted = graphene.Field(
        ArticleType,
        id=graphene.String(required=True),
        subscribe=graphene.Boolean(),
    )

    def resolve_article_deleted(root, info, id, subscribe=True):
        node = info.field_nodes[0]
        selections = node.selection_set.selections if node.selection_set else ()
        requested_fields = [
            sel.name.value for sel in selections if hasattr(sel, "name")
        ]
        return async_to_sync(root.detect_register_group_status)(
            [f"ArticleDeleted.{id}"],
            requested_fields=requested_fields,
            variables=info.variable_values,
            subscribe=subscribe,
        )


class Query(graphene.ObjectType):
    hello = graphene.String()

    def resolve_hello(self, info):
        return "world"


class Subscription(ArticleSubscription, graphene.ObjectType):
    pass


schema = graphene.Schema(query=Query, subscription=Subscription)

Note: Prefer the subscribe keyword (GraphQL argument and/or detect_register_group_status(..., subscribe=...)). The legacy positional subscripe flag is still accepted on detect_register_group_status, register_group, and un_register_group for backward compatibility; if you use subscripe without subscribe, extensions.cyparta may include deprecationNotes.

---

Permission class example

has_permission receives the Django user from scope["user"], the validated channel group name, the GraphQL operation id, the ASGI scope, and subscription variables (when provided).

# articles/permissions.py
from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser

from .models import Article


class SubscriptionGroupPermission:
    async def has_permission(self, user, group_name, operation_id=None, scope=None, variables=None):
        if user is None or isinstance(user, AnonymousUser) or getattr(user, "is_anonymous", True):
            return False

        if group_name == "ArticleCreated":
            return True

        if group_name.startswith("ArticleUpdated.") or group_name.startswith("ArticleDeleted."):
            _prefix, pk_str = group_name.split(".", 1)  # e.g. ArticleUpdated.7 -> pk_str == "7"
            try:
                pk = int(pk_str)
            except ValueError:
                return False

            article = await database_sync_to_async(
                lambda: Article.objects.filter(pk=pk).first()
            )()
            if article is None:
                return False
            if article.is_public:
                return True
            return article.owner_id == getattr(user, "pk", None)

        return False

# settings.py
CYPARTA_WS_GROUP_PERMISSION_CLASS = "articles.permissions.SubscriptionGroupPermission"

---

Custom event serializer example

Normalize or redact payloads before group_send. Signature: value, optional group, optional scope (sync or async).

# articles/ws_serializers.py
def serialize_subscription_event(value, group=None, scope=None):
    if hasattr(value, "_meta") and hasattr(value, "pk"):
        return {"kind": "article", "id": value.pk, "title": getattr(value, "title", "")}
    if isinstance(value, dict):
        return value
    return str(value)

# settings.py
CYPARTA_WS_EVENT_SERIALIZER = "articles.ws_serializers.serialize_subscription_event"
CYPARTA_WS_DROP_EVENT_ON_SERIALIZATION_ERROR = True

The serializer import is cached by dotted path. After changing CYPARTA_WS_EVENT_SERIALIZER in tests or at runtime, call:

from CypartaGraphqlSubscriptionsTools import events

events.reset_event_serializer_cache()

---

Manual publishing

From synchronous code (views, signals, management commands), publish to every socket in a channel group:

from asgiref.sync import async_to_sync

from CypartaGraphqlSubscriptionsTools.events import trigger_subscription
from .models import Article


def notify_article_updated(article: Article):
    async_to_sync(trigger_subscription)(f"ArticleUpdated.{article.pk}", article)

trigger_subscription validates the group name, serializes value (custom serializer if configured), then group_send. Use validate_group_name if you need to pre-validate strings in your own code.

---

Client examples

Subprotocol must be requested by the client (Sec-WebSocket-Protocol). All frames are JSON objects.

graphql-transport-ws

1. connection_init — must be sent first; server replies connection_ack.
2. subscribe — includes id and payload (query, optional variables, optional operationName).
3. Registration ack — first next frame often has data: null and extensions.cyparta (action, registeredGroups, subscribe, …).
4. Live event — next with payload.data shaped as { "<responseKey>": <serialized value> } (response key = field alias or field name).
5. complete — client stops the operation; server may also send complete with the same id.

Example sequence (illustrative payloads):

{"type": "connection_init"}

{"type": "connection_ack"}

{
  "id": "1",
  "type": "subscribe",
  "payload": {
    "query": "subscription { articleUpdated(id: \"5\") { id title } }",
    "variables": {}
  }
}

{
  "id": "1",
  "type": "next",
  "payload": {
    "data": null,
    "errors": null,
    "extensions": {
      "cyparta": {
        "action": "register",
        "registeredGroups": ["ArticleUpdated.5"],
        "subscribe": true,
        "subscripe": true
      }
    }
  }
}

{
  "id": "1",
  "type": "next",
  "payload": {
    "data": {
      "articleUpdated": {
        "pk": 5,
        "fields": {"id": 5, "title": "Hello", "body": "..."},
        "group": "ArticleUpdated.5"
      }
    },
    "errors": null
  }
}

{"id": "1", "type": "complete"}

Legacy graphql-ws

Same ordering rule: connection_init before start.

{"type": "connection_init"}

{"type": "connection_ack"}

{
  "id": "1",
  "type": "start",
  "payload": {
    "query": "subscription { articleUpdated(id: \"5\") { id title } }",
    "variables": {}
  }
}

{"id": "1", "type": "stop"}

{"type": "connection_terminate"}

Legacy protocol uses data instead of next for result frames; errors use a single error object in payload instead of an array.

---

Group names

Rules align with Django Channels BaseChannelLayer.valid_group_name:

• Type: non-empty string only at the API boundary.
• Length: strictly less than the Channels maximum name length.
• Characters: ASCII a-z, A-Z, 0-9, _, -, . only when CYPARTA_WS_STRICT_GROUP_NAMES is True (default).

Example	Valid?
ArticleCreated	Yes
ArticleUpdated.42	Yes
my-feed_v1	Yes
`` (empty)	No
Article Updated (space)	No (strict); normalized with underscores if strict is False
bad/slash	No (strict)

In Python:

from CypartaGraphqlSubscriptionsTools.utils import GroupNameInvalid, validate_group_name

try:
    safe = validate_group_name("ArticleUpdated.1")
except GroupNameInvalid as exc:
    # exc.client_message is safe for clients
    ...

---

Authentication

The consumer reads scope["user"] for CYPARTA_WS_REQUIRE_AUTH and for CYPARTA_WS_GROUP_PERMISSION_CLASS. You can rely on session/cookie auth only (AuthMiddlewareStack — see Quick start) or add DRF token auth for WebSockets with TokenAuthMiddleware below.

WebSocket token authentication (DRF authtoken)

The browser WebSocket API cannot set the Authorization header, so browser clients should pass the API token in the query string. Non-browser clients (wscat, Postman, services) can use Authorization: Token <key> on the WebSocket handshake.

CypartaGraphqlSubscriptionsTools.middleware.TokenAuthMiddleware:

• Resolves the token in this order: Authorization: Token <key> (wins if present and valid), then query token, auth, authToken, accessToken.
• Sets scope["user"], or AnonymousUser() if missing/invalid; malformed header/query is ignored safely.
• Uses rest_framework.authtoken.models.Token with select_related("user").

1) Install and enable DRF authtoken

pip install djangorestframework

# settings.py
INSTALLED_APPS = [
    # ...
    "rest_framework",
    "rest_framework.authtoken",
    # ...
]

python manage.py migrate   # creates authtoken_token

2) Wire ASGI: token middleware + session (optional) + WebSocket routes

Use TokenAuthMiddleware on the WebSocket branch only. HTTP stays on get_asgi_application() as in Quick start.

# asgi.py
import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from CypartaGraphqlSubscriptionsTools.middleware import TokenAuthMiddleware
from myproject.routing import websocket_urlpatterns  # your URLRouter patterns

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

django_asgi_app = get_asgi_application()

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

TokenAuthMiddleware runs first; AuthMiddlewareStack can still fill user from the session for the same connection if you use cookies. If both apply, you may need a custom order or a single source of truth for user in your app.

3) Create a token for a user

Django admin: Users → add Token for a user, or in a shell:

from django.contrib.auth import get_user_model
from rest_framework.authtoken.models import Token

user = get_user_model().objects.get(username="alice")
token, _ = Token.objects.get_or_create(user=user)
print(token.key)  # pass this to the client (over HTTPS / WSS only in production)

Or expose your own REST endpoint that returns Token.objects.get_or_create after username/password login.

4) Browser client (query string — recommended)

Build the WebSocket URL with the token. Request the GraphQL WebSocket subprotocol your server expects (graphql-transport-ws or graphql-ws).

const TOKEN = "..."; // from your login API — never commit secrets
const host = window.location.host;
const proto = window.location.protocol === "https:" ? "wss:" : "ws:";
const qs = new URLSearchParams({ token: TOKEN }).toString();

const ws = new WebSocket(
  `${proto}//${host}/ws/graphql/?${qs}`,
  "graphql-transport-ws"
);

ws.onopen = () => {
  ws.send(JSON.stringify({ type: "connection_init" }));
};

ws.onmessage = (ev) => console.log(JSON.parse(ev.data));

Equivalent query keys: token, auth, authToken, accessToken.

5) CLI / tools (Authorization header)

When the client can set handshake headers (Postman, custom scripts, some native stacks):

GET ws://127.0.0.1:8000/ws/graphql/ HTTP/1.1
Upgrade: websocket
Sec-WebSocket-Protocol: graphql-transport-ws
Authorization: Token abc123yourtokenhere

If both header and query include a token, the header takes precedence.

Example with wscat (if your build forwards headers — otherwise use the query-string URL in the connect URL):

# Query string (works everywhere)
wscat -c "ws://127.0.0.1:8000/ws/graphql/?token=YOUR_TOKEN" -s graphql-transport-ws

6) Production notes

• Prefer wss:// and HTTPS so tokens are not sent in clear text.
• Treat tokens like passwords: short TTL, revoke on logout, never log query strings with token= in access logs without redaction.

---

Payload filtering (requested_fields)

When trigger_subscription sends a dict whose fields value is itself a dict (the usual shape for serialized Model instances), the consumer can narrow fields to the keys the client selected on the subscription field.

• Pass requested_field names as a list of strings from the subscription resolver (see Articles example).
• If requested_fields is None or empty, or the payload has no dict fields, the payload is returned unchanged.
• Filtering never mutates the original dict; clients still receive top-level keys like pk unchanged when only fields is subset.

---

Troubleshooting

Symptom	What to check
Socket connects but subscription does nothing	connection_init must run before subscribe / start. Confirm GRAPHENE["SCHEMA"], subscription field names, and group_send group strings match register_group names exactly.
Close 1002 right after connect	Unsupported or missing Sec-WebSocket-Protocol. Client must request graphql-transport-ws or graphql-ws.
Close 4401	subscribe / start sent before connection_ack (init not completed).
Close 4429	Duplicate connection_init on the same WebSocket.
Close 4413	Outbox overflow with CYPARTA_WS_OUTBOX_OVERFLOW_STRATEGY = "close_connection"; client is too slow or CYPARTA_WS_OUTBOX_MAXSIZE is too small for your burst rate.
No event after trigger_subscription	No subscriber in that exact group name; channel_layer / Redis misconfiguration; serialization dropped the event (CYPARTA_WS_DROP_EVENT_ON_SERIALIZATION_ERROR); invalid group name skipped or raised per CYPARTA_WS_RAISE_ON_INVALID_TRIGGER_GROUP.

---

Upgrade notes

• v4.1.7 — README: expanded WebSocket token authentication — DRF setup, full asgi.py, creating tokens, browser JavaScript (graphql-transport-ws), headers/wscat, production notes; Quick start link to auth section.
• v4.1.6 — TokenAuthMiddleware: token from Authorization: Token … (priority) or query token / auth / authToken / accessToken; safe handling of malformed header/query; AnonymousUser when missing or invalid; select_related("user"). Requires djangorestframework + authtoken app for this middleware. Tests and [test] extra include djangorestframework.
• v4.1.5 — README overhaul (quick start, production settings, Articles example, permissions, serializers, clients, troubleshooting). Cover and architecture images; MANIFEST.in ships cover.jpg / graph.jpg with the sdist. PyPI-friendly image URLs in README.
• v4.1.4 — drop_oldest calls task_done() after discarding the oldest queue item (consistent unfinished count for join()). close_connection schedules at most one disconnect per socket. _safe_passthrough stringifies dict keys for JSON-safe payloads. README and test extras refined (pytest-django for DB tests).
• v4.1.3 — Cached CYPARTA_WS_EVENT_SERIALIZER import; CYPARTA_WS_DROP_EVENT_ON_SERIALIZATION_ERROR; get_subscription_payload(action) on the mixin; per-group try/except around publishes; CYPARTA_WS_OUTBOX_OVERFLOW_STRATEGY.
• v4.1.2 — Prefer subscribe= without positional subscripe; mixin publishes on_commit; after_delete for deletes; optional event serializer and CYPARTA_WS_RAISE_ON_INVALID_TRIGGER_GROUP.
• v4.1.1 — validate_group_name / GroupNameInvalid; subscribe kwarg on register helpers; duplicate connection_init closes with 4429.
• v4.1.0 — Breaking: removed in-package demo model and CypartaGraphqlSubscriptionsTools.schema; define Query / Subscription / schema** in your project. Permission class is dotted path + **has_permission** instance API. All-or-nothing group registration per call. **filter_requested_fields` does not mutate inputs.

---

License

MIT — see LICENSE.