# -*- 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. """Core app interface for application implementations.""" from __future__ import annotations __all__: typing.Sequence[str] = ( "CacheAware", "EventManagerAware", "EntityFactoryAware", "EventFactoryAware", "ExecutorAware", "GatewayBotAware", "IntentsAware", "NetworkSettingsAware", "RESTAware", "RESTBotAware", "Runnable", "InteractionServerAware", "ShardAware", "VoiceAware", ) import abc import typing from hikari import presences from hikari import undefined from hikari.internal import fast_protocol if typing.TYPE_CHECKING: import datetime from concurrent import futures from typing_extensions import Self from hikari import channels from hikari import guilds from hikari import intents as intents_ from hikari import snowflakes from hikari import users as users_ from hikari.api import cache as cache_ from hikari.api import config from hikari.api import entity_factory as entity_factory_ from hikari.api import event_factory as event_factory_ from hikari.api import event_manager as event_manager_ from hikari.api import interaction_server as interaction_server_ from hikari.api import rest as rest_ from hikari.api import shard as gateway_shard from hikari.api import voice as voice_ @typing.runtime_checkable class NetworkSettingsAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for any component aware of network settings.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def http_settings(self) -> config.HTTPSettings: """HTTP settings in use by this component.""" raise NotImplementedError @property @abc.abstractmethod def proxy_settings(self) -> config.ProxySettings: """Proxy settings in use by this component.""" raise NotImplementedError @typing.runtime_checkable class EventManagerAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for a event manager-aware object. event manager-aware components are able to manage event listeners and waiters. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def event_manager(self) -> event_manager_.EventManager: """Event manager for this object.""" raise NotImplementedError @typing.runtime_checkable class EntityFactoryAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for an entity factory-aware object. These components will be able to construct library entities. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def entity_factory(self) -> entity_factory_.EntityFactory: """Entity factory implementation for this object.""" raise NotImplementedError @typing.runtime_checkable class ExecutorAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for an executor-aware object. These components will contain an `executor` attribute that may return a [`concurrent.futures.Executor`][] or [`None`][] if the default [`asyncio`][] thread pool for the event loop is used. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def executor(self) -> typing.Optional[futures.Executor]: """Executor to use for blocking operations. This may return [`None`][] if the default [`asyncio`][] thread pool should be used instead. """ raise NotImplementedError @typing.runtime_checkable class EventFactoryAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for an event factory-aware object. These components are able to construct library events. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def event_factory(self) -> event_factory_.EventFactory: """Event factory component.""" raise NotImplementedError @typing.runtime_checkable class IntentsAware(fast_protocol.FastProtocolChecking, typing.Protocol): """A component that is aware of the application intents.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def intents(self) -> intents_.Intents: """Intents registered for the application.""" raise NotImplementedError @typing.runtime_checkable class RESTAware( EntityFactoryAware, NetworkSettingsAware, ExecutorAware, fast_protocol.FastProtocolChecking, typing.Protocol ): """Structural supertype for a REST-aware object. These are able to perform REST API calls. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def rest(self) -> rest_.RESTClient: """REST client to use for HTTP requests.""" raise NotImplementedError @typing.runtime_checkable class VoiceAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for a voice-aware object. This is an object that provides a `voice` property to allow the creation of custom voice client instances. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def voice(self) -> voice_.VoiceComponent: """Voice connection manager component for this application.""" raise NotImplementedError @typing.runtime_checkable class ShardAware( IntentsAware, NetworkSettingsAware, ExecutorAware, VoiceAware, fast_protocol.FastProtocolChecking, typing.Protocol ): """Structural supertype for a shard-aware object. These will expose a mapping of shards, the intents in use and the bot user object. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def heartbeat_latencies(self) -> typing.Mapping[int, float]: """Mapping of shard ID to heartbeat latency. Any shards that are not yet started will be `float('nan')`. """ raise NotImplementedError @property @abc.abstractmethod def heartbeat_latency(self) -> float: """Average heartbeat latency of all started shards. If no shards are started, this will return `float('nan')`. """ raise NotImplementedError @property @abc.abstractmethod def shards(self) -> typing.Mapping[int, gateway_shard.GatewayShard]: """Mapping of shards in this application instance. Each shard ID is mapped to the corresponding shard instance. If the application has not started, it is acceptable to assume the result of this call will be an empty mapping. """ raise NotImplementedError @property @abc.abstractmethod def shard_count(self) -> int: """Number of shards in the total application. This may not be the same as the size of `shards`. If the application is auto-sharded, this may be `0` until the shards are started. """ raise NotImplementedError @abc.abstractmethod def get_me(self) -> typing.Optional[users_.OwnUser]: """Return the bot user, if known. This should be available as soon as the bot has fired the [`hikari.events.lifetime_events.StartingEvent`][]. Until then, this may or may not be [`None`][]. Returns ------- typing.Optional[hikari.users.OwnUser] The bot user, if known, otherwise [`None`][]. """ raise NotImplementedError @abc.abstractmethod async def update_presence( self, *, idle_since: undefined.UndefinedNoneOr[datetime.datetime] = undefined.UNDEFINED, status: undefined.UndefinedOr[presences.Status] = undefined.UNDEFINED, activity: undefined.UndefinedNoneOr[presences.Activity] = undefined.UNDEFINED, afk: undefined.UndefinedOr[bool] = undefined.UNDEFINED, ) -> None: """Update the presence on all shards. This call will patch the presence on each shard. This means that unless you explicitly specify a parameter, the previous value will be retained. This means you do not have to track the global presence in your code. !!! note This will only send the update payloads to shards that are alive. Any shards that are not alive will cache the new presence for when they do start. !!! note If you want to set presences per shard, access the shard you wish to update (e.g. by using [`hikari.GatewayBot.shards`][]), and call [`hikari.api.shard.GatewayShard.update_presence`][] on that shard. This method is simply a facade to make performing this in bulk simpler. Parameters ---------- idle_since The datetime that the user started being idle. If undefined, this will not be changed. afk Whether to be marked 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. """ raise NotImplementedError @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 bot 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. Raises ------ RuntimeError If the guild passed isn't covered by any of the shards in this sharded client. """ @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`][]. RuntimeError If the guild passed isn't covered by any of the shards in this sharded client. """ @typing.runtime_checkable class InteractionServerAware(RESTAware, EntityFactoryAware, fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for a interaction REST server-aware object.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def interaction_server(self) -> interaction_server_.InteractionServer: """Interaction server this app is bound to.""" raise NotImplementedError @typing.runtime_checkable class CacheAware(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for a cache-aware object. Any cache-aware objects are able to access the Discord application cache. """ __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def cache(self) -> cache_.Cache: """Immutable cache implementation for this object.""" raise NotImplementedError @typing.runtime_checkable class Runnable(fast_protocol.FastProtocolChecking, typing.Protocol): """Structural super-type for an application which can be run independently.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def is_alive(self) -> bool: """Whether the application is running or not. This is useful as some functions might raise [`hikari.errors.ComponentStateConflictError`][] if this is [`False`][]. """ raise NotImplementedError @abc.abstractmethod async def close(self) -> None: """Kill the application by shutting all components down.""" @abc.abstractmethod async def join(self) -> None: """Wait indefinitely until the application closes. This can be placed in a task and cancelled without affecting the application runtime itself. Any exceptions raised by shards will be propagated to here. """ raise NotImplementedError @abc.abstractmethod def run(self) -> None: """Start the application and block until it's finished running.""" raise NotImplementedError @abc.abstractmethod async def start(self) -> None: """Start the application and then return.""" raise NotImplementedError @typing.runtime_checkable class GatewayBotAware( RESTAware, Runnable, ShardAware, EventFactoryAware, EventManagerAware, CacheAware, fast_protocol.FastProtocolChecking, typing.Protocol, ): """Structural supertype for a component that has all the gateway components.""" __slots__: typing.Sequence[str] = () @typing.runtime_checkable class RESTBotAware(InteractionServerAware, Runnable, fast_protocol.FastProtocolChecking, typing.Protocol): """Structural supertype for a component that has all the RESTful components.""" __slots__: typing.Sequence[str] = () @property @abc.abstractmethod def on_shutdown(self) -> typing.Sequence[typing.Callable[[Self], typing.Coroutine[typing.Any, typing.Any, None]]]: """Sequence of the bot's asynchronous shutdown callbacks.""" raise NotImplementedError @property @abc.abstractmethod def on_startup(self) -> typing.Sequence[typing.Callable[[Self], typing.Coroutine[typing.Any, typing.Any, None]]]: """Sequence of the bot's asynchronous startup callbacks.""" raise NotImplementedError @abc.abstractmethod def add_shutdown_callback( self, callback: typing.Callable[[RESTBotAware], typing.Coroutine[typing.Any, typing.Any, None]], / ) -> None: """Add an asynchronous callback to be called when the bot shuts down. Parameters ---------- callback The asynchronous shutdown callback to add. """ raise NotImplementedError @abc.abstractmethod def remove_shutdown_callback( self, callback: typing.Callable[[RESTBotAware], typing.Coroutine[typing.Any, typing.Any, None]], / ) -> None: """Remove an asynchronous shutdown callback from the bot. Parameters ---------- callback The shutdown callback to remove. Raises ------ ValueError If the callback was not registered. """ raise NotImplementedError @abc.abstractmethod def add_startup_callback( self, callback: typing.Callable[[RESTBotAware], typing.Coroutine[typing.Any, typing.Any, None]], / ) -> None: """Add an asynchronous callback to be called when the bot starts up. Parameters ---------- callback The asynchronous startup callback to add. """ raise NotImplementedError @abc.abstractmethod def remove_startup_callback( self, callback: typing.Callable[[RESTBotAware], typing.Coroutine[typing.Any, typing.Any, None]], / ) -> None: """Remove an asynchronous startup callback from the bot. Parameters ---------- callback The asynchronous startup callback to remove. Raises ------ ValueError If the callback was not registered. """ raise NotImplementedError