GraphQL Subscriptions¶

django-graphex ships an optional GraphQL subscriptions engine built on Django Channels 4. It is the modern successor of the standalone graphene-django-subscriptions package, merged here as an opt-in extra so that the base install never depends on channels.

uv add "django-graphex[subscriptions]"
# or
pip install "django-graphex[subscriptions]"

How it works¶

v2.0 ships one serialize-once engine behind two standards-based transports — you pick the transport at routing time, the engine is the same:

• SSE (Server-Sent Events) — a single async HTTP text/event-stream response. Simplest to deploy; great for a one-way notification feed.
• WebSocket (graphql-transport-ws) — the modern bidirectional protocol used by graphql-ws clients; multiplexes many subscriptions over one socket.

The data path:

1. A client sends a GraphQL subscription { ... } operation over the chosen transport (an SSE HTTP request or a WS subscribe message). Authentication is the transport's own request/scope — there is no separate channelId handshake.
2. The engine runs the subscription's authorize/scope hooks before joining any broadcast group, then streams events.
3. When a model instance changes, an in-house signal binding serializes it once and broadcasts the payload to the group. Every subscriber receives the notification, projected to the fields it requested.

Automatic save and delete notifications are queued on the database connection that performed the write. A transaction on that alias delays delivery until its commit and discards the notification on rollback. A transaction open only on default does not delay a write committed in autocommit mode on another alias. The same rule applies to the missing-channel-layer warning. This is a per-connection notification guarantee, not a cross-database transaction or all-or-nothing delivery guarantee.

Broadcast groups are named after the model, the subscription's Meta.stream and the action, so two subscriptions on the same model under different streams never receive each other's events — including when they disagree on payload_mode. Group names are an internal detail; nothing in your code should depend on their spelling.

A cross-process channel layer (Redis) is required when the producer (the process running model writes) and the subscriber processes are separate; the in-memory layer is fine for development.

Delivery failures¶

Automatic model notifications are best-effort per group: an event goes to up to four channel-layer groups and a failure to deliver to one group never prevents the others, in both ASGI and synchronous (WSGI, Celery) processes. Each failure logs one Automatic notification dropped error through the django_graphex.subscriptions.bindings logger with the model, stream, action, stage and the exception class chain (for example redis.exceptions.ConnectionError). Exception messages, group names and event data are not logged. A logging failure never fails the model write.

Close diagnostics¶

Closing a subscription is best-effort. If closing its delivery source raises a RuntimeError, the server emits one warning through the django_graphex.subscriptions.delivery logger, with the original exception and traceback (exc_info). The error remains suppressed: this diagnostic does not add a client frame or guarantee that backend group cleanup succeeded. An already-closing async generator may also cause a benign warning. Keep server logs access-controlled because exception details are not sent to clients. Normal cooperative cancellation does not trigger this warning.

1. Define a subscription¶

A Subscription is declared like a DjangoModelType, through Meta:

from django_graphex.subscriptions import Subscription
from myapp.models import User


class UserSubscription(Subscription):
    class Meta:
        model = User                          # required, a Django model class
        stream = "users"                      # required, a str
        queryset = None                        # optional
        description = "User Subscription"     # optional
        payload_mode = None                    # optional, see "Notification payload"

The notification payload is serialized through the native (Pydantic) backend.

This generates a real GraphQL subscription field whose output type is the model's projected fields (the serialized instance) — a true streaming subscription, not the legacy one-shot confirmation object. It exposes:

• Arguments: action (required, see the enum below), id (optional — scope to one instance by pk), and filter (optional — see Filtering notifications). The legacy channelId and operation arguments are gone (the transport handles connection lifecycle).
• Enum: ActionSubscriptionEnum {CREATE, UPDATE, DELETE, ALL_ACTIONS}.

2. Mount it on the schema¶

from django_graphex.core import ObjectType
from django_graphex.schema import DjangoGraphQLSchema
from myapp.subscriptions import UserSubscription


class Subscription(ObjectType):
    user_subscription = UserSubscription.Field()


schema = DjangoGraphQLSchema(query=Query, subscription=Subscription)

3. Serve subscriptions over SSE or WebSocket¶

Both transports are factories that take the live native schema and return a Django view (SSE) or a Channels consumer class (WebSocket). Mount either, or both.

SSE (HTTP text/event-stream)¶

subscription_sse_view(schema=...) returns an async Django view (requires Django >= 5.2). It parses the subscription operation, runs the engine's authorize/scope hooks before joining any group, and streams next / complete frames.

# urls.py
from django.urls import path
from django_graphex.subscriptions.transports.sse import subscription_sse_view
from myapp.schema import schema

urlpatterns = [
    path("graphql/stream", subscription_sse_view(schema=schema.graphql_schema)),
]

WebSocket (graphql-transport-ws)¶

subscription_ws_consumer(schema=...) returns a Channels AsyncJsonWebsocketConsumer subclass speaking the graphql-transport-ws protocol (connection_init/ack, multiplexed subscribe, ping/pong, per-id complete). Route it via your ASGI app:

# project/asgi.py
import os
from django.core.asgi import get_asgi_application

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

from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path
from django_graphex.subscriptions.transports.ws import subscription_ws_consumer
from myapp.schema import schema

application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,
        "websocket": URLRouter(
            [path("ws/graphql/", subscription_ws_consumer(schema=schema.graphql_schema).as_asgi())]
        ),
    }
)

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

# Development: in-memory layer. Production: use a Redis layer.
CHANNEL_LAYERS = {
    "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
}
# Production example:
# CHANNEL_LAYERS = {
#     "default": {
#         "BACKEND": "channels_redis.core.RedisChannelLayer",
#         "CONFIG": {"hosts": [("127.0.0.1", 6379)]},
#     },
# }

