cypartagraphqlsubscriptionstools 4.0.1

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

CypartaGraphqlSubscriptionsTools

Graphene + Django GraphQL subscriptions over Django Channels (async WebSockets). Supports graphql-transport-ws and graphql-ws via the Sec-WebSocket-Protocol header.

For background on the two protocols, see GraphQL over WebSockets: subscription-transport-ws vs graphql-ws.

---

What you get (v4.0.1)

• Per-connection bounded outbox — asyncio.Queue + one sender task (slow clients cannot queue unbounded work). Configure with CYPARTA_WS_OUTBOX_MAXSIZE (default 256).
• Multi-operation aware — each client subscribe uses a transport id; groups and payloads are keyed per operation (_ops / _group_ops).
• Live payload shape — each event is sent as GraphQL ExecutionResult data shaped as { "<responseKey>": value }, where response key is the subscription root field’s alias if present, otherwise the field name (from graphql.parse).
• Lifecycle helpers — optional CypartaSubscriptionModelMixin (django-lifecycle hooks) and trigger_subscription for channel layer broadcasts.

---

Requirements

• Python ≥ 3.9
• Django, Graphene / graphene-django, Channels, django-lifecycle (see setup.py / requirements.txt for pinned versions in this repo).

Install:

pip install cypartagraphqlsubscriptionstools

Or from source:

pip install -e .

---

1. Install and enable the app

Add the app to INSTALLED_APPS:

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

Configure channel layers (Redis for production; in-memory is fine for local dev):

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

Optional:

# Max queued outbound messages per WebSocket before drops (default 256).
CYPARTA_WS_OUTBOX_MAXSIZE = 512

Point ASGI_APPLICATION at your routing module (see below).

---

2. Wire ASGI and WebSocket routing

Mount CypartaGraphqlSubscriptionsConsumer on a URL your GraphQL WS client will use.

Option A — reuse the package URL patterns

# your_project/routing.py
from channels.routing import URLRouter
from django.urls import path

from CypartaGraphqlSubscriptionsTools.routing import websocket_urlpatterns

# Or merge with your own patterns:
urlpatterns_websocket = [
    *websocket_urlpatterns,
    # path("ws/other/", OtherConsumer.as_asgi()),
]

Option B — single explicit path

from django.urls import path
from CypartaGraphqlSubscriptionsTools.consumers import CypartaGraphqlSubscriptionsConsumer

websocket_urlpatterns = [
    path("graphql/", CypartaGraphqlSubscriptionsConsumer.as_asgi()),
]

ASGI entry (typical pattern):

# your_project/asgi.py
import os

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

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "your_project.settings")
django_asgi_app = get_asgi_application()

from your_project.routing import urlpatterns_websocket  # adjust import

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

# settings.py
ASGI_APPLICATION = "your_project.asgi.application"

---

3. Point Graphene at your schema

The consumer runs subscriptions with graphene_settings.SCHEMA:

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

# your_project/schema.py
import graphene

from your_app.graphql.subscriptions import Subscription as AppSubscription


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

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


class Subscription(AppSubscription):
    pass


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

---

4. WebSocket protocol (what the client must do)

1. Negotiate subprotocol — the handshake must include Sec-WebSocket-Protocol: graphql-transport-ws or graphql-ws. Unsupported values are rejected with close code 1002.
2. connection_init — send first; the server replies with connection_ack and only then accepts subscribe.
3. subscribe — must include a string id (GraphQL transport operation id). If subscribe arrives before connection_init, the socket is closed with 4401.
4. complete — may be sent even before connection_init only when an id is present; the server tears down that operation and sends {"type": "complete", "id": "<id>"} when applicable.

Ping / keepalive: the server periodically sends ping (transport-ws) or ka (graphql-ws).

---

5. Writing subscription resolvers

Inside a subscription resolver, root is the CypartaGraphqlSubscriptionsConsumer instance. Join or leave channel groups with detect_register_group_status. Because the consumer is async, call it from sync Graphene code with async_to_sync:

from asgiref.sync import async_to_sync
from CypartaGraphqlSubscriptionsTools.utils import get_model_name_instance

Typical pattern:

async_to_sync(root.detect_register_group_status)(
    name_list,           # e.g. ["MyModelCreated"]
    subscripe,           # True = join groups, False = leave
    requested_fields,    # optional list of field names for payload filtering, or None
    operation_id=None,   # optional; omit during normal subscribe execution (uses active op id)
)

requested_fields — when not None, only those keys are kept under the serialized fields dict in the pushed payload (see filter_requested_fields in utils.py).

Group names — align with what you pass to trigger_subscription (see below). The mixin uses:

• {ModelName}Created
• {ModelName}Updated.{pk}
• {ModelName}Deleted.{pk}

Example: model created

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

from CypartaGraphqlSubscriptionsTools.utils import get_model_name_instance
from your_app.models import MyModel


class MyModelType(DjangoObjectType):
    class Meta:
        model = MyModel


class Subscription(graphene.ObjectType):
    my_model_created = graphene.Field(MyModelType, subscripe=graphene.Boolean(required=True))

    def resolve_my_model_created(root, info, subscripe):
        requested_fields = [
            s.name.value for s in info.field_nodes[0].selection_set.selections
        ]
        model_name = get_model_name_instance(MyModelType)
        return async_to_sync(root.detect_register_group_status)(
            [f"{model_name}Created"],
            subscripe,
            requested_fields,
        )

Use the same idea for Updated / Deleted with groups like f"{model_name}Updated.{id}" (match your client arguments and your trigger_subscription calls).

---

6. Model mixin (optional)

Subclass CypartaSubscriptionModelMixin so creates / updates / deletes emit channel events (requires django-lifecycle on the model):

# your_app/models.py
from django.db import models
from CypartaGraphqlSubscriptionsTools.mixins import CypartaSubscriptionModelMixin


class Article(CypartaSubscriptionModelMixin, models.Model):
    title = models.CharField(max_length=200)

---

7. Publishing events from your code

Use trigger_subscription to send a message to everyone in a channel group. Values that are models.Model instances are passed through serialize_value (JSON serialize + shape with pk, fields, optional group):

from asgiref.sync import async_to_sync
from CypartaGraphqlSubscriptionsTools.events import trigger_subscription


async_to_sync(trigger_subscription)("MyModelCreated", instance)

Custom group names work as long as subscription resolvers register the same strings.

---

8. Optional WebSocket auth middleware

The package includes TokenAuthMiddleware (Authorization: Token <key> → sets scope["user"]). It expects Django REST framework’s Token model to be available if you use it:

# asgi.py (excerpt)
from channels.auth import AuthMiddlewareStack
from channels.routing import URLRouter

from CypartaGraphqlSubscriptionsTools.middleware import TokenAuthMiddleware
from your_project.routing import urlpatterns_websocket

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

---

9. Upgrading from older releases

• RxPY removed — delivery uses a bounded queue + sender task.
• Adapter settings removed — no CYPARTA_GRAPHQL_SUBSCRIPTION_ADAPTER, CYPARTA_LEGACY_SUBSCRIPTION_DATA, or adapt_channel_event.
• Synthetic register/unregister “next” acks removed — wire traffic is normal GraphQL next/data payloads plus complete when an operation ends.
• Subscribe must include id; connection_init before subscribe is enforced (4401 if violated).
• Payload data uses { responseKey: ... } (alias-aware) for live subscription events.

---

Links

• Django Channels installation
• Channel layers