from __future__ import annotations import asyncio import logging from typing import TYPE_CHECKING, Any, Callable, Coroutine, Literal, Optional, TypeVar, Union, cast, overload from typing_extensions import ParamSpec import aiohttp from .errors import RevoltError from .channel import (DMChannel, GroupDMChannel, SavedMessageChannel, TextChannel, VoiceChannel, channel_factory) from .http import HttpClient from .invite import Invite from .message import Message from .state import State from .utils import Missing, Ulid from .websocket import WebsocketHandler from .emoji import Emoji from .server import Server from .user import User try: import ujson as json except ImportError: import json if TYPE_CHECKING: from .channel import Channel from .file import File from .types import ApiInfo import revolt __all__ = ("Client",) logger: logging.Logger = logging.getLogger("revolt") P = ParamSpec("P") R = TypeVar("R") class Client: """The client for interacting with revolt Parameters ----------- session: :class:`aiohttp.ClientSession` The aiohttp session to use for http request and the websocket token: :class:`str` The bots token api_url: :class:`str` The api url for the revolt instance you are connecting to, by default it uses the offical instance hosted at revolt.chat max_messages: :class:`int` The max amount of messages stored in the cache, by default this is 5k bot: :class:`bool` Denotes whether the account used is a bot account or user account, by default this it assumes a bot account """ def __init__(self, session: aiohttp.ClientSession, token: str, *, api_url: str = "https://api.revolt.chat", max_messages: int = 5000, bot: bool = True): self.session: aiohttp.ClientSession = session self.token: str = token self.api_url: str = api_url self.max_messages: int = max_messages self.bot: bool = bot self.api_info: ApiInfo self.http: HttpClient self.state: State self.websocket: WebsocketHandler self.temp_listeners: dict[str, list[tuple[Callable[..., bool], asyncio.Future[Any]]]] = {} self.listeners: dict[str, list[Callable[..., Coroutine[Any, Any, Any]]]] = {} super().__init__() def dispatch(self, event: str, *args: Any) -> None: """Dispatch an event, this is typically used for testing and internals. Parameters ---------- event: class:`str` The name of the event to dispatch, not including `on_` args: :class:`Any` The arguments passed to the event """ if temp_listeners := self.temp_listeners.get(event, None): for check, future in temp_listeners: if check(*args): if len(args) == 1: future.set_result(args[0]) else: future.set_result(args) self.temp_listeners[event] = [(c, f) for c, f in temp_listeners if not f.done()] for listener in self.listeners.get(event, []): asyncio.create_task(listener(*args)) if func := getattr(self, f"on_{event}", None): asyncio.create_task(func(*args)) async def get_api_info(self) -> ApiInfo: async with self.session.get(self.api_url) as resp: text = await resp.text() try: return json.loads(text) except: raise RevoltError(f"Cant fetch api info:\n{text}") async def start(self, *, reconnect: bool = True) -> None: """Starts the client""" api_info = await self.get_api_info() self.api_info = api_info self.http = HttpClient(self.session, self.token, self.api_url, self.api_info, self.bot) self.state = State(self.http, api_info, self.max_messages) self.websocket = WebsocketHandler(self.session, self.token, api_info["ws"], self.dispatch, self.state) await self.websocket.start(reconnect) async def stop(self) -> None: await self.websocket.websocket.close() def get_user(self, id: str) -> User: """Gets a user from the cache Parameters ----------- id: :class:`str` The id of the user Returns -------- :class:`User` The user """ return self.state.get_user(id) def get_channel(self, id: str) -> Channel: """Gets a channel from the cache Parameters ----------- id: :class:`str` The id of the channel Returns -------- :class:`Channel` The channel """ return self.state.get_channel(id) def get_server(self, id: str) -> Server: """Gets a server from the cache Parameters ----------- id: :class:`str` The id of the server Returns -------- :class:`Server` The server """ return self.state.get_server(id) async def wait_for(self, event: str, *, check: Optional[Callable[..., bool]] = None, timeout: Optional[float] = None) -> Any: """Waits for an event Parameters ----------- event: :class:`str` The name of the event to wait for, without the `on_` check: Optional[Callable[..., :class:`bool`]] A function that says what event to wait_for, this function takes the same parameters as the event you are waiting for and should return a bool saying if that is the event you want timeout: Optional[:class:`float`] Time in seconds to wait for the event. By default it waits forever Raises ------- asyncio.TimeoutError If timeout is provided and it was reached Returns -------- Any The parameters of the event """ if not check: check = lambda *_: True future = asyncio.get_running_loop().create_future() self.temp_listeners.setdefault(event, []).append((check, future)) return await asyncio.wait_for(future, timeout) def listen(self, name: str | None = None) -> Callable[[Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]]: """Registers a listener for an event, multiple listeners can be registered to the same event without conflict Parameters ----------- name: Optional[:class:`str`] The name of the event to register this under, this defaults to the function's name """ def inner(func: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, Coroutine[Any, Any, R]]: nonlocal name if not name: if not func.__name__.startswith("on_"): raise RevoltError("listener name must begin with `on_`") name = func.__name__[3:] self.listeners.setdefault(name, []).append(func) return func return inner @overload def remove_listener(self, func: Callable[P, Coroutine[Any, Any, R]], *, event: str = ...) -> Callable[..., Coroutine[Any, Any, R]] | None: ... @overload def remove_listener(self, func: Callable[P, Coroutine[Any, Any, Any]], *, event: None = ...) -> None: ... def remove_listener(self, func: Callable[P, Coroutine[Any, Any, R]], *, event: str | None = None) -> Callable[..., Coroutine[Any, Any, R]] | None: """Removes a listener registered, if the `event` parameter is passed, the listener will only be removed from that event, this can be used if the same listener is registed to multiple events at once. Parameters ----------- func: Callable The function for the listener to be removed event: Optional[:class:`str`] The name of the event to remove this from, passing `None` will make this remove the listener from all events this is registered under """ if event is None: for listeners in self.listeners.values(): try: listeners.remove(func) except ValueError: pass else: try: self.listeners[event].remove(func) return func except ValueError: pass @property def user(self) -> User: """:class:`User` the user corrasponding to the client""" user = self.websocket.user assert user return user @property def users(self) -> list[User]: """list[:class:`User`] All users the client can see""" return list(self.state.users.values()) @property def servers(self) -> list[Server]: """list[:class:'Server'] All servers the client can see""" return list(self.state.servers.values()) @property def global_emojis(self) -> list[Emoji]: return self.state.global_emojis async def fetch_user(self, user_id: str) -> User: """Fetchs a user Parameters ----------- user_id: :class:`str` The id of the user you are fetching Returns -------- :class:`User` The user with the matching id """ payload = await self.http.fetch_user(user_id) return User(payload, self.state) async def fetch_dm_channels(self) -> list[Union[DMChannel, GroupDMChannel]]: """Fetchs all dm channels the client has made Returns -------- list[Union[:class:`DMChanel`, :class:`GroupDMChannel`]] A list of :class:`DMChannel` or :class`GroupDMChannel` """ channel_payloads = await self.http.fetch_dm_channels() return cast(list[Union[DMChannel, GroupDMChannel]], [channel_factory(payload, self.state) for payload in channel_payloads]) async def fetch_channel(self, channel_id: str) -> Union[DMChannel, GroupDMChannel, SavedMessageChannel, TextChannel, VoiceChannel]: """Fetches a channel Parameters ----------- channel_id: :class:`str` The id of the channel Returns -------- Union[:class:`DMChannel`, :class:`GroupDMChannel`, :class:`SavedMessageChannel`, :class:`TextChannel`, :class:`VoiceChannel`] The channel with the matching id """ payload = await self.http.fetch_channel(channel_id) return channel_factory(payload, self.state) async def fetch_server(self, server_id: str) -> Server: """Fetchs a server Parameters ----------- server_id: :class:`str` The id of the server you are fetching Returns -------- :class:`Server` The server with the matching id """ payload = await self.http.fetch_server(server_id) return Server(payload, self.state) async def fetch_invite(self, code: str) -> Invite: """Fetchs an invite Parameters ----------- code: :class:`str` The code of the invite you are fetching Returns -------- :class:`Invite` The invite with the matching code """ payload = await self.http.fetch_invite(code) return Invite(payload, code, self.state) def get_message(self, message_id: str) -> Message: """Gets a message from the cache Parameters ----------- message_id: :class:`str` The id of the message you are getting Returns -------- :class:`Message` The message with the matching id Raises ------- LookupError This raises if the message is not found in the cache """ for message in self.state.messages: if message.id == message_id: return message raise LookupError async def edit_self(self, **kwargs: Any) -> None: """Edits the client's own user Parameters ----------- avatar: Optional[:class:`File`] The avatar to change to, passing in ``None`` will remove the avatar """ if kwargs.get("avatar", Missing) is None: del kwargs["avatar"] remove = ["Avatar"] else: remove = None await self.state.http.edit_self(remove, kwargs) async def edit_status(self, **kwargs: Any) -> None: """Edits the client's own status Parameters ----------- presence: :class:`PresenceType` The presence to change to text: Optional[:class:`str`] The text to change the status to, passing in ``None`` will remove the status """ if kwargs.get("text", Missing) is None: del kwargs["text"] remove = ["StatusText"] else: remove = None if presence := kwargs.get("presence"): kwargs["presence"] = presence.value await self.state.http.edit_self(remove, {"status": kwargs}) async def edit_profile(self, **kwargs: Any) -> None: """Edits the client's own profile Parameters ----------- content: Optional[:class:`str`] The new content for the profile, passing in ``None`` will remove the profile content background: Optional[:class:`File`] The new background for the profile, passing in ``None`` will remove the profile background """ remove: list[str] = [] if kwargs.get("content", Missing) is None: del kwargs["content"] remove.append("ProfileContent") if kwargs.get("background", Missing) is None: del kwargs["background"] remove.append("ProfileBackground") await self.state.http.edit_self(remove, {"profile": kwargs}) async def fetch_emoji(self, emoji_id: str) -> Emoji: """Fetches an emoji Parameters ----------- emoji_id: str The id of the emoji Returns -------- :class:`Emoji` The emoji with the corrasponding id """ emoji = await self.state.http.fetch_emoji(emoji_id) return Emoji(emoji, self.state) async def upload_file(self, file: File, tag: Literal['attachments', 'avatars', 'backgrounds', 'icons', 'banners', 'emojis']) -> Ulid: """Uploads a file to revolt Parameters ----------- file: :class:`File` The file to upload tag: :class:`str` The type of file to upload, this should a string of either `'attachments'`, `'avatars'`, `'backgrounds'`, `'icons'`, `'banners'` or `'emojis'` Returns -------- :class:`Ulid` The id of the file that was uploaded """ asset = await self.http.upload_file(file, tag) ulid = Ulid() ulid.id = asset["id"] return ulid # events async def on_ready(self) -> None: pass async def on_message(self, message: revolt.Message) -> None: pass async def on_raw_message_update(self, payload: revolt.types.MessageUpdateEventPayload) -> None: pass async def on_message_update(self, before: revolt.Message, after: revolt.Message) -> None: pass async def on_raw_message_delete(self, payload: revolt.types.MessageDeleteEventPayload) -> None: pass async def on_message_delete(self, message: revolt.Message) -> None: pass async def on_channel_create(self, channel: revolt.Channel) -> None: pass async def on_channel_update(self, before: revolt.Channel, after: revolt.Channel) -> None: pass async def on_channel_delete(self, channel: revolt.Channel) -> None: pass async def on_typing_start(self, channel: revolt.Channel, user: revolt.User) -> None: pass async def on_typing_stop(self, channel: revolt.Channel, user: revolt.User) -> None: pass async def on_server_update(self, before: revolt.Server, after: revolt.Server) -> None: pass async def on_server_delete(self, server: revolt.Server) -> None: pass async def on_server_join(self, server: revolt.Server) -> None: pass async def on_member_update(self, before: revolt.Member, after: revolt.Member) -> None: pass async def on_member_join(self, member: revolt.Member) -> None: pass async def on_member_leave(self, member: revolt.Member) -> None: pass async def on_role_create(self, role: revolt.Role) -> None: pass async def on_role_update(self, before: revolt.Role, after: revolt.Role) -> None: pass async def on_role_delete(self, role: revolt.Role) -> None: pass async def on_user_update(self, before: revolt.User, after: revolt.User) -> None: pass async def on_user_relationship_update(self, user: revolt.User, before: revolt.RelationshipType, after: revolt.RelationshipType) -> None: pass async def on_raw_reaction_add(self, payload: revolt.types.MessageReactEventPayload) -> None: pass async def on_reaction_add(self, message: revolt.Message, user: revolt.User, emoji_id: str) -> None: pass async def on_raw_reaction_remove(self, payload: revolt.types.MessageUnreactEventPayload) -> None: pass async def on_reaction_remove(self, message: revolt.Message, user: revolt.User, emoji_id: str) -> None: pass async def on_raw_reaction_clear(self, payload: revolt.types.MessageRemoveReactionEventPayload) -> None: pass async def on_reaction_clear(self, message: revolt.Message, user: revolt.User, emoji_id: str) -> None: pass async def raw_bulk_message_delete(self, payload: revolt.types.BulkMessageDeleteEventPayload) -> None: pass async def bulk_message_delete(self, messages: list[revolt.Message]) -> None: pass