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.

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.

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

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, merged over the client filter with server precedence — the client can neither widen nor drop it. 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).

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: 

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.

WebSocket: the wire protocol¶

The consumer speaks graphql-transport-ws — your client must request that subprotocol when opening the socket. One socket multiplexes any number of operations, each identified by a client-chosen id:

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.

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 guarantee that a gone client is a gone subscriber — teardown discards every channel-layer group the operation had joined.

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.
• The other three lookups fall back to a single-row database 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 serialization entirely (the fastest option for hot models, where clients typically refetch on notification). 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.

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 SSE transport needs no equivalent: one text/event-stream request carries exactly one subscription, so its concurrency is bounded by your server's connection limits rather than by an in-process registry.

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 and errors propagate synchronously as before.

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.

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.