GraphQLWSConsumer

A channels websocket consumer for GraphQL.

This handles the connections, then hands off to the appropriate handler based on the subprotocol.

To use this, place it in your ProtocolTypeRouter for your channels project, e.g:

from strawberry.channels import GraphQLHttpRouter
from channels.routing import ProtocolTypeRouter
from django.core.asgi import get_asgi_application

application = ProtocolTypeRouter({
    "http": URLRouter([
        re_path("^graphql", GraphQLHTTPRouter(schema=schema)),
        re_path("^", get_asgi_application()),
    ]),
    "websocket": URLRouter([
        re_path("^ws/graphql", GraphQLWebSocketRouter(schema=schema)),
    ]),
})

Constructor:

def __init__(
    self,
    schema: BaseSchema,
    keep_alive: bool = False,
    keep_alive_interval: float = 1,
    debug: bool = False,
    subscription_protocols: Tuple[str, str] = (GRAPHQL_TRANSPORT_WS_PROTOCOL, GRAPHQL_WS_PROTOCOL),
    connection_init_wait_timeout: datetime.timedelta | None = None,
) -> None:
  ...

Methods:

• pick_preferred_protocol

  Signature:

  def pick_preferred_protocol(self, accepted_subprotocols: Sequence[str]) -> str | None:
    ...

  Parameters:

  1. accepted_subprotocols:

     Type

     Sequence[str]

• connect

  Signature:

  def connect(self) -> None:
    ...

• receive

  Signature:

  def receive(self, args: str = (), kwargs: Any = {}) -> None:
    ...

  Parameters:

  1. args:

     Type

     str

     Default

     ()

  2. kwargs:

     Type

     Any

     Default

     {}

• receive_json

  Signature:

  def receive_json(self, content: Any, kwargs: Any = {}) -> None:
    ...

  Parameters:

  1. content:

     Type

     Any

  2. kwargs:

     Type

     Any

     Default

     {}

• disconnect

  Signature:

  def disconnect(self, code: int) -> None:
    ...

  Parameters:

  1. code:

     Type

     int

• get_root_value

  Signature:

  def get_root_value(self, request: ChannelsConsumer) -> RootValue | None:
    ...

  Parameters:

  1. request:

     Type

     ChannelsConsumer

• get_context

  Signature:

  def get_context(self, request: ChannelsConsumer, connection_params: Any) -> Context:
    ...

  Parameters:

  1. request:

     Type

     ChannelsConsumer

  2. connection_params:

     Type

     Any

Attributes:

1. graphql_transport_ws_handler_class:

2. graphql_ws_handler_class:

3. connection_init_wait_timeout:

4. schema:

5. keep_alive:

6. keep_alive_interval:

7. debug:

8. protocols: