BaseGraphQLTransportWSHandler

Constructor:

def __init__(
    self,
    schema: BaseSchema,
    debug: bool,
    connection_init_wait_timeout: timedelta,
) -> None:
  ...

Methods:

• get_context

  Return the operations context.

  Signature:

  def get_context(self) -> Any:
    ...

• get_root_value

  Return the schemas root value.

  Signature:

  def get_root_value(self) -> Any:
    ...

• send_json

  Send the data JSON encoded to the WebSocket client.

  Signature:

  def send_json(self, data: dict) -> None:
    ...

  Parameters:

  1. data:

     Type

     dict

• close

  Close the WebSocket with the passed code and reason.

  Signature:

  def close(self, code: int, reason: str) -> None:
    ...

  Parameters:

  1. code:

     Type

     int

  2. reason:

     Type

     str

• handle_request

  Handle the request this instance was created for.

  Signature:

  def handle_request(self) -> Any:
    ...

• handle

  Signature:

  def handle(self) -> Any:
    ...

• shutdown

  Signature:

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

• on_request_accepted

  Signature:

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

• handle_connection_init_timeout

  Signature:

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

• handle_task_exception

  Signature:

  def handle_task_exception(self, error: Exception) -> None:
    ...

  Parameters:

  1. error:

     Type

     Exception

• handle_message

  Signature:

  def handle_message(self, message: dict) -> None:
    ...

  Parameters:

  1. message:

     Type

     dict

• handle_connection_init

  Signature:

  def handle_connection_init(self, message: ConnectionInitMessage) -> None:
    ...

  Parameters:

  1. message:

     Type

     ConnectionInitMessage

• handle_ping

  Signature:

  def handle_ping(self, message: PingMessage) -> None:
    ...

  Parameters:

  1. message:

     Type

     PingMessage

• handle_pong

  Signature:

  def handle_pong(self, message: PongMessage) -> None:
    ...

  Parameters:

  1. message:

     Type

     PongMessage

• handle_subscribe

  Signature:

  def handle_subscribe(self, message: SubscribeMessage) -> None:
    ...

  Parameters:

  1. message:

     Type

     SubscribeMessage

• operation_task

  The operation task’s top level method. Cleans-up and de-registers the operation once it is done.

  Signature:

  def operation_task(self, result_source: AsyncGenerator, operation: Operation) -> None:
    ...

  Parameters:

  1. result_source:

     Type

     AsyncGenerator

  2. operation:

     Type

     Operation

• handle_async_results

  Signature:

  def handle_async_results(self, result_source: AsyncGenerator, operation: Operation) -> None:
    ...

  Parameters:

  1. result_source:

     Type

     AsyncGenerator

  2. operation:

     Type

     Operation

• forget_id

  Signature:

  def forget_id(self, id: str) -> None:
    ...

  Parameters:

  1. id:

     Type

     str

• handle_complete

  Signature:

  def handle_complete(self, message: CompleteMessage) -> None:
    ...

  Parameters:

  1. message:

     Type

     CompleteMessage

• handle_invalid_message

  Signature:

  def handle_invalid_message(self, error_message: str) -> None:
    ...

  Parameters:

  1. error_message:

     Type

     str

• send_message

  Signature:

  def send_message(self, message: GraphQLTransportMessage) -> None:
    ...

  Parameters:

  1. message:

     Type

     GraphQLTransportMessage

• cleanup_operation

  Signature:

  def cleanup_operation(self, operation_id: str) -> None:
    ...

  Parameters:

  1. operation_id:

     Type

     str

• reap_completed_tasks

  Await tasks that have completed.

  Signature:

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

Attributes:

1. task_logger:

   Type

   logging.Logger

2. schema:

3. debug:

4. connection_init_wait_timeout:

5. connection_init_timeout_task:

   Type

   asyncio.Task | None

6. connection_init_received:

7. connection_acknowledged:

8. connection_timed_out:

9. operations:

   Type

   Dict[str, Operation]

10. completed_tasks:

    Type

    List[asyncio.Task]

11. connection_params:

    Type

    Dict[str, Any] | None