Per-connection schema (permission-scoped subscriptions)¶

Both transports accept an optional schema_provider in addition to the plain schema=. A schema_provider is a callable provider(user) -> GraphQLSchema resolved per connection with the connection's user (request.user for SSE, scope["user"] for WebSocket). When given, it wins over schema=. The provider contract is synchronous: the async transports run it through a thread-sensitive adapter, so a real user's initially cold permission lookup can use Django's synchronous ORM without running on the event loop. Request context variables propagate to that worker; do not depend on an arbitrary event-loop thread-local value being present there. Provider exceptions retain their transport behavior: SSE raises before opening a stream, while WebSocket frames the failed operation as an error without closing unrelated operations. This does not turn a coroutine-returning provider into a supported async provider; return a GraphQLSchema directly.

This lets a subscription connection use the same pruned schema as HTTP: wire the provider to the bundled pruned_schema_for helper so a subscription action the user is not permitted to observe is absent from the connection's schema and a subscribe naming it fails at validation (Cannot query field), exactly like the HTTP endpoint.

The bundled pruned_schema_for helper is gated by PERMISSION_SCOPED_SCHEMA — the same one flag that gates the HTTP AuthenticatedGraphQLView — read per connection (never at import):

• PERMISSION_SCOPED_SCHEMA = False (the default): the helper returns the full schema, so a provider wired to it is inert and every field validates.
• PERMISSION_SCOPED_SCHEMA = True: the helper returns the schema pruned to the connection user's permissions. An active superuser always receives the full schema.

from django_graphex.core.permission_signature_cache import pruned_schema_for
from django_graphex.subscriptions.transports.sse import subscription_sse_view
from django_graphex.subscriptions.transports.ws import subscription_ws_consumer
from myapp.schema import schema

_full = schema.graphql_schema

# SSE: the provider is resolved with request.user.
sse_view = subscription_sse_view(
    schema_provider=lambda user: pruned_schema_for(user, _full),
)

# WebSocket: the provider is resolved once per socket with scope["user"].
ws_consumer = subscription_ws_consumer(
    schema_provider=lambda user: pruned_schema_for(user, _full),
)

Passing a plain schema= (no schema_provider) is fully backward compatible — the transport behaves exactly as before, so existing wiring such as subscription_ws_consumer(schema=schema.graphql_schema) keeps working unchanged. You must pass at least one of schema= or schema_provider=.

A custom schema_provider callable that does not route through pruned_schema_for is your own code and is not gated by the setting — the flag gates the bundled helper only, so whatever schema your callable returns is used as-is.

From a DjangoModelType (one definition)¶

If you already use a DjangoModelType for queries and mutations, you can get its subscription from the same class — no separate Subscription subclass. Add stream (and optionally payload_mode) to its Meta:

from django_graphex.types import DjangoModelType
from myapp.models import User

class UserModelType(DjangoModelType):
    class Meta:
        model = User
        stream = "users"          # enables the subscription
        payload_mode = "full"     # optional; defaults to the global setting

Mount it on the schema — then serve it through either transport (see Serve subscriptions):

# schema.py
from django_graphex.core import ObjectType
from django_graphex.schema import DjangoGraphQLSchema

class Subscription(ObjectType):
    user_subscription = UserModelType.SubscriptionField()

schema = DjangoGraphQLSchema(query=Query, subscription=Subscription)

UserModelType.subscription_type() builds (and caches) the Subscription lazily, so the base install stays Channels-free until you actually wire a subscription. The generated subscription supports the same arguments — including filter. Setting Meta.stream is required to use SubscriptionField() / subscription_type(). The transport (SSE or WS) is chosen at routing time; the subscription class is transport-agnostic.

Authorization and row-scoping¶

The generated subscription honors the type's authorization and scoping:

