# -*- 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
@typing.final
[docs]class GatewayCompression(str, enums.Enum):
"""Types of gateway compression that may be supported."""
[docs] TRANSPORT_ZLIB_STREAM = "transport_zlib_stream"
"""Transport compression using ZLIB."""
[docs] PAYLOAD_ZLIB_STREAM = "payload_zlib_stream"
"""Payload compression using ZLIB."""
[docs]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
[docs] 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
[docs] def id(self) -> int:
"""0-based shard ID for this shard."""
@property
@abc.abstractmethod
[docs] def intents(self) -> intents_.Intents:
"""Intents set on this shard."""
@property
@abc.abstractmethod
[docs] def is_alive(self) -> bool:
"""Whether the shard is alive."""
@property
@abc.abstractmethod
[docs] def is_connected(self) -> bool:
"""Whether the shard is connected."""
@property
@abc.abstractmethod
[docs] def shard_count(self) -> int:
"""Return the total number of shards expected in the entire application."""
@abc.abstractmethod
[docs] def get_user_id(self) -> snowflakes.Snowflake:
"""Return the user ID.
Returns
-------
hikari.snowflakes.Snowflake
The user ID for the application user.
"""
@abc.abstractmethod
[docs] async def close(self) -> None:
"""Close the websocket if it is connected."""
@abc.abstractmethod
[docs] 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
[docs] async def start(self) -> None:
"""Start the shard, wait for it to become ready."""
@abc.abstractmethod
[docs] 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.
Other Parameters
----------------
idle_since : hikari.undefined.UndefinedNoneOr[datetime.datetime]
The datetime that the user started being idle. If undefined, this
will not be changed.
afk : hikari.undefined.UndefinedOr[bool]
If `True`, the user is marked as AFK. If `False`,
the user is marked as being active. If undefined, this will not be
changed.
activity : hikari.undefined.UndefinedNoneOr[hikari.presences.Activity]
The activity to appear to be playing. If undefined, this will not be
changed.
status : hikari.undefined.UndefinedOr[hikari.presences.Status]
The web status to show. If undefined, this will not be changed.
"""
@abc.abstractmethod
[docs] 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 : hikari.snowflakes.SnowflakeishOr[hikari.guilds.PartialGuild]
The guild or guild ID to update the voice state for.
channel : typing.Optional[hikari.snowflakes.SnowflakeishOr[hikari.channels.GuildVoiceChannel]]
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 : bool
If specified and `True`, the bot will mute itself in that
voice channel. If `False`, then it will unmute itself.
self_deaf : bool
If specified and `True`, the bot will deafen itself in that
voice channel. If `False`, then it will undeafen itself.
"""
@abc.abstractmethod
[docs] 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 : hikari.guilds.Guild
The guild to request chunk for.
Other Parameters
----------------
include_presences : hikari.undefined.UndefinedOr[bool]
If provided, whether to request presences.
query : str
If not `""`, request the members which username starts with the string.
limit : int
Maximum number of members to send matching the query.
users : hikari.undefined.UndefinedOr[hikari.snowflakes.SnowflakeishSequence[hikari.users.User]]
If provided, the users to request for.
nonce : hikari.undefined.UndefinedOr[str]
If provided, the nonce to be sent with guild chunks.
Raises
------
ValueError
When 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 `GUILD_MEMBERS` or when trying to
request the full list of members without `GUILD_PRESENCES`.
"""