Source code for discohook.channel

from typing import TYPE_CHECKING, Any, Dict, List, Optional

import aiohttp

from .embed import Embed
from .emoji import PartialEmoji
from .enums import ChannelType
from .file import File
from .message import Message
from .models import AllowedMentions, MessageReference
from .params import _SendingPayload
from .poll import Poll
from .view import View

if TYPE_CHECKING:
    from .client import Client


[docs]class PartialChannel: """ Represents a partial discord channel object. Parameters ---------- channel_id: str The channel's ID. guild_id: str | None The guild's ID. client: :class:`Client` The client that the channel belongs to. """ def __init__( self, client: "Client", channel_id: str, guild_id: Optional[str] = None ): self.client = client self.id: str = channel_id self.guild_id = guild_id def __eq__(self, other): return self.id == other.id @property def mention(self) -> str: """ Returns the channel-mentionable string. Returns ------- :class:`str` """ return f"<#{self.id}>"
[docs] async def send( self, content: Optional[str] = None, *, embed: Optional[Embed] = None, embeds: Optional[List[Embed]] = None, view: Optional[View] = None, tts: Optional[bool] = False, file: Optional[File] = None, files: Optional[List[File]] = None, poll: Optional[Poll] = None, allowed_mentions: Optional[AllowedMentions] = None, message_reference: Optional[MessageReference] = None, ): """ Sends a message to the channel. Parameters ---------- content: Optional[:class:`str`] The content of the message. embed: Optional[:class:`Embed`] The embed to send with the message. embeds: Optional[List[:class:`Embed`]] A list of embeds to send with the message. view: Optional[:class:`View`] The view to send with the message. tts: Optional[:class:`bool`] Whether the message should be sent with text-to-speech. file: Optional[File] A file to send with the message. files: Optional[List[File]] A list of files to send with the message. poll: Optional[:class:`Poll`] The poll to send with the message. allowed_mentions: Optional[:class:`AllowedMentions`] The allowed mentions for the message. message_reference: Optional[:class:`MessageReference`] The message reference for the message. """ if view: self.client.load_view(view) payload = _SendingPayload( content=content, embed=embed, embeds=embeds, view=view, tts=tts, file=file, files=files, poll=poll, allowed_mentions=allowed_mentions, message_reference=message_reference, ) resp = await self.client.http.send_message(self.id, payload.to_form()) data = await resp.json() return Message(self.client, data)
[docs] async def edit( self, *, name: Optional[str] = None, kind: Optional[ChannelType] = None, position: Optional[int] = None, topic: Optional[str] = None, nsfw: Optional[bool] = None, rate_limit_per_user: Optional[int] = None, bitrate: Optional[int] = None, user_limit: Optional[int] = None, permission_overwrites: Optional[List[Dict[str, Any]]] = None, parent_id: Optional[str] = None, rtc_region: Optional[str] = None, video_quality_mode: Optional[int] = None, default_auto_archive_duration: Optional[int] = None, flags: Optional[int] = None, available_tags: Optional[List[Dict[str, Any]]] = None, icon: Optional[str] = None, default_reaction_emoji: Optional[PartialEmoji] = None, default_thread_rate_limit_per_user: Optional[int] = None, default_sort_order: Optional[int] = None, default_forum_layout: Optional[int] = None, ) -> "Channel": """ Edits all kinds of channels. Parameters ---------- name: Optional[:class:`str`] The new name of the channel. kind: Optional[:class:`ChannelType`] The new type of the channel. position: Optional[:class:`int`] The new position of the channel. topic: Optional[:class:`str`] The new topic of the channel. nsfw: Optional[:class:`bool`] Whether the channel should be marked as nsfw. rate_limit_per_user: Optional[:class:`int`] The duration of the slowmode in seconds. Must be between 0 and 21600. Applies to text and forum channels. bitrate: Optional[:class:`int`] The new bitrate of the channel. Must be between 8000 and 96000. Applies to voice channels. user_limit: Optional[:class:`int`] The new user limit of the channel. Must be between 0 and 99. Applies to voice channels. permission_overwrites: Optional[List[:class:`dict`]] A list of permission overwrites to apply to the channel. Applies to all channel types. parent_id: Optional[:class:`str`] The id of the parent category to move the channel to. Applies to all channel types. rtc_region: Optional[:class:`str`] The new region of the channel. Applies to voice channels. video_quality_mode: Optional[:class:`int`] The new video quality mode of the channel. Applies to voice channels. default_auto_archive_duration: Optional[:class:`int`] The new default auto archive duration of the channel. Applies to text and forum channels. flags: Optional[:class:`int`] The new flags of the channel. Applies to all channel types. available_tags: Optional[List[:class:`dict`]] The new available tags of the channel. Applies to text and forum channels. icon: Optional[:class:`str`] The new icon of the channel. Applies to Group DMs. Must be a base64 encoded string. default_reaction_emoji: Optional[:class:`PartialEmoji`] The new default reaction emoji of the channel. Applies to text and forum channels. default_thread_rate_limit_per_user: Optional[:class:`int`] The new default thread rate limit per user of the channel. Applies to text and forum channels. default_sort_order: Optional[:class:`int`] The new default sort order of the channel. Applies to text and forum channels. default_forum_layout: Optional[:class:`int`] The new default forum layout of the channel. Applies to text and forum channels. Returns ------- :class:`Channel` The edited channel. """ payload = {} if name: payload["name"] = name if kind: payload["type"] = kind if position: payload["position"] = position if topic: payload["topic"] = topic if nsfw: payload["nsfw"] = nsfw if rate_limit_per_user: payload["rate_limit_per_user"] = rate_limit_per_user if bitrate: payload["bitrate"] = bitrate if user_limit: payload["user_limit"] = user_limit if permission_overwrites: payload["permission_overwrites"] = permission_overwrites if parent_id: payload["parent_id"] = parent_id if rtc_region: payload["rtc_region"] = rtc_region if video_quality_mode: payload["video_quality_mode"] = video_quality_mode if default_auto_archive_duration: payload["default_auto_archive_duration"] = default_auto_archive_duration if flags: payload["flags"] = flags if available_tags: payload["available_tags"] = available_tags if icon: payload["icon"] = icon if default_reaction_emoji: payload["default_reaction_emoji"] = default_reaction_emoji.to_dict() if default_thread_rate_limit_per_user: payload["default_thread_rate_limit_per_user"] = ( default_thread_rate_limit_per_user ) if default_sort_order: payload["default_sort_order"] = default_sort_order if default_forum_layout: payload["default_forum_layout"] = default_forum_layout resp = await self.client.http.edit_channel(self.id, payload) data = await resp.json() return Channel(self.client, data)
[docs] async def fetch_message(self, message_id: str) -> Optional[Message]: """ Fetches a message from the channel. Parameters ---------- message_id: :class:`str` The id of the message to fetch. Returns ------- :class:`Message` The fetched message. """ resp = await self.client.http.fetch_channel_message(self.id, message_id) data = await resp.json() return Message(self.client, data)
[docs] async def fetch_messages( self, limit: int = 50, *, before: Optional[str] = None, after: Optional[str] = None, around: Optional[str] = None, ) -> List[Message]: """ Fetches messages from the channel. Parameters ---------- limit: Optional[:class:`int`] The maximum amount of messages to fetch. before: Optional[:class:`str`] The id of the message to fetch before. after: Optional[:class:`str`] The id of the message to fetch after. around: Optional[:class:`str`] The id of the message to fetch around. Returns ------- List[:class:`Message`] The fetched messages. """ params = {"limit": limit} if before: params["before"] = before if after: params["after"] = after if around: params["around"] = around resp = await self.client.http.fetch_channel_messages(self.id, params=params) data = await resp.json() return [Message(self.client, msg) for msg in data]
[docs] async def purge( self, limit: int = 50, *, before: Optional[str] = None, after: Optional[str] = None, around: Optional[str] = None, ) -> List[Message]: """ Deletes messages from the channel in bulk. Parameters ---------- limit: Optional[:class:`int`] The maximum amount of messages to delete. before: Optional[:class:`str`] The id of the message to delete before. after: Optional[:class:`str`] The id of the message to delete after. around: Optional[:class:`str`] The id of the message to delete around. Returns ------- List[:class:`Message`] The deleted messages. """ messages = await self.fetch_messages( limit=limit, before=before, after=after, around=around ) ids = [msg.id for msg in messages] if len(ids) < 2: await self.client.http.delete_channel_message(self.id, ids[0]) return messages await self.client.http.delete_channel_messages(self.id, {"messages": ids}) return messages
[docs] async def delete(self): await self.client.http.delete_channel(self.id)
[docs] async def crosspost(self, message_id: str): resp = await self.client.http.crosspost_channel_message(self.id, message_id) data = await resp.json() return Message(self.client, data)
[docs] async def start_thread( self, name: str, *, auto_archive_duration: int = 60, invitable: bool = True, rate_limit_per_user: int = 0, reason: Optional[str] = None, ) -> "Channel": """ Creates a thread from the channel. Parameters ---------- name: :class:`str` The name of the thread. auto_archive_duration: Optional[:class:`int`] The duration in minutes to automatically archive the thread after recent activity. invitable: Optional[:class:`bool`] Whether non-moderators can add other non-moderators to the thread. rate_limit_per_user: Optional[:class:`int`] Amount of seconds a user has to wait before sending another message (0-21600) reason Returns ------- """ payload = { "name": name, "auto_archive_duration": auto_archive_duration, "type": ChannelType.private_thread, "invitable": invitable, "rate_limit_per_user": rate_limit_per_user, } resp = await self.client.http.start_thread_without_message( self.id, payload, reason=reason ) data = await resp.json() return Channel(self.client, data)
[docs]class Channel(PartialChannel): """ Represents a discord channel object. Attributes ---------- id: :class:`str` The id of the channel. type: Optional[:class:`int`] The type of the channel. guild_id: Optional[:class:`str`] The id of the guild the channel belongs to. position: Optional[:class:`int`] The position of the channel. permission_overwrites: Optional[List[:class:`dict`]] A list of permission overwrites for the channel. name: Optional[:class:`str`] The name of the channel. topic: Optional[:class:`str`] The topic of the channel. nsfw: Optional[:class:`bool`] Whether the channel is nsfw. last_message_id: Optional[:class:`str`] The id of the last message sent in the channel. bitrate: Optional[:class:`int`] The bitrate of the channel if it is a voice channel. user_limit: Optional[:class:`int`] The user limit of the channel if it is a voice channel. rate_limit_per_user: Optional[:class:`int`] The rate limit per user of the channel if it is a text channel. recipients: Optional[List[:class:`dict`]] A list of recipients of the channel if it is a dm channel. icon: Optional[:class:`str`] The icon of the channel if it is a dm channel. owner_id: Optional[:class:`str`] The id of the owner of the channel if it is a dm channel. application_id: Optional[:class:`str`] The id of the application of the channel if it is a group dm channel. parent_id: Optional[:class:`str`] The id of the parent category of the channel. last_pin_timestamp: Optional[:class:`str`] The timestamp of the last pinned message in the channel. rtc_region: Optional[:class:`str`] The rtc region of the channel. video_quality_mode: Optional[:class:`int`] The video quality mode of the channel. message_count: Optional[:class:`int`] The message count of the channel. member_count: Optional[:class:`int`] The member count of the channel. thread_metadata: Optional[:class:`dict`] The thread metadata of the channel. member: Optional[:class:`dict`] The member of the channel. Appears in thread channels. default_auto_archive_duration: Optional[:class:`int`] The default auto archive duration of the channel. Appears in thread channels. permissions: Optional[:class:`str`] The permissions of the channel. flags: Optional[:class:`int`] The flags of the channel. total_message_sent: Optional[:class:`int`] The total message sent of the channel. available_tags: Optional[List[:class:`str`]] A list of available tags of the channel. Appears in thread channels. default_reaction_emoji: Optional[:class:`dict`] The default reaction emoji of the channel. Appears in thread channels. default_thread_rate_limit_per_user: Optional[:class:`int`] The default rate limit per user of the channel. Appears in thread channels. default_sort_order: Optional[:class:`int`] The default sort order of the channel.Appears in forum channels. default_forum_layout: Optional[:class:`int`] The default channel layout of the channel.Appears in forum channels. """ def __init__(self, client: "Client", data: dict): super().__init__(client, data["id"], data.get("guild_id")) self.type = data.get("type") self.guild_id = data.get("guild_id") self.position = data.get("position") self.permission_overwrites = data.get("permission_overwrites") self.name = data.get("name") self.topic = data.get("topic") self.nsfw = data.get("nsfw") self.last_message_id = data.get("last_message_id") self.bitrate = data.get("bitrate") self.user_limit = data.get("user_limit") self.rate_limit_per_user = data.get("rate_limit_per_user") self.recipients = data.get("recipients") self.icon = data.get("icon") self.owner_id = data.get("owner_id") self.application_id = data.get("application_id") self.managed = data.get("managed") self.parent_id = data.get("parent_id") self.last_pin_timestamp = data.get("last_pin_timestamp") self.rtc_region = data.get("rtc_region") self.video_quality_mode = data.get("video_quality_mode") self.message_count = data.get("message_count") self.member_count = data.get("member_count") self.thread_metadata = data.get("thread_metadata") self.member = data.get("member") self.default_auto_archive_duration = data.get("default_auto_archive_duration") self.permissions = data.get("permissions") self.flags = data.get("flags") self.total_message_sent = data.get("total_message_sent") self.available_tags = data.get("available_tags") self.applied_tags = data.get("applied_tags") self.default_reaction_emoji = data.get("default_reaction_emoji") self.default_thread_rate_limit_per_user = data.get( "default_thread_rate_limit_per_user" ) self.default_sort_order = data.get("default_sort_order") self.default_forum_layout = data.get("default_forum_layout")
[docs] @classmethod async def from_response(cls, client: "Client", response: aiohttp.ClientResponse): return cls(client, await response.json())
[docs] @classmethod def from_dict(cls, client: "Client", data: dict): return cls(client, data)