• permission_classes / authorize gate the subscribe itself. Authorize runs at registration (the HTTP request, so info.context.user is available) for the read-like "subscribe" action; a denial yields ok: False / error and no group is joined. IsAuthenticatedOrReadOnly therefore lets anyone subscribe to a public stream, while IsAuthenticated requires a login.
• Per-action check (defense in depth). The requested subscription action (CREATE / UPDATE / DELETE / ALL_ACTIONS) is forwarded to the permission check, so a class can gate each action independently. Because a subscription payload returns instance data, DjangoModelPermissions treats a subscribe action as composite: it requires the codenames of its perms_map subscribe row plus those of the row the action maps to — with the default mapping, the model's view permission plus the action's write verb (ALL_ACTIONS requires every write verb). Because both halves are resolved through get_required_permissions, a subclass that customizes perms_map for subscribe (or for create / update / delete) is honored here too. A user permitted only CREATE is therefore denied a subscribe for UPDATE at runtime, even if the request reaches the full schema — this is the runtime half of the same model that PERMISSION_SCOPED_SCHEMA enforces at the schema layer (the action's enum value is pruned away). Custom permission_classes read the action via has_subscribe_permission(info, model, **kwargs) (the action arrives under kwargs["subscription_action"]).
• subscription_scope(info, **kwargs) returns a server-forced filter mapping (e.g. {"owner": info.context.user.pk}). The info a subscribe hook receives is the transport's own context object, and it exposes both info.user and info.context.user — the second is the spelling a resolver uses, so hook code reads the same either way. It is evaluated at subscribe time and enforced per event at delivery, combined with the client filter by AND: an event is delivered only when it matches both. The client can narrow the scope but never widen or drop it, and a client value that contradicts the scope on the same key (scope status: PUBLISHED, client filter: {status: {exact: DRAFT}}) delivers nothing instead of silently falling back to the scope. Equality scopes on a serialized field (like owner) are decided in memory, so there is no per-event query.

from django_graphex.permissions import IsAuthenticated
from django_graphex.types import DjangoModelType
from myapp.models import Note

class NoteModelType(DjangoModelType):
    permission_classes = [IsAuthenticated]

    class Meta:
        model = Note
        stream = "notes"
        payload_mode = "full"

    @classmethod
    def subscription_scope(cls, info, **kwargs):
        return {"owner": info.context.user.pk}   # only my notes

Indexed groups (optional, for high fan-out)¶

With only subscription_scope, a change is broadcast to one group per model+action and every subscriber's connection is woken to evaluate the filter in memory (cheap, but O(subscribers)). For streams with very many concurrent subscribers partitioned by a value (per owner, per tenant, per room), declare subscription_index_fields to route each change to a value-scoped group so only the matching subscribers are woken:

from django_graphex.types import DjangoModelType
from myapp.models import Message

class MessageModelType(DjangoModelType):
    class Meta:
        model = Message
        stream = "messages"
        payload_mode = "full"
        subscription_index_fields = ("tenant", "room")   # compound (AND) index

    @classmethod
    def subscription_scope(cls, info, **kwargs):
        return {"tenant": info.context.tenant_id, "room": kwargs.get("room")}

How it works: at subscribe the group name gets a canonical suffix built from the scope (e.g. messages-create:room=42&tenant=7); at broadcast the same suffix is built from the changed instance (reading each field's attname, so a foreign key yields its raw id without a query). The two names match by construction, so Channels delivers only to that group -- no group enumeration is involved.

• Opt-in and additive. Omit it and everything works exactly as before.
• Must be a subset of the scope. Every index field has to be present in what subscription_scope returns; if any is missing the subscriber transparently falls back to the coarse group (still correct, just not narrowed).
• Good index fields are high-cardinality equality keys (FKs like owner, tenant, room). A low-cardinality boolean would only create two groups and buy nothing.
• Independent of payload_mode. The index is captured from the live instance at signal time, not derived from the payload, so it works in id-only mode too.
• The full filter is still applied on delivery, so indexing is purely a routing optimization -- correctness never depends on it.

Browser client view¶

SubscriptionClientView serves a self-contained HTML page to try subscriptions live. It speaks the same standards as the transports: graphql-transport-ws over WebSocket and graphql-sse over a streamed fetch() POST — the browser EventSource API is not used to start the subscription, since EventSource cannot carry the operation (see the SSE wire protocol). The response it reads back is a standard text/event-stream, though — see the tip below for watching it live in DevTools. Add it to your URLConf like the admin — because it is served from your own origin, there is no CORS to configure:

# urls.py
from django_graphex.subscriptions import SubscriptionClientView

urlpatterns = [
    ...,
    path("graphql/client/", SubscriptionClientView.as_view()),
]

Open /graphql/client/: it connects to the chosen transport, lets you send a subscription { … } operation, and streams the notifications back. The editor ships a runnable document, not a commented-out example:

subscription {
  yourSubscription(action: ALL_ACTIONS) {
    id
  }
}

yourSubscription is a placeholder, and the introspection the client already runs for autocomplete rewrites it to the first field your subscription root advertises — so on a schema with any generated subscription, pressing ▶ works on first press. Every generated subscription field takes that same required action argument and exposes id, which is why one placeholder fits every schema. The rewrite only ever touches an editor still holding the shipped default; once you type, the client leaves your document alone.

The editor also supports Tab to indent (Shift+Tab to outdent) and schema-aware autocomplete — it introspects the configured endpoint and suggests types, fields, arguments and enum values as you type (Ctrl+Space to trigger, Enter/Tab to accept).

When introspection is disabled, GraphQL-core 3.3 also omits schema-derived suggestions from SSE and WebSocket subscription startup errors. This applies to invalid documents and variables before a source joins any group; client values and error paths remain visible. Public mode retains native suggestions. The current 3.2 runtime retains its recognized legacy error-filtering behavior.

The endpoints default to the page's own origin with the routes /ws/graphql/ (WS), /graphql/stream (SSE) and /graphql/ (HTTP); override them if yours differ:

path(
    "graphql/client/",
    SubscriptionClientView.as_view(
        ws_path="/ws/graphql/",
        sse_path="/graphql/stream",
        http_path="/graphql/",
    ),
),

5. Subscribe from a client¶

Send a normal GraphQL subscription operation over the SSE or WebSocket transport. The selection set is the model's projected fields — each event delivers the (optionally filtered) serialized instance:

subscription {
  userSubscription(action: UPDATE, id: 5) {
    id
    username
    email
  }
}

• action: ALL_ACTIONS subscribes to CREATE, UPDATE and DELETE.
• Omit id to subscribe to every instance for that action.
• Unsubscribing is the transport's own lifecycle — see Unsubscribing. There is no operation: UNSUBSCRIBE argument anymore.
• Which fields are deliverable depends on the payload mode — in id-only mode only id is present (see Notification payload).
• Relations are delivered as flat pks (post: ID, tags: [ID]) — selecting their subfields is a validation error (see Relations are flat IDs).

SSE: the wire protocol¶

The SSE transport implements the server side of graphql-sse's distinct connections mode: one HTTP request carries one subscription. The operation travels in a POST body — a JSON object with query, variables and operationName — and the response is a text/event-stream that stays open:

curl -N -X POST http://localhost:8000/graphql/stream \
  -H "Content-Type: application/json" \
  -d '{"query": "subscription { userSubscription(action: UPDATE) { id username } }"}'

Each delivered event is a next frame carrying a standard GraphQL result; the terminal complete frame ends the stream (its empty data: line is part of the protocol — an EventSource-style listener never fires complete without it):

event: next
data: {"data": {"userSubscription": {"id": "5", "username": "neo"}}}

event: complete
data: 

The optional idle heartbeat sends : heartbeat SSE comments without restarting the subscription source or changing any next/complete frame. It is off by default. See SSE connection controls.

Error surfacing splits at the moment the stream is committed:

• Pre-stream (HTTP 400) — a missing query, a GraphQL syntax error, a non-subscription operation, or a JSON body that is not an object ([1,2,3], "x", null) is rejected as a plain 400 Bad Request before any stream starts — the same classification GraphQLView gives those bodies.
• In-stream — everything after that (validation errors and authorize-denials) is delivered inside the committed 200 text/event-stream response as a single next frame carrying {"errors": [...]} followed by the complete frame — never an HTTP 4xx, because a status code cannot change once the streaming response has begun.
• Masked errors — with DEBUG=False, an unexpected exception from the subscribe resolver, an event resolver or the stream itself while it is iterated (a channel-layer error, a scope lookup) is framed with the allowlisted message, DATA_ACCESS_ERROR or INTERNAL_SERVER_ERROR, its path and an errorId, exactly as the HTTP views do (see Production execution errors); one ERROR record with the traceback is logged on django_graphex.internal_errors. A deliberate GraphQLError keeps its message and DEBUG=True keeps the raw text.
• Optional account cap — a full authenticated-account limit returns plain HTTP 429 before source startup; an unavailable admission backend returns HTTP 503 before returning a streaming response, including when ownership cannot be confirmed after source startup (that source is closed). Existing streams are not interrupted by a rejected request. If a running stream later loses its Redis lease, its already-committed 200 response emits a generic next error and complete; it cannot change to HTTP 503.

WebSocket: the wire protocol¶

The consumer speaks graphql-transport-ws — your client must offer that exact subprotocol when opening the socket. For a browser WebSocket client, pass it as the constructor's second argument:

const socket = new WebSocket("wss://example.com/ws/graphql/", "graphql-transport-ws");

That selects the transport protocol; the client must still implement its connection_init, subscribe, and other message frames. A client offering other names as alternatives is accepted only if graphql-transport-ws is among them. With no matching offer, the server rejects the handshake before accepting the socket or starting subscription work and logs:

WebSocket connection rejected: client must offer the 'graphql-transport-ws' subprotocol.

Configure the client's WebSocket subprotocol offer to resolve this error; a GraphQL connection_init frame cannot fix a failed handshake. The rejection uses ASGI close code 4406, but a pre-accept close is normally sent on the wire as HTTP 403. Browser APIs may show only a generic connection error or 1006, not the server log, a GraphQL JSON error, or a guaranteed 4406/reason.

Optional connection authentication¶

By default, connection_init ignores its payload and uses the user supplied by Channels middleware in scope["user"]. Existing session-based clients, including the Playground, need no change. To authenticate an initialization credential, opt in with an async on_connect(scope, payload) callback:

from collections.abc import Mapping
from typing import Any

from django.contrib.auth.base_user import AbstractBaseUser

from django_graphex.subscriptions.transports.ws import subscription_ws_consumer
from myapp.auth import verify_token  # Project-defined async credential verifier.
from myapp.schema import schema

async def authenticate_socket(
    scope: Mapping[str, Any], payload: Mapping[str, Any]
) -> AbstractBaseUser | None:
    """Authenticate the socket through the project's credential service.

    Args:
        scope: The Channels connection scope, including any session identity.
        payload: The normalized initialization payload containing credentials.

    Returns:
        The authenticated user for this socket, or None for rejected credentials.
    """
    token = payload.get("token")
    # The project-defined async verifier may also raise for invalid credentials.
    return await verify_token(token)

consumer = subscription_ws_consumer(
    schema=schema.graphql_schema,
    on_connect=authenticate_socket,
)

The client sends {"type": "connection_init", "payload": {"token": "…"}} after opening the socket with the graphql-transport-ws subprotocol. The callback runs before connection_ack, schema resolution, or subscription work. Its returned authenticated user replaces any session user for this socket; both permission checks and schema_provider see that one identity. It does not log a Django session in or alter HTTP/SSE authentication. The application owns credential format, verification, expiry, and reconnect behavior; the transport does not supply JWT validation or mid-connection refresh.

With a callback, an absent or null payload becomes {}, while a non-object payload closes the already accepted socket with 4400 before callback invocation. Missing, anonymous, or rejected user credentials close with 4403 and a generic server diagnostic; no ACK, schema resolution, or fallback to a previous session identity occurs. The initialization deadline remains active while the callback runs: a stalled callback closes with 4408; a second connection_init closes with 4429, and a subscribe before ACK with 4401. Do not place tokens in client-visible errors or application logs.

Once connected, one socket multiplexes any number of operations, each identified by a client-chosen id:

After the server sends complete, it has finished this operation's source cleanup attempt and released its ID slot. A client may immediately send a new subscribe with the same ID; a genuinely concurrent duplicate still closes with 4409.

The subscribe payload carries the same query / variables / operationName keys as an HTTP POST. A subscribe-time failure (parse error, validation error, authorize-deny, a payload that is not a JSON object, or the per-connection cap) is answered with {"type": "error", "id": "1", "payload": [...]} instead of next frames, and the server answers a client {"type": "ping"} with {"type": "pong"}. The graphql-ws JavaScript client speaks this protocol out of the box.

Protocol violations close the socket with the spec's codes: 4400 malformed message, 4401 subscribe before the ack, 4408 no connection_init within the init timeout, 4409 duplicate operation id, 4429 a second connection_init. 4400 covers any frame the consumer cannot dispatch — including a body that is not valid JSON and a non-string id — and the close tears down every live operation on that socket first, so a malformed frame can never leave a task or a joined group behind. Because any client can send such a frame, it is logged at DEBUG, not as an ERROR with a traceback. Errors in error and next frames are masked in production like the HTTP views' (see Production execution errors).

A failure that happens after delivery started is not a close and not an error frame: it arrives as {"type": "next", "payload": {"errors": [...]}} followed by the terminal complete, so the client always gets a protocol signal instead of a stream that goes quiet forever. The SSE transport does the same thing with its own framing (event: next carrying errors, then event: complete) — once the 200 text/event-stream response is committed an HTTP error is impossible.

Unsubscribing¶

Both transports tear down a gone client's source by discarding its joined channel-layer groups. If cancellation arrives while a group join is in flight, startup also attempts to discard that source's channel from the attempted group. A conforming channel layer ignores a discard for a membership that was never created, as required by the Channels channel-layer specification. Backend discard failures can still prevent cleanup; this is not a guarantee across backend outages or distributed races. A failed source startup can be retried on the same source; an explicitly closed source stays closed. Cleanup failures are logged internally without replacing the startup error.

WebSocket — send a complete frame for the operation id:

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

This cancels only operation "1": its task is cancelled and its groups are discarded, while the other operations multiplexed on the same socket keep streaming. Per the protocol the server does not echo a complete back for a client-initiated one — a server complete means "this stream ended on its own". Closing the socket cancels all operations and discards every joined group.

SSE — one request is one stream, so unsubscribing means closing the HTTP connection: abort the fetch() (AbortController.abort()), dispose the graphql-sse client subscription, or Ctrl-C the curl. Django ≥ 5.2 guarantees the disconnect cancels the streaming generator, and the transport's cleanup then leaves every joined group — no ghost subscribers. Teardown is also registered on the response itself, so a client that aborts during the subscribe handshake — before the response body is ever read — releases its groups too.

Filtering notifications¶

id scopes by the changed object's own primary key. To scope by field values instead — e.g. a post-detail page that should only receive the comments of that post — pass the optional filter argument. It is a real generated input object, <Model>SubscriptionFilterInput, with the same nested shape queries use:

subscription {
  commentSubscription(
    action: ALL_ACTIONS
    filter: { post: { exact: 7 } }       # only comments whose post == 7
  ) {
    id
    text
  }
}

The type exposes exactly the subscription's projected output fields (see only_fields / exclude_fields), each with exactly four lookups — exact, iexact, in, isnull. Nothing else is expressible: the schema itself is the boundary now, so your IDE autocompletes the valid keys and an invalid one fails validation before the request ever reaches the engine.

Filters are evaluated per connection at delivery time:

• exact is decided in memory against the serialized payload — no extra query. This is the fast path the whole serialize-once engine is built around, so prefer it for scoping.
• A native Float exact filter matches an event snapshot when both values are actual finite floats with exactly equal numeric values. In particular, -0.0 and 0.0 now match in either direction. This also changes the low-level split_filters result for a legacy dictionary containing those two float values. It does not parse numeric strings, use an epsilon, change Boolean/null matching, or normalize indexed group keys. Non-finite values retain the previous string-comparison behavior. A full-payload field must be present in the snapshot for this in-memory comparison; missing fields and other lookups retain their database-check behavior.
• A typed Decimal exact filter compares finite numeric values from the event-time JSON snapshot, so 1.0 matches a serialized "1.00". This also works for valid high-precision values, including snapshots longer than 256 characters, and for full-payload delete events after the row is gone. No separate snapshot-length cutoff is added by this comparison. Server scope values supplied as strings keep their existing string comparison, while malformed or non-finite values retain the prior fallback behavior. The engine does not reread a possibly changed row to resolve this equality.
• A typed DateTime exact filter compares the captured full-payload ISO value as a DateTime, without millisecond rounding or a later row lookup. All six microsecond digits survive transport; equivalent timezone-aware offsets match the same instant, while different microseconds do not. Naive values are not assigned a timezone, and this does not change other scalar filters. For example, an event captured at 2026-10-18T12:00:00.123456+00:00 matches exact: "2026-10-18T17:30:00.123456+05:30" and delivers that captured DateTime through the existing output scalar. An older notification already truncated to milliseconds cannot regain its lost digits.
• Numeric snapshots and typed filters compare by value whichever of int, float or Decimal each side is: an integer or float snapshot (a Float or Decimal field saved with a plain number, so the payload carries 1 or 1.5) matches 1.0 or Decimal("1.50"). When a float is on either side the other operand is first rounded to the nearest float, which is what a float column stores: 2**53 + 1 equals float(2**53) and Decimal("0.1") equals 0.1. Without a float the comparison is exact. Booleans, NaN, Infinity and integers beyond the float range never match.
• The other three lookups fall back to a single-row database check while the row exists. A deleted row cannot satisfy that check.
• Combine them: filter: { post: { exact: 7 }, status: { in: ["open", "urgent"] } }.
• Omitting filter delivers every event in the group.

commentSubscription(action: ALL_ACTIONS, filters: "{\"post\": 7}")

commentSubscription(action: ALL_ACTIONS, filter: { post: { exact: 7 } })

class UserSubscription(Subscription):
    class Meta:
        model = User
        stream = "users"
        exclude_fields = ("password", "user_permissions")

What the client receives is a normal GraphQL result — its own selection set projected from the event (wrapped in the transport's frame: an SSE next event or a WS next{id, payload} message). Subscribing with { id username email } in the default id-only mode delivers:

{"data": {"userSubscription": {"id": "5", "username": null, "email": null}}}

…and in full-payload mode, the serialized values:

{"data": {"userSubscription": {"id": "5", "username": "neo", "email": "neo@example.com"}}}

Notification payload¶

On every model change the binding builds the notification data. By default it is id-only — {"id": <pk>} — which skips full-payload serialization on create and update (the fastest option for hot models, where clients typically refetch on notification). An automatic delete additionally captures projected scalar values privately for equality matching; the public data remains only the primary key. Set it to full to serialize the whole instance through the native (Pydantic) backend.

Two controls, in order of precedence:

# settings.py — make every subscription serialize the full instance by default
DJANGO_GRAPHEX = {
    "SUBSCRIPTION_PAYLOAD_MODE": "full",
}

# ...or decide per subscription, regardless of the global default
class UserSubscription(Subscription):
    class Meta:
        model = User
        stream = "users"
        payload_mode = "full"   # full payload for this one; "full"/"id_only"/None

Relations are flat IDs (no nested selections)¶

The generated event type — <Model>SubscriptionEvent — renders every relation flat: a ForeignKey / OneToOneField field is the related object's pk scalar (ID for the usual auto pk, otherwise the pk field's own scalar), and a ManyToManyField is a list of pks ([ID]). This is the serialize-once design working as intended: a change is serialized once (FK → pk, M2M → [pk]) and that single flat payload feeds every subscriber with zero per-subscriber database queries — resolving a nested object per subscriber would reintroduce exactly the per-event N+1 the engine exists to avoid.

Select relations as scalars:

subscription {
  commentSubscription(action: CREATE) {
    id
    text
    post      # ID — the related post's pk, not an object
    tags      # [ID] — the related pks
  }
}

Selecting subfields on a relation fails GraphQL validation — before any event is delivered:

subscription {
  commentSubscription(action: CREATE) {
    post { id title }
  }
}

Field 'post' must not have a selection since type 'ID' has no subfields.

(On SSE the error arrives in-stream; on WebSocket as an error frame — see the wire protocols.)

The payload mode does not change this schema — payload_mode only decides which keys the broadcast payload carries, and the flat resolvers simply read their key from it:

In id-only mode the payload is just {"id": <pk>}, so any selected non-id field resolves null — the selection still validates against the same schema, and nothing errors at delivery. In full mode the broadcast carries all concrete fields plus the M2M pk lists, and each subscriber's selection projects from that single dict.

Per-connection field selection¶

Each subscriber's GraphQL selection set is its own projection: two clients on the same stream requesting different fields each receive only the fields they asked for, resolved per event from the single serialized payload (no shared serializer state, no re-serialization).

Security hardening¶

Validate the WebSocket handshake's Origin¶

Wrap the routed WebSocket application in Channels' AllowedHostsOriginValidator, outside the auth stack:

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.sessions import SessionMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AllowedHostsOriginValidator(          # <- the guard
        SessionMiddlewareStack(
            AuthMiddlewareStack(
                URLRouter([path("ws/graphql/", MyWSConsumer.as_asgi())])
            )
        )
    ),
})

