cypartagraphqlsubscriptionstools 4.1.5

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

CypartaGraphqlSubscriptionsTools

Version 4.1.5

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.

---

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.

Wrap WebSocket URLs in AuthMiddlewareStack so Channels populates user (see Quick start).

Token in query string or header: implement thin ASGI middleware that copies scope, resolves user (e.g. with channels.db.database_sync_to_async), sets scope["user"], then delegates to the inner application. Compose it with AuthMiddlewareStack according to whether you still need session/cookie auth; see the Channels authentication docs.

Example sketch (query param ?token=... — replace lookup with your Authorization: Token <key> parsing if you use headers):

# articles/middleware.py
from urllib.parse import parse_qs

from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser


@database_sync_to_async
def user_for_token(key: str):
    if not key:
        return AnonymousUser()
    from rest_framework.authtoken.models import Token

    try:
        return Token.objects.select_related("user").get(key=key).user
    except Token.DoesNotExist:
        return AnonymousUser()


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

    async def __call__(self, scope, receive, send):
        if scope["type"] == "websocket":
            query = parse_qs(scope.get("query_string", b"").decode())
            token = (query.get("token") or [None])[0]
            scope = dict(scope)
            scope["user"] = await user_for_token(token)
        return await self.inner(scope, receive, send)

For Authorization: Token <key>, decode dict(scope["headers"]) (byte keys/values), find b"authorization", strip the Token prefix, and resolve the user the same way.

---

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