hikari/api/shard.py
# -*- 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.
"""Provides an interface for gateway shard implementations to conform to."""
from __future__ import annotations
__all__: typing.Sequence[str] = ("GatewayDataFormat", "GatewayCompression", "GatewayShard")
import abc
import typing
from hikari import undefined
from hikari.internal import enums
if typing.TYPE_CHECKING:
import datetime
from hikari import channels
from hikari import guilds
from hikari import intents as intents_
from hikari import presences
from hikari import snowflakes
from hikari import users as users_
@typing.final
class GatewayDataFormat(str, enums.Enum):
"""Format of inbound gateway payloads."""
JSON = "json"
"""Javascript serialized object notation."""
ETF = "etf"
"""Erlang transmission format."""
@typing.final
class GatewayCompression(str, enums.Enum):
"""Types of gateway compression that may be supported."""
TRANSPORT_ZLIB_STREAM = "transport_zlib_stream"
"""Transport compression using ZLIB."""
PAYLOAD_ZLIB_STREAM = "payload_zlib_stream"
"""Payload compression using ZLIB."""
class GatewayShard(abc.ABC):
"""Interface for a definition of a v10 compatible websocket gateway.
Each instance should represent a single shard.
"""
__slots__: typing.Sequence[str] = ()
@property
@abc.abstractmethod
def heartbeat_latency(self) -> float:
"""Shard's most recent heartbeat latency.
If the information is not yet available, then this will be
`float('nan')` instead.
"""
@property
@abc.abstractmethod
def id(self) -> int:
"""0-based shard ID for this shard."""
@property
@abc.abstractmethod
def intents(self) -> intents_.Intents:
"""Intents set on this shard."""
@property
@abc.abstractmethod
def is_alive(self) -> bool:
"""Whether the shard is alive."""
@property
@abc.abstractmethod
def is_connected(self) -> bool:
"""Whether the shard is connected."""
@property
@abc.abstractmethod
def shard_count(self) -> int:
"""Return the total number of shards expected in the entire application."""
@abc.abstractmethod
def get_user_id(self) -> snowflakes.Snowflake:
"""Return the user ID.
Returns
-------
hikari.snowflakes.Snowflake
The user ID for the application user.
"""
@abc.abstractmethod
async def close(self) -> None:
"""Close the websocket if it is connected."""
@abc.abstractmethod
async def join(self) -> None:
"""Wait indefinitely until the websocket closes.
This can be placed in a task and cancelled without affecting the
websocket runtime itself.
"""
@abc.abstractmethod
async def start(self) -> None:
"""Start the shard, wait for it to become ready."""
@abc.abstractmethod
async def update_presence(
self,
*,
idle_since: undefined.UndefinedNoneOr[datetime.datetime] = undefined.UNDEFINED,
afk: undefined.UndefinedOr[bool] = undefined.UNDEFINED,
activity: undefined.UndefinedNoneOr[presences.Activity] = undefined.UNDEFINED,
status: undefined.UndefinedOr[presences.Status] = undefined.UNDEFINED,
) -> None:
"""Update the presence of the shard user.
If the shard is not alive, no physical data will be sent, however,
the new presence settings will be remembered for when the shard
does connect.
Parameters
----------
idle_since
The datetime that the user started being idle. If undefined, this
will not be changed.
afk
Whether to mark the user as AFK. If undefined, this will not be
changed.
activity
The activity to appear to be playing. If undefined, this will not be
changed.
status
The web status to show. If undefined, this will not be changed.
"""
@abc.abstractmethod
async def update_voice_state(
self,
guild: snowflakes.SnowflakeishOr[guilds.PartialGuild],
channel: typing.Optional[snowflakes.SnowflakeishOr[channels.GuildVoiceChannel]],
*,
self_mute: undefined.UndefinedOr[bool] = undefined.UNDEFINED,
self_deaf: undefined.UndefinedOr[bool] = undefined.UNDEFINED,
) -> None:
"""Update the voice state for this shard in a given guild.
Parameters
----------
guild
The guild or guild ID to update the voice state for.
channel
The channel or channel ID to update the voice state for. If [`None`][]
then the bot will leave the voice channel that it is in for the
given guild.
self_mute
If specified and [`True`][], the bot will mute itself in that
voice channel. If [`False`][], then it will unmute itself.
self_deaf
If specified and [`True`][], the bot will deafen itself in that
voice channel. If [`False`][], then it will undeafen itself.
"""
@abc.abstractmethod
async def request_guild_members(
self,
guild: snowflakes.SnowflakeishOr[guilds.PartialGuild],
*,
include_presences: undefined.UndefinedOr[bool] = undefined.UNDEFINED,
query: str = "",
limit: int = 0,
users: undefined.UndefinedOr[snowflakes.SnowflakeishSequence[users_.User]] = undefined.UNDEFINED,
nonce: undefined.UndefinedOr[str] = undefined.UNDEFINED,
) -> None:
"""Request for a guild chunk.
!!! note
To request the full list of members, set `query` to `""` (empty
string) and `limit` to `0`.
Parameters
----------
guild
The guild to request chunk for.
include_presences
If provided, whether to request presences.
query
If not `""`, request the members which username starts with the string.
limit
Maximum number of members to send matching the query.
users
If provided, the users to request for.
nonce
If provided, the nonce to be sent with guild chunks.
Raises
------
ValueError
If trying to specify `users` with `query`/`limit`, if `limit` is not between
0 and 100, both inclusive or if `users` length is over 100.
hikari.errors.MissingIntentError
When trying to request presences without the [`hikari.intents.Intents.GUILD_MEMBERS`][] or when trying to
request the full list of members without [`hikari.intents.Intents.GUILD_PRESENCES`][].
"""