It checks the handshake's Origin against ALLOWED_HOSTS, so a foreign origin never completes the handshake and never reaches connection_init. Note that ALLOWED_HOSTS = ["*"] makes it accept everything — convenient in development, and worth remembering when you read a test that seems to pass without it. The playground ships this wiring, and pins it with a test that rebuilds the app under a real host list.

Transport-level authentication¶

v2.0 has no separate channelId handshake or channel-ownership cache to guard — the legacy two-channel protocol (and its SUBSCRIPTIONS_CHANNEL_GUARD setting) was removed. Authentication is now the transport's own request/scope:

• SSE authenticates the HTTP request (request.user / session / middleware) before the engine joins any group.
• WebSocket authenticates the connection scope at connection_init (the auth boundary) before any subscription is accepted. An optional on_connect callback can replace that user after validating the init payload; no callback preserves the existing session behavior.

The subscription's authorize_subscription / permission_classes hooks run before any group_add, so a denial short-circuits before the source is created — there is no window where an unauthorized subscriber is joined.

Per-connection subscription cap¶

One graphql-transport-ws socket multiplexes any number of operations, and every live operation joins its own channel-layer group — so an unbounded socket lets a single client amplify one connection into hundreds of subscribers. MAX_SUBSCRIPTIONS_PER_CONNECTION (default 50) bounds it, the same way MAX_BATCH_SIZE bounds a batch request:

