Back to top

hikari.api.shard

Provides an interface for gateway shard implementations to conform to.

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.
"""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 V8 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 and connected."""

    @property
    @abc.abstractmethod
    def shard_count(self) -> int:
        """Return the total number of shards expected in the entire application."""

    @abc.abstractmethod
    async def get_user_id(self) -> snowflakes.Snowflake:
        """Return the user ID.

        If the shard has not connected fully yet, this should wait until the ID
        is set before returning.

        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.

        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
    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
    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`.
        """
#  
@typing.final
class GatewayCompression(builtins.str, hikari.internal.enums.Enum):
View Source 
@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."""

Types of gateway compression that may be supported.

Variables and properties
#  PAYLOAD_ZLIB_STREAM

Payload compression using ZLIB.

#  TRANSPORT_ZLIB_STREAM

Transport compression using ZLIB.

#  name: str

Return the name of the enum member as a str.

#  value

Return the value of the enum member.

Methods
#  def __init__(cls, value: Any):
View Source 
    def __call__(cls, value: typing.Any) -> typing.Any:
        """Cast a value to the enum, returning the raw value that was passed if value not found."""
        try:
            return cls._value_to_member_map_[value]
        except KeyError:
            # If we can't find the value, just return what got casted in
            return value

Cast a value to the enum, returning the raw value that was passed if value not found.

#  def capitalize(self, /):

Return a capitalized version of the string.

More specifically, make the first character have upper case and the rest lower case.

#  def casefold(self, /):

Return a version of the string suitable for caseless comparisons.

#  def center(self, width, fillchar=' ', /):

Return a centered string of length width.

Padding is done using the specified fill character (default is a space).

#  def count(unknown):

S.count(sub[, start[, end]]) -> int

Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.

#  def encode(self, /, encoding='utf-8', errors='strict'):

Encode the string using the codec registered for encoding.

encoding The encoding in which to encode the string. errors The error handling scheme to use for encoding errors. The default is 'strict' meaning that encoding errors raise a UnicodeEncodeError. Other possible values are 'ignore', 'replace' and 'xmlcharrefreplace' as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.

#  def endswith(unknown):

S.endswith(suffix[, start[, end]]) -> bool

Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.

#  def expandtabs(self, /, tabsize=8):

Return a copy where all tab characters are expanded using spaces.

If tabsize is not given, a tab size of 8 characters is assumed.

#  def find(unknown):

S.find(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

#  def format(unknown):

S.format(args, *kwargs) -> str

Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces ('{' and '}').

#  def format_map(unknown):

S.format_map(mapping) -> str

Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces ('{' and '}').

#  def index(unknown):

S.index(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

#  def isalnum(self, /):

Return True if the string is an alpha-numeric string, False otherwise.

A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.

#  def isalpha(self, /):

Return True if the string is an alphabetic string, False otherwise.

A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.

#  def isascii(self, /):

Return True if all characters in the string are ASCII, False otherwise.

ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.

#  def isdecimal(self, /):

Return True if the string is a decimal string, False otherwise.

A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.

#  def isdigit(self, /):

Return True if the string is a digit string, False otherwise.

A string is a digit string if all characters in the string are digits and there is at least one character in the string.

#  def isidentifier(self, /):

Return True if the string is a valid Python identifier, False otherwise.

Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as "def" or "class".

#  def islower(self, /):

Return True if the string is a lowercase string, False otherwise.

A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.

#  def isnumeric(self, /):

Return True if the string is a numeric string, False otherwise.

A string is numeric if all characters in the string are numeric and there is at least one character in the string.

#  def isprintable(self, /):

Return True if the string is printable, False otherwise.

A string is printable if all of its characters are considered printable in repr() or if it is empty.

#  def isspace(self, /):

Return True if the string is a whitespace string, False otherwise.

A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.

#  def istitle(self, /):

Return True if the string is a title-cased string, False otherwise.

In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.

#  def isupper(self, /):

Return True if the string is an uppercase string, False otherwise.

A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.

#  def join(self, iterable, /):

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: '.'.join(['ab', 'pq', 'rs']) -> 'ab.pq.rs'

#  def ljust(self, width, fillchar=' ', /):

Return a left-justified string of length width.

Padding is done using the specified fill character (default is a space).

#  def lower(self, /):

Return a copy of the string converted to lowercase.

#  def lstrip(self, chars=None, /):

Return a copy of the string with leading whitespace removed.

If chars is given and not None, remove characters in chars instead.

#  def maketrans(unknown):

Return a translation table usable for str.translate().

If there is only one argument, it must be a dictionary mapping Unicode ordinals (integers) or characters to Unicode ordinals, strings or None. Character keys will be then converted to ordinals. If there are two arguments, they must be strings of equal length, and in the resulting dictionary, each character in x will be mapped to the character at the same position in y. If there is a third argument, it must be a string, whose characters will be mapped to None in the result.

#  def partition(self, sep, /):

Partition the string into three parts using the given separator.

This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing the original string and two empty strings.

#  def removeprefix(self, prefix, /):

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

#  def removesuffix(self, suffix, /):

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.

#  def replace(self, old, new, count=-1, /):

Return a copy with all occurrences of substring old replaced by new.

count Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.

If the optional argument count is given, only the first count occurrences are replaced.

#  def rfind(unknown):

S.rfind(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

#  def rindex(unknown):

S.rindex(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

#  def rjust(self, width, fillchar=' ', /):

Return a right-justified string of length width.

Padding is done using the specified fill character (default is a space).

#  def rpartition(self, sep, /):

Partition the string into three parts using the given separator.

This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing two empty strings and the original string.

#  def rsplit(self, /, sep=None, maxsplit=-1):

Return a list of the words in the string, using sep as the delimiter string.

sep The delimiter according which to split the string. None (the default value) means split according to any whitespace, and discard empty strings from the result. maxsplit Maximum number of splits to do. -1 (the default value) means no limit.

Splits are done starting at the end of the string and working to the front.

#  def rstrip(self, chars=None, /):

Return a copy of the string with trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

#  def split(self, /, sep=None, maxsplit=-1):

Return a list of the words in the string, using sep as the delimiter string.

sep The delimiter according which to split the string. None (the default value) means split according to any whitespace, and discard empty strings from the result. maxsplit Maximum number of splits to do. -1 (the default value) means no limit.

#  def splitlines(self, /, keepends=False):

Return a list of the lines in the string, breaking at line boundaries.

Line breaks are not included in the resulting list unless keepends is given and true.

#  def startswith(unknown):

S.startswith(prefix[, start[, end]]) -> bool

Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.

#  def strip(self, chars=None, /):

Return a copy of the string with leading and trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

#  def swapcase(self, /):

Convert uppercase characters to lowercase and lowercase characters to uppercase.

#  def title(self, /):

Return a version of the string where each word is titlecased.

More specifically, words start with uppercased characters and all remaining cased characters have lower case.

#  def translate(self, table, /):

Replace each character in the string using the given translation table.

table Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.

The table must implement lookup/indexing via __getitem__, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.

#  def upper(self, /):

Return a copy of the string converted to uppercase.

#  def zfill(self, width, /):

Pad a numeric string with zeros on the left, to fill a field of the given width.

The string is never truncated.

#  
@typing.final
class GatewayDataFormat(builtins.str, hikari.internal.enums.Enum):
View Source 
@typing.final
class GatewayDataFormat(str, enums.Enum):
    """Format of inbound gateway payloads."""

    JSON = "json"
    """Javascript serialized object notation."""
    ETF = "etf"
    """Erlang transmission format."""

Format of inbound gateway payloads.

Variables and properties
#  ETF

Erlang transmission format.

#  JSON

Javascript serialized object notation.

#  name: str

Return the name of the enum member as a str.

#  value

Return the value of the enum member.

Methods
#  def __init__(cls, value: Any):
View Source 
    def __call__(cls, value: typing.Any) -> typing.Any:
        """Cast a value to the enum, returning the raw value that was passed if value not found."""
        try:
            return cls._value_to_member_map_[value]
        except KeyError:
            # If we can't find the value, just return what got casted in
            return value

Cast a value to the enum, returning the raw value that was passed if value not found.

#  def capitalize(self, /):

Return a capitalized version of the string.

More specifically, make the first character have upper case and the rest lower case.

#  def casefold(self, /):

Return a version of the string suitable for caseless comparisons.

#  def center(self, width, fillchar=' ', /):

Return a centered string of length width.

Padding is done using the specified fill character (default is a space).

#  def count(unknown):

S.count(sub[, start[, end]]) -> int

Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.

#  def encode(self, /, encoding='utf-8', errors='strict'):

Encode the string using the codec registered for encoding.

encoding The encoding in which to encode the string. errors The error handling scheme to use for encoding errors. The default is 'strict' meaning that encoding errors raise a UnicodeEncodeError. Other possible values are 'ignore', 'replace' and 'xmlcharrefreplace' as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.

#  def endswith(unknown):

S.endswith(suffix[, start[, end]]) -> bool

Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.

#  def expandtabs(self, /, tabsize=8):

Return a copy where all tab characters are expanded using spaces.

If tabsize is not given, a tab size of 8 characters is assumed.

#  def find(unknown):

S.find(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

#  def format(unknown):

S.format(args, *kwargs) -> str

Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces ('{' and '}').

#  def format_map(unknown):

S.format_map(mapping) -> str

Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces ('{' and '}').

#  def index(unknown):

S.index(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

#  def isalnum(self, /):

Return True if the string is an alpha-numeric string, False otherwise.

A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.

#  def isalpha(self, /):

Return True if the string is an alphabetic string, False otherwise.

A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.

#  def isascii(self, /):

Return True if all characters in the string are ASCII, False otherwise.

ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.

#  def isdecimal(self, /):

Return True if the string is a decimal string, False otherwise.

A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.

#  def isdigit(self, /):

Return True if the string is a digit string, False otherwise.

A string is a digit string if all characters in the string are digits and there is at least one character in the string.

#  def isidentifier(self, /):

Return True if the string is a valid Python identifier, False otherwise.

Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as "def" or "class".

#  def islower(self, /):

Return True if the string is a lowercase string, False otherwise.

A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.

#  def isnumeric(self, /):

Return True if the string is a numeric string, False otherwise.

A string is numeric if all characters in the string are numeric and there is at least one character in the string.

#  def isprintable(self, /):

Return True if the string is printable, False otherwise.

A string is printable if all of its characters are considered printable in repr() or if it is empty.

#  def isspace(self, /):

Return True if the string is a whitespace string, False otherwise.

A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.

#  def istitle(self, /):

Return True if the string is a title-cased string, False otherwise.

In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.

#  def isupper(self, /):

Return True if the string is an uppercase string, False otherwise.

A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.

#  def join(self, iterable, /):

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: '.'.join(['ab', 'pq', 'rs']) -> 'ab.pq.rs'

#  def ljust(self, width, fillchar=' ', /):

Return a left-justified string of length width.

Padding is done using the specified fill character (default is a space).

#  def lower(self, /):

Return a copy of the string converted to lowercase.

#  def lstrip(self, chars=None, /):

Return a copy of the string with leading whitespace removed.

If chars is given and not None, remove characters in chars instead.

#  def maketrans(unknown):

Return a translation table usable for str.translate().

If there is only one argument, it must be a dictionary mapping Unicode ordinals (integers) or characters to Unicode ordinals, strings or None. Character keys will be then converted to ordinals. If there are two arguments, they must be strings of equal length, and in the resulting dictionary, each character in x will be mapped to the character at the same position in y. If there is a third argument, it must be a string, whose characters will be mapped to None in the result.

#  def partition(self, sep, /):

Partition the string into three parts using the given separator.

This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing the original string and two empty strings.

#  def removeprefix(self, prefix, /):

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

#  def removesuffix(self, suffix, /):

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.

#  def replace(self, old, new, count=-1, /):

Return a copy with all occurrences of substring old replaced by new.

count Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.

If the optional argument count is given, only the first count occurrences are replaced.

#  def rfind(unknown):

S.rfind(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

#  def rindex(unknown):

S.rindex(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

#  def rjust(self, width, fillchar=' ', /):

Return a right-justified string of length width.

Padding is done using the specified fill character (default is a space).

#  def rpartition(self, sep, /):

Partition the string into three parts using the given separator.

This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing two empty strings and the original string.

#  def rsplit(self, /, sep=None, maxsplit=-1):

Return a list of the words in the string, using sep as the delimiter string.

sep The delimiter according which to split the string. None (the default value) means split according to any whitespace, and discard empty strings from the result. maxsplit Maximum number of splits to do. -1 (the default value) means no limit.

Splits are done starting at the end of the string and working to the front.

#  def rstrip(self, chars=None, /):

Return a copy of the string with trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

#  def split(self, /, sep=None, maxsplit=-1):

Return a list of the words in the string, using sep as the delimiter string.

sep The delimiter according which to split the string. None (the default value) means split according to any whitespace, and discard empty strings from the result. maxsplit Maximum number of splits to do. -1 (the default value) means no limit.

#  def splitlines(self, /, keepends=False):

Return a list of the lines in the string, breaking at line boundaries.

Line breaks are not included in the resulting list unless keepends is given and true.

#  def startswith(unknown):

S.startswith(prefix[, start[, end]]) -> bool

Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.

#  def strip(self, chars=None, /):

Return a copy of the string with leading and trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

#  def swapcase(self, /):

Convert uppercase characters to lowercase and lowercase characters to uppercase.

#  def title(self, /):

Return a version of the string where each word is titlecased.

More specifically, words start with uppercased characters and all remaining cased characters have lower case.

#  def translate(self, table, /):

Replace each character in the string using the given translation table.

table Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.

The table must implement lookup/indexing via __getitem__, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.

#  def upper(self, /):

Return a copy of the string converted to uppercase.

#  def zfill(self, width, /):

Pad a numeric string with zeros on the left, to fill a field of the given width.

The string is never truncated.

#  class GatewayShard(abc.ABC):
View Source 
class GatewayShard(abc.ABC):
    """Interface for a definition of a V8 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 and connected."""

    @property
    @abc.abstractmethod
    def shard_count(self) -> int:
        """Return the total number of shards expected in the entire application."""

    @abc.abstractmethod
    async def get_user_id(self) -> snowflakes.Snowflake:
        """Return the user ID.

        If the shard has not connected fully yet, this should wait until the ID
        is set before returning.

        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.

        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
    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
    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`.
        """

Interface for a definition of a V8 compatible websocket gateway.

Each instance should represent a single shard.

Variables and properties
#  heartbeat_latency: float

Shard's most recent heartbeat latency.

If the information is not yet available, then this will be float('nan') instead.

#  id: int

0-based shard ID for this shard.

#  intents: hikari.intents.Intents

Intents set on this shard.

#  is_alive: bool

Whether the shard is alive and connected.

#  shard_count: int

Return the total number of shards expected in the entire application.

Methods
#  
@abc.abstractmethod
async def close(self) -> None:
View Source 
    @abc.abstractmethod
    async def close(self) -> None:
        """Close the websocket if it is connected."""

Close the websocket if it is connected.

#  
@abc.abstractmethod
async def get_user_id(self) -> hikari.snowflakes.Snowflake:
View Source 
    @abc.abstractmethod
    async def get_user_id(self) -> snowflakes.Snowflake:
        """Return the user ID.

        If the shard has not connected fully yet, this should wait until the ID
        is set before returning.

        Returns
        -------
        hikari.snowflakes.Snowflake
            The user ID for the application user.
        """

Return the user ID.

If the shard has not connected fully yet, this should wait until the ID is set before returning.

Returns
#  
@abc.abstractmethod
async def join(self) -> None:
View Source 
    @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.
        """

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 request_guild_members(
   self,
   guild: Union[hikari.guilds.PartialGuild, hikari.snowflakes.Snowflake, int],
   *,
   include_presences: Union[bool, hikari.undefined.UndefinedType] = UNDEFINED,
   query: str = '',
   limit: int = 0,
   users: Union[Sequence[Union[hikari.users.User, hikari.snowflakes.Snowflake, int]], hikari.undefined.UndefinedType] = UNDEFINED,
   nonce: Union[str, hikari.undefined.UndefinedType] = UNDEFINED
) -> None:
View Source 
    @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: 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`.
        """

Request for a guild chunk.

Note: To request the full list of members, set query to "" (empty string) and limit to 0.

Parameters
Other Parameters
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.
#  
@abc.abstractmethod
async def start(self) -> None:
View Source 
    @abc.abstractmethod
    async def start(self) -> None:
        """Start the shard, wait for it to become ready."""

Start the shard, wait for it to become ready.

#  
@abc.abstractmethod
async def update_presence(
   self,
   *,
   idle_since: Union[datetime.datetime, hikari.undefined.UndefinedType, NoneType] = UNDEFINED,
   afk: Union[bool, hikari.undefined.UndefinedType] = UNDEFINED,
   activity: Union[hikari.presences.Activity, hikari.undefined.UndefinedType, NoneType] = UNDEFINED,
   status: Union[hikari.presences.Status, hikari.undefined.UndefinedType] = UNDEFINED
) -> None:
View Source 
    @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.

        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.
        """

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
#  
@abc.abstractmethod
async def update_voice_state(
   self,
   guild: Union[hikari.guilds.PartialGuild, hikari.snowflakes.Snowflake, int],
   channel: Union[hikari.channels.GuildVoiceChannel, hikari.snowflakes.Snowflake, int, NoneType],
   *,
   self_mute: Union[bool, hikari.undefined.UndefinedType] = UNDEFINED,
   self_deaf: Union[bool, hikari.undefined.UndefinedType] = UNDEFINED
) -> None:
View Source 
    @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 : 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.
        """

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.