hikari.events.shard_events
Events relating to specific shards events.
View Source
# -*- coding: utf-8 -*- # cython: language_level=3 # Copyright (c) 2020 Nekokatt # Copyright (c) 2021-present davfsa # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """Events relating to specific shards events.""" from __future__ import annotations __all__: typing.Sequence[str] = ( "ShardEvent", "ShardPayloadEvent", "ShardStateEvent", "ShardConnectedEvent", "ShardDisconnectedEvent", "ShardReadyEvent", "ShardResumedEvent", "MemberChunkEvent", ) import abc import typing import attr from hikari.events import base_events from hikari.internal import attr_extensions from hikari.internal import collections if typing.TYPE_CHECKING: from hikari import applications from hikari import guilds from hikari import presences as presences_ from hikari import snowflakes from hikari import traits from hikari import users from hikari.api import shard as gateway_shard class ShardEvent(base_events.Event, abc.ABC): """Base class for any event that was shard-specific.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def shard(self) -> gateway_shard.GatewayShard: """Shard that received this event.""" @attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardPayloadEvent(ShardEvent): """Event fired for most shard events with their raw payload. .. note:: This will only be dispatched for real dispatch events received from Discord and not artificial events like the `ShardStateEvent` events. """ app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. name: str = attr.field() """Name of the received event.""" payload: typing.Mapping[str, typing.Any] = attr.field() """The raw payload for this event.""" class ShardStateEvent(ShardEvent, abc.ABC): """Base class for any event concerning the state/connectivity of a shard. This currently wraps connection/disconnection/ready/resumed events only. """ __slots__: typing.Sequence[str] = () @attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardConnectedEvent(ShardStateEvent): """Event fired when a shard connects.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. @attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardDisconnectedEvent(ShardStateEvent): """Event fired when a shard disconnects.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. @attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardReadyEvent(ShardStateEvent): """Event fired when a shard declares it is ready.""" shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. actual_gateway_version: int = attr.field(repr=True) """Actual gateway version being used.""" session_id: str = attr.field(repr=True) """ID for this session.""" my_user: users.OwnUser = attr.field(repr=True) """User for the current bot account this connection is authenticated with.""" unavailable_guilds: typing.Sequence[snowflakes.Snowflake] = attr.field(repr=False) """Sequence of the IDs for all guilds this bot is currently in. All guilds will start off "unavailable" and should become available after a few seconds of connecting one-by-one. """ application_id: snowflakes.Snowflake = attr.field(repr=True) """ID of the application this ready event is for.""" application_flags: applications.ApplicationFlags = attr.field(repr=True) """Flags of the application this ready event is for.""" @property def app(self) -> traits.RESTAware: # <<inherited docstring from Event>>. return self.my_user.app @attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardResumedEvent(ShardStateEvent): """Event fired when a shard resumes an existing session.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. @attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class MemberChunkEvent(ShardEvent, typing.Sequence["guilds.Member"]): """Event fired when a member chunk payload is received on a gateway shard.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. guild_id: snowflakes.Snowflake = attr.field(repr=True) # <<docstring inherited from ShardEvent>>. members: typing.Mapping[snowflakes.Snowflake, guilds.Member] = attr.field(repr=False) """Mapping of user IDs to the objects of the members in this chunk.""" chunk_index: int = attr.field(repr=True) """Zero-indexed position of this within the queued up chunks for this request.""" chunk_count: int = attr.field(repr=True) """Total number of expected chunks for the request this is associated with.""" not_found: typing.Sequence[snowflakes.Snowflake] = attr.field(repr=True) """Sequence of the snowflakes that were not found while making this request. This is only applicable when user IDs are specified while making the member request the chunk is associated with. """ presences: typing.Mapping[snowflakes.Snowflake, presences_.MemberPresence] = attr.field(repr=False) """Mapping of user IDs to found member presence objects. This will be empty if no presences are found or `include_presences` is not passed as `True` while requesting the member chunks. """ nonce: typing.Optional[str] = attr.field(repr=True) """String nonce used to identify the request member chunks are associated with. This is the nonce value passed while requesting member chunks or `None` if there was no nonce passed. """ @typing.overload def __getitem__(self, index_or_slice: int, /) -> guilds.Member: ... @typing.overload def __getitem__(self, index_or_slice: slice, /) -> typing.Sequence[guilds.Member]: ... def __getitem__( self, index_or_slice: typing.Union[int, slice], / ) -> typing.Union[guilds.Member, typing.Sequence[guilds.Member]]: return collections.get_index_or_slice(self.members, index_or_slice) def __iter__(self) -> typing.Iterator[guilds.Member]: return iter(self.members.values()) def __len__(self) -> int: return len(self.members)
View Source
@attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class MemberChunkEvent(ShardEvent, typing.Sequence["guilds.Member"]): """Event fired when a member chunk payload is received on a gateway shard.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. guild_id: snowflakes.Snowflake = attr.field(repr=True) # <<docstring inherited from ShardEvent>>. members: typing.Mapping[snowflakes.Snowflake, guilds.Member] = attr.field(repr=False) """Mapping of user IDs to the objects of the members in this chunk.""" chunk_index: int = attr.field(repr=True) """Zero-indexed position of this within the queued up chunks for this request.""" chunk_count: int = attr.field(repr=True) """Total number of expected chunks for the request this is associated with.""" not_found: typing.Sequence[snowflakes.Snowflake] = attr.field(repr=True) """Sequence of the snowflakes that were not found while making this request. This is only applicable when user IDs are specified while making the member request the chunk is associated with. """ presences: typing.Mapping[snowflakes.Snowflake, presences_.MemberPresence] = attr.field(repr=False) """Mapping of user IDs to found member presence objects. This will be empty if no presences are found or `include_presences` is not passed as `True` while requesting the member chunks. """ nonce: typing.Optional[str] = attr.field(repr=True) """String nonce used to identify the request member chunks are associated with. This is the nonce value passed while requesting member chunks or `None` if there was no nonce passed. """ @typing.overload def __getitem__(self, index_or_slice: int, /) -> guilds.Member: ... @typing.overload def __getitem__(self, index_or_slice: slice, /) -> typing.Sequence[guilds.Member]: ... def __getitem__( self, index_or_slice: typing.Union[int, slice], / ) -> typing.Union[guilds.Member, typing.Sequence[guilds.Member]]: return collections.get_index_or_slice(self.members, index_or_slice) def __iter__(self) -> typing.Iterator[guilds.Member]: return iter(self.members.values()) def __len__(self) -> int: return len(self.members)
Event fired when a member chunk payload is received on a gateway shard.
Variables and properties
Total number of expected chunks for the request this is associated with.
Zero-indexed position of this within the queued up chunks for this request.
Mapping of user IDs to the objects of the members in this chunk.
String nonce used to identify the request member chunks are associated with.
This is the nonce value passed while requesting member chunks or None
if there was no nonce passed.
Sequence of the snowflakes that were not found while making this request.
This is only applicable when user IDs are specified while making the member request the chunk is associated with.
Mapping of user IDs to found member presence objects.
This will be empty if no presences are found or include_presences
is not passed as True
while requesting the member chunks.
Methods
self,
*,
app: hikari.traits.RESTAware,
shard: hikari.api.shard.GatewayShard,
guild_id: hikari.snowflakes.Snowflake,
members: Mapping[hikari.snowflakes.Snowflake, hikari.guilds.Member],
chunk_index: int,
chunk_count: int,
not_found: Sequence[hikari.snowflakes.Snowflake],
presences: Mapping[hikari.snowflakes.Snowflake, hikari.presences.MemberPresence],
nonce: Optional[str]
):
View Source
def __init__(self, *, app, shard, guild_id, members, chunk_index, chunk_count, not_found, presences, nonce): self.app = app self.shard = shard self.guild_id = guild_id self.members = members self.chunk_index = chunk_index self.chunk_count = chunk_count self.not_found = not_found self.presences = presences self.nonce = nonce
Method generated by attrs for class MemberChunkEvent.
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
def count(self, value): 'S.count(value) -> integer -- return number of occurrences of value' return sum(1 for v in self if v is value or v == value)
S.count(value) -> integer -- return number of occurrences of value
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
def index(self, value, start=0, stop=None): '''S.index(value, [start, [stop]]) -> integer -- return first index of value. Raises ValueError if the value is not present. Supporting start and stop arguments is optional, but recommended. ''' if start is not None and start < 0: start = max(len(self) + start, 0) if stop is not None and stop < 0: stop += len(self) i = start while stop is None or i < stop: try: v = self[i] if v is value or v == value: return i except IndexError: break i += 1 raise ValueError
S.index(value, [start, [stop]]) -> integer -- return first index of value. Raises ValueError if the value is not present.
Supporting start and stop arguments is optional, but recommended.
View Source
@attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardConnectedEvent(ShardStateEvent): """Event fired when a shard connects.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>.
Event fired when a shard connects.
Variables and properties
Methods
View Source
def __init__(self, *, app, shard): self.app = app self.shard = shard
Method generated by attrs for class ShardConnectedEvent.
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
@attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardDisconnectedEvent(ShardStateEvent): """Event fired when a shard disconnects.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>.
Event fired when a shard disconnects.
Variables and properties
Methods
View Source
def __init__(self, *, app, shard): self.app = app self.shard = shard
Method generated by attrs for class ShardDisconnectedEvent.
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
class ShardEvent(base_events.Event, abc.ABC): """Base class for any event that was shard-specific.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def shard(self) -> gateway_shard.GatewayShard: """Shard that received this event."""
Base class for any event that was shard-specific.
Variables and properties
App instance for this application.
Shard that received this event.
Methods
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
@attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardPayloadEvent(ShardEvent): """Event fired for most shard events with their raw payload. .. note:: This will only be dispatched for real dispatch events received from Discord and not artificial events like the `ShardStateEvent` events. """ app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. name: str = attr.field() """Name of the received event.""" payload: typing.Mapping[str, typing.Any] = attr.field() """The raw payload for this event."""
Event fired for most shard events with their raw payload.
Note: This will only be dispatched for real dispatch events received from Discord and not artificial events like the ShardStateEvent
events.
Variables and properties
Name of the received event.
The raw payload for this event.
Methods
self,
*,
app: hikari.traits.RESTAware,
shard: hikari.api.shard.GatewayShard,
name: str,
payload: Mapping[str, Any]
):
View Source
def __init__(self, *, app, shard, name, payload): self.app = app self.shard = shard self.name = name self.payload = payload
Method generated by attrs for class ShardPayloadEvent.
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
@attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardReadyEvent(ShardStateEvent): """Event fired when a shard declares it is ready.""" shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>. actual_gateway_version: int = attr.field(repr=True) """Actual gateway version being used.""" session_id: str = attr.field(repr=True) """ID for this session.""" my_user: users.OwnUser = attr.field(repr=True) """User for the current bot account this connection is authenticated with.""" unavailable_guilds: typing.Sequence[snowflakes.Snowflake] = attr.field(repr=False) """Sequence of the IDs for all guilds this bot is currently in. All guilds will start off "unavailable" and should become available after a few seconds of connecting one-by-one. """ application_id: snowflakes.Snowflake = attr.field(repr=True) """ID of the application this ready event is for.""" application_flags: applications.ApplicationFlags = attr.field(repr=True) """Flags of the application this ready event is for.""" @property def app(self) -> traits.RESTAware: # <<inherited docstring from Event>>. return self.my_user.app
Event fired when a shard declares it is ready.
Variables and properties
Actual gateway version being used.
App instance for this application.
Flags of the application this ready event is for.
ID of the application this ready event is for.
User for the current bot account this connection is authenticated with.
ID for this session.
Methods
self,
*,
shard: hikari.api.shard.GatewayShard,
actual_gateway_version: int,
session_id: str,
my_user: hikari.users.OwnUser,
unavailable_guilds: Sequence[hikari.snowflakes.Snowflake],
application_id: hikari.snowflakes.Snowflake,
application_flags: hikari.applications.ApplicationFlags
):
View Source
def __init__(self, *, shard, actual_gateway_version, session_id, my_user, unavailable_guilds, application_id, application_flags): self.shard = shard self.actual_gateway_version = actual_gateway_version self.session_id = session_id self.my_user = my_user self.unavailable_guilds = unavailable_guilds self.application_id = application_id self.application_flags = application_flags
Method generated by attrs for class ShardReadyEvent.
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
@attr_extensions.with_copy @attr.define(kw_only=True, weakref_slot=False) class ShardResumedEvent(ShardStateEvent): """Event fired when a shard resumes an existing session.""" app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<inherited docstring from Event>>. shard: gateway_shard.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True}) # <<docstring inherited from ShardEvent>>.
Event fired when a shard resumes an existing session.
Variables and properties
Methods
View Source
def __init__(self, *, app, shard): self.app = app self.shard = shard
Method generated by attrs for class ShardResumedEvent.
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.
View Source
class ShardStateEvent(ShardEvent, abc.ABC): """Base class for any event concerning the state/connectivity of a shard. This currently wraps connection/disconnection/ready/resumed events only. """ __slots__: typing.Sequence[str] = ()
Base class for any event concerning the state/connectivity of a shard.
This currently wraps connection/disconnection/ready/resumed events only.
Variables and properties
App instance for this application.
Shard that received this event.
Methods
View Source
@classmethod def bitmask(cls) -> int: """Bitmask for this event.""" return cls.__bitmask
Bitmask for this event.
View Source
@classmethod def dispatches(cls) -> typing.Sequence[typing.Type[Event]]: """Sequence of the event classes this event is dispatched as.""" return cls.__dispatches
Sequence of the event classes this event is dispatched as.