DJANGO_GRAPHEX = {
    "MAX_SUBSCRIPTIONS_PER_CONNECTION": 50,  # None disables the cap
}

A subscribe past the cap is rejected, not fatal: the server answers that operation id with an error frame naming the limit, and every subscription already running on the socket keeps streaming. The slot frees itself as soon as an operation ends — a client complete, a stream that ends on its own, or a disconnect — so a client that stays under the cap never notices it.

The WebSocket cap is per socket, not per account. SSE has an independent, optional per-authenticated-account cap below. One SSE request still carries exactly one subscription.

SSE connection controls¶

Both SSE controls are independent and disabled by default. The heartbeat value defaults to 15 seconds and the account-cap value defaults to five, but neither value enables its control. To enable either one:

import os

DJANGO_GRAPHEX = {
    "SSE_HEARTBEAT_ENABLED": True,
    "SSE_HEARTBEAT_INTERVAL": 15,  # seconds; positive finite number
    "SSE_CONNECTION_LIMIT_ENABLED": True,
    "SSE_MAX_CONNECTIONS_PER_USER": 5,  # positive integer
    "SSE_CONNECTION_REDIS_URL": os.environ["SSE_CONNECTION_REDIS_URL"],
}

Enable only the keys you need; heartbeat does not require Redis. The cap requires one reachable Redis deployment shared by all SSE workers and the optional Python redis client (the subscriptions extra brings it through channels-redis). Set a redis:// or rediss:// URL only when enabling the cap; keep credentials outside source control. A missing client, URL, or unavailable Redis instance fails new authenticated admissions with HTTP 503. Redis connect and command calls request two-second per-socket timeouts. This is not a two-second end-to-end deadline: Redis client retries and URL query options can extend the total wait or override the requested timeouts. Keep operator-supplied retry and timeout options bounded for the deployment. Anonymous requests bypass this cap without an IP-based bucket. The cap does not alter WebSocket operations or authentication policy.

