Skip to content

Gateway

Internal API

These classes are for internal use only.

Users should interact with Client and other public API classes instead.

Client

gateway

logger = logging.getLogger('scurrypy.gateway') module-attribute

MIN_BACKOFF = 5 module-attribute

GatewayMetrics dataclass

heartbeat_rrt: float instance-attribute

EventQueueProtocol

Bases: Protocol


              flowchart TD
              scurrypy.core.gateway.EventQueueProtocol[EventQueueProtocol]

              

              click scurrypy.core.gateway.EventQueueProtocol href "" "scurrypy.core.gateway.EventQueueProtocol"

Internal contract for the GatewayClient's event queue used by the Client. Meant for testing.

put async

get async

GatewayClientProtocol

Bases: Protocol


              flowchart TD
              scurrypy.core.gateway.GatewayClientProtocol[GatewayClientProtocol]

              

              click scurrypy.core.gateway.GatewayClientProtocol href "" "scurrypy.core.gateway.GatewayClientProtocol"

Internal contract for the GatewayClient used by the Client. Meant for testing.

event_queue: EventQueueProtocol instance-attribute

shard_id: int instance-attribute

total_shards: int instance-attribute

start async

close_ws async

GatewayClient

Bases: GatewayClientProtocol


              flowchart TD
              scurrypy.core.gateway.GatewayClient[GatewayClient]
              scurrypy.core.gateway.GatewayClientProtocol[GatewayClientProtocol]

                              scurrypy.core.gateway.GatewayClientProtocol --> scurrypy.core.gateway.GatewayClient
                


              click scurrypy.core.gateway.GatewayClient href "" "scurrypy.core.gateway.GatewayClient"
              click scurrypy.core.gateway.GatewayClientProtocol href "" "scurrypy.core.gateway.GatewayClientProtocol"

shard_id: int = 0 instance-attribute

total_shards: int = 0 instance-attribute

seq: int | None = None instance-attribute

session_id: str | None = None instance-attribute

allow_resume: bool = False instance-attribute

reconnect_immediately: bool = False instance-attribute

backoff: int = MIN_BACKOFF instance-attribute

metrics: GatewayMetrics = GatewayMetrics(0.0, 0.0) instance-attribute

heartbeat_task: asyncio.Task[Any] | None = None instance-attribute

heartbeat_interval: int = 0 instance-attribute

event_queue: asyncio.Queue[tuple[str, JSON]] = asyncio.Queue() instance-attribute

base_url = 'wss://gateway.discord.gg' instance-attribute

url_params = '?v=10&encoding=json' instance-attribute

ws: websockets.ClientConnection property

wait_reconnect async

Sleep for exponentially increasing time between reconnects.

start async

Start this websocket's connection.

Parameters:

  • token (str) –

    the bot's token

  • intents (int) –

    the bot's intents

connect_ws async

Connect to Discord's Gateway (websocket).

send async

Send data through the websocket.

Parameters:

  • data (dict) –

    data to send

receive async

Receive data through the websocket.

Returns:

  • dict –

    websocket data

heartbeat async

Heartbeat task to keep connection alive.

identify async

Send an IDENTIFY payload to handshake for bot.

Parameters:

  • token (str) –

    the bot's token

  • intents (int) –

    the bot's intents

Raises:

  • ConnectionError –

    invalid intents

resume async

Send a RESUME payload to resume a connection.

Parameters:

  • token (str) –

    the bot's token

close_ws async

Close the websocket connection if one is still open and cancels heartbeat.

Events

gateway_events

SessionStartLimit dataclass

Bases: Event, DataModel


              flowchart TD
              scurrypy.events.gateway_events.SessionStartLimit[SessionStartLimit]
              scurrypy.events.base_event.Event[Event]
              scurrypy.core.model.DataModel[DataModel]

                              scurrypy.events.base_event.Event --> scurrypy.events.gateway_events.SessionStartLimit
                                scurrypy.core.model.DataModel --> scurrypy.events.base_event.Event
                

                scurrypy.core.model.DataModel --> scurrypy.events.gateway_events.SessionStartLimit
                


              click scurrypy.events.gateway_events.SessionStartLimit href "" "scurrypy.events.gateway_events.SessionStartLimit"
              click scurrypy.events.base_event.Event href "" "scurrypy.events.base_event.Event"
              click scurrypy.core.model.DataModel href "" "scurrypy.core.model.DataModel"

Represents the Session Start Limit object.

total: int instance-attribute

Total remaining shards.

remaining: int instance-attribute

Shards left to connect.

reset_after: int instance-attribute

When remaining resets from now (in ms).

max_concurrency: int instance-attribute

How many shards can be started at once.

GatewayEvent dataclass

Bases: Event, DataModel


              flowchart TD
              scurrypy.events.gateway_events.GatewayEvent[GatewayEvent]
              scurrypy.events.base_event.Event[Event]
              scurrypy.core.model.DataModel[DataModel]

                              scurrypy.events.base_event.Event --> scurrypy.events.gateway_events.GatewayEvent
                                scurrypy.core.model.DataModel --> scurrypy.events.base_event.Event
                

                scurrypy.core.model.DataModel --> scurrypy.events.gateway_events.GatewayEvent
                


              click scurrypy.events.gateway_events.GatewayEvent href "" "scurrypy.events.gateway_events.GatewayEvent"
              click scurrypy.events.base_event.Event href "" "scurrypy.events.base_event.Event"
              click scurrypy.core.model.DataModel href "" "scurrypy.core.model.DataModel"

Represents the Gateway Event object.

url: str instance-attribute

Gateway URL to connect.

shards: int instance-attribute

Recommended shard count for the aaplication.

session_start_limit: SessionStartLimit instance-attribute

Session start info.