Admission, renewal, and release use atomic Redis operations keyed by a stable account identity and a unique stream token. The sixth concurrent stream under the example cap receives HTTP 429 before subscribing; the first five continue. Each active stream renews its own lease even when heartbeat is off; ownership is checked after startup, when body consumption begins, and before delivering a result whenever renewal is due. Already expired tokens are not recreated. A cancelled admission settles before token-specific cleanup, though backend failure can still defer cleanup until expiry. Normal completion, disconnect, source-start failure, and response close release the token. A crash or failed release relies on a 45-second lease expiry; a renewal that sees an expired token stops the stream instead of recreating its slot. A delayed consumer may therefore lose a lease and need to reconnect. A renewal that fails (timeout, connection error) is not treated as a lost lease: the stream stays open and retries every five seconds while the last confirmed renewal is less than 40 seconds old (the 45-second TTL minus a five-second safety margin). A later success resets that window. Only when failures outlast it, or Redis answers that the token is no longer owned, does the stream end with the usual lease-lost frame; the frame never carries backend details. A short Redis outage therefore no longer drops every stream at once. All streams of a process share one Redis client (connection pool) per URL. The client pings idle connections every 30 seconds and enables TCP keepalive; lease admission retries once after a connection error, so a connection dropped by a proxy does not fail the next admission with HTTP 503. Redis failover and network partitions can weaken a strict global count: this is an availability guard, not a distributed consensus guarantee. Keep the server/gateway connection limit as a separate safeguard.

Maintainers can run the explicit real-Redis SSE integration suite using the isolated setup in integration/live_redis/README.md. It is separate from the default test suite, which does not require a Redis service. Its direct Lua checks use a private Unix socket, where the shared client retains the same timeouts and idle health checks but omits TCP-only socket keepalive. This does not add Unix URLs to the public SSE_CONNECTION_REDIS_URL setting.

GraphQL execution middleware¶

Both transports build the DJANGO_GRAPHEX['MIDDLEWARE'] chain once per connection and run it on the subscribe entry and on every per-event delivery execute — the same chain GraphQLView runs for queries and mutations. So AuthenticatedFieldsMiddleware (the enforcement half of private_subscription, see Security) actually protects subscription fields: an unauthenticated subscriber is denied before any group_add, and no event is ever delivered to it.

Filter key validation¶

Since 2.1.0 the schema is the first gate: <Model>SubscriptionFilterInput declares only the projected output fields, each with only the four allowed lookups, so anything else fails GraphQL validation before the subscribe resolver runs. A second, runtime check (_validate_client_filters) still validates the flattened key as defence in depth for anything that reaches the engine without going through schema coercion:

1. the root (everything before the first __) must be a declared output field of the subscription's serialized payload;
2. every remaining segment must be a Django lookup or transform registered on that field;
3. every remaining segment must also be one of the four allowed lookups: exact, iexact, in, isnull;
4. the whole mapping must be a query the ORM itself can build.

Schema rejections arrive as a normal GraphQL validation error and no group is joined; a runtime rejection likewise raises before any group is joined, and its message names the offending lookup alongside the allowed ones.

Step 4 exists because a Django field's declared lookup registry is wider than what the query compiler accepts once a join is involved. A to-many field (tags) declares iexact and then refuses it, so { tags: { iexact: 3 } } passed steps 1–3 and blew up at delivery instead — deep inside the stream, after the SSE 200 was already committed. The engine now builds the same query at subscribe time (no SQL is issued, only the lookup resolution), so an unsatisfiable filter is a clean denial with no group joined.

Why: the filter argument executes as ORM lookups at event-delivery time and delivery is observable, so a filter key is a boolean oracle. Rooting it on a declared field bounds which column it can probe; restricting the lookup bounds how much each probe reveals. An ordered or pattern lookup answers a comparison, so a few hundred probes recover a password hash character by character; equality and membership answer only "is it exactly this value?", which forces a whole-value guess.

What remains: a client can still test equality against any declared output field. That is a weak oracle for a high-entropy value and a real one for a low-entropy value — a boolean flag falls in two probes. Declared fields are the names returned by the subscription's backend.output_field_names() after the Meta.only_fields / Meta.exclude_fields projection, so use that projection to keep sensitive columns out of both the payload and the filter surface. If you need tighter control still, override subscription_scope to enforce server-side filters that cannot be widened or removed by the client.

class CommentSubscription(Subscription):
    """Comment change notifications."""

    class Meta:
        model = Comment
        stream = "comments"
        payload_mode = "full"
        exclude_fields = ("internal_note",)   # <- not inherited from CommentType

Percent-encoded index group names¶

Subscription index fields (see subscription_index_fields) use = and & as delimiters in the group name suffix. Field values are percent-encoded (urllib.parse.quote) before the suffix is assembled, so a value like "a=b" no longer produces a name that is ambiguous with two separate key-value pairs.

Asserts replaced with explicit raises¶

The validation guards in Subscription.__init_subclass_with_meta__ that were expressed as assert statements are now raise TypeError(...) calls. Python's -O flag strips assert at compile time; the explicit raises survive.

Safe async_to_sync in signal handlers¶

The signal binding (SubscriptionBinding) calls channel_layer.group_send from a synchronous Django signal handler. Under an ASGI server a running event loop is already present on the calling thread; a naive async_to_sync(…)() call in that context deadlocks.

bindings.py detects whether a loop is running (asyncio.get_running_loop). On the ASGI path (loop present), the coroutine is scheduled as a fire-and-forget task via loop.create_task — the call returns immediately without blocking the loop thread, and a done-callback logs any delivery failure via the module logger. On the WSGI / sync path (no running loop), async_to_sync is used. Explicit calls to SubscriptionBinding.broadcast retain synchronous error propagation; automatic model-signal notifications use the best-effort policy below.

Automatic notification failures¶

Automatic post_save and post_delete notifications are best-effort. An ordinary error while preparing a notification snapshot or delivering it after commit drops that event and logs only the model label, stream, action, stage, and exception class. The payload, routing index, primary key, user, and exception message are not logged by this boundary. A successful database write still succeeds, and a failed delivery does not stop later on_commit callbacks. This also applies to generated mutations that save the model.

This is a change from synchronous automatic delivery errors propagating into the caller. If an application depended on that exception to fail a mutation, it must handle notification delivery explicitly instead. Explicit SubscriptionBinding.broadcast calls retain their previous strict behavior. Model validation and database write errors still propagate; rollback still emits no event. Cancellation and other BaseException control signals are not converted into successful writes.

Best-effort delivery does not guarantee that a subscriber receives the event: there is no retry, outbox, or durable queue. Monitor the metadata-only drop logs and use an application-owned outbox if delivery guarantees are required. An event's payload and routing data remain frozen at signal time, even when the same model instance changes before the transaction commits.

Commit-time broadcast delivery¶

Subscription broadcasts (_on_save, _on_delete) are now deferred via django.db.transaction.on_commit. This means:

• No phantom notifications: if a nested write runs inside transaction.atomic() and a later child save raises an IntegrityError (triggering a rollback), the broadcast callbacks are discarded — subscribers never receive events for rows that were rolled back.
• Auto-commit path unchanged: when no explicit transaction is open, Django's auto-commit mode causes on_commit to call the callback immediately, so broadcast delivery is instant for standalone saves.
• Nested atomic() blocks: callbacks accumulate until the outermost transaction commits, matching Django's standard on_commit semantics.
• Signal-time snapshots: each post_save / post_delete captures its primary key, routing index and payload before registering the callback. Saving the same Python object again — or deleting it, which clears its pk — cannot rewrite an earlier event while the transaction is still open. Every post_save remains a separate event; changes are not coalesced.

Automatic and manual write paths¶

Automatic notifications come from Django's post_save and post_delete signals, not from every ORM write. bulk_create, bulk_update, QuerySet.update, and M2M.add do not send those model signals and therefore do not automatically broadcast through this binding. Use an ordinary model save() when signal-driven delivery is required.

For a write path that deliberately bypasses signals, the manual API is Subscription.get_binding().broadcast(action, instance), where Subscription is your concrete subscription class. This call captures and sends the instance's current state immediately; unlike automatic signal delivery, it is not queued with on_commit and can notify before a rollback. If the event must describe committed state, arrange the manual call through application-owned hooks after the outermost commit, using a fresh instance loaded at that point for an update. For a deletion, retain the needed event data in application code before the row disappears. Do not defer a lambda over a mutable model instance and assume it preserves the earlier state.

ORM access in the subscribe hooks¶

authorize_subscription and subscription_scope are synchronous classmethods (user-overridable), and both transports run them on the event loop — not in a thread. A blocking ORM query inside either one therefore raises SynchronousOnlyOperation ("You cannot call this from an async context"), which the engine reports as a subscribe denial, not as a crash. Reading info.context.user is free: both transports resolve the user before the hooks run. Anything past it — a related row, a permission lookup, a Model.objects call — must be lifted by you:

from asgiref.sync import sync_to_async
from graphql import GraphQLError


class PostSubscription(Subscription):
    class Meta:
        model = Post

    @classmethod
    async def authorize_subscription(cls, info, **kwargs):
        user = info.context.user
        if not user.is_authenticated:
            raise GraphQLError("authentication required")
        allowed = await sync_to_async(user.has_perm)("blog.view_post")
        if not allowed:
            raise GraphQLError("not allowed")

Both hooks may be declared async def (as above) or stay synchronous; an awaitable return is awaited either way. Keep a synchronous hook free of blocking ORM access.

JavaScript path escaping¶

SubscriptionClientView injects ws_path and http_path into inline JavaScript via json.dumps, which escapes double quotes and backslashes. Injecting raw strings would allow XSS if a path contained a double-quote (") that could close the surrounding JS string literal.

Migrating from graphene-django-subscriptions / v1.x¶

The transport changed in v2.0: the HTTP channelId handshake, the GraphqlAPIDemultiplexer consumer, and SubscriptionGraphQLView were removed in favor of native SSE and graphql-transport-ws. To migrate:

1. Install the extra: uv add "django-graphex[subscriptions]" (or pip install "django-graphex[subscriptions]").
2. Update imports to django_graphex.subscriptions.
3. Replace the demultiplexer consumer + SubscriptionGraphQLView URL with the native transports: route subscription_ws_consumer(schema=...) for WebSocket and/or mount subscription_sse_view(schema=...) for SSE (see Serve subscriptions).
4. Update clients: drop the channelId / operation arguments and the {ok, error, stream, operation, action} confirmation selection — select the model's fields instead, and rely on the transport for unsubscribe (abort the SSE request or send a graphql-transport-ws complete — see Unsubscribing).
5. Notifications are id-only by default. If your clients relied on the full serialized payload, set Meta.payload_mode = "full" on those subscriptions or DJANGO_GRAPHEX["SUBSCRIPTION_PAYLOAD_MODE"] = "full" globally. See Notification payload.
6. Configure a Redis channel layer for multi-process deployments.