2017 lines
70 KiB
Python
2017 lines
70 KiB
Python
# SPDX-License-Identifier: MIT
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
from abc import ABC
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
List,
|
|
Mapping,
|
|
Optional,
|
|
Protocol,
|
|
Sequence,
|
|
Tuple,
|
|
TypeVar,
|
|
Union,
|
|
cast,
|
|
overload,
|
|
runtime_checkable,
|
|
)
|
|
|
|
from . import utils
|
|
from .context_managers import Typing
|
|
from .enums import (
|
|
ChannelType,
|
|
PartyType,
|
|
ThreadLayout,
|
|
ThreadSortOrder,
|
|
VideoQualityMode,
|
|
try_enum_to_int,
|
|
)
|
|
from .errors import ClientException
|
|
from .file import File
|
|
from .flags import ChannelFlags, MessageFlags
|
|
from .invite import Invite
|
|
from .mentions import AllowedMentions
|
|
from .object import Object
|
|
from .partial_emoji import PartialEmoji
|
|
from .permissions import PermissionOverwrite, Permissions
|
|
from .role import Role
|
|
from .sticker import GuildSticker, StickerItem
|
|
from .ui.action_row import components_to_dict
|
|
from .utils import _overload_with_permissions
|
|
from .voice_client import VoiceClient, VoiceProtocol
|
|
|
|
__all__ = (
|
|
"Snowflake",
|
|
"User",
|
|
"PrivateChannel",
|
|
"GuildChannel",
|
|
"Messageable",
|
|
"Connectable",
|
|
)
|
|
|
|
VoiceProtocolT = TypeVar("VoiceProtocolT", bound=VoiceProtocol)
|
|
|
|
if TYPE_CHECKING:
|
|
from datetime import datetime
|
|
|
|
from typing_extensions import Self
|
|
|
|
from .asset import Asset
|
|
from .channel import CategoryChannel, DMChannel, PartialMessageable
|
|
from .client import Client
|
|
from .embeds import Embed
|
|
from .emoji import Emoji
|
|
from .enums import InviteTarget
|
|
from .guild import Guild, GuildMessageable
|
|
from .guild_scheduled_event import GuildScheduledEvent
|
|
from .iterators import HistoryIterator
|
|
from .member import Member
|
|
from .message import Message, MessageReference, PartialMessage
|
|
from .state import ConnectionState
|
|
from .threads import AnyThreadArchiveDuration, ForumTag
|
|
from .types.channel import (
|
|
Channel as ChannelPayload,
|
|
DefaultReaction as DefaultReactionPayload,
|
|
GuildChannel as GuildChannelPayload,
|
|
OverwriteType,
|
|
PermissionOverwrite as PermissionOverwritePayload,
|
|
)
|
|
from .types.threads import PartialForumTag as PartialForumTagPayload
|
|
from .ui.action_row import Components, MessageUIComponent
|
|
from .ui.view import View
|
|
from .user import ClientUser
|
|
from .voice_region import VoiceRegion
|
|
|
|
MessageableChannel = Union[GuildMessageable, DMChannel, PartialMessageable]
|
|
SnowflakeTime = Union["Snowflake", datetime]
|
|
|
|
MISSING = utils.MISSING
|
|
|
|
|
|
@runtime_checkable
|
|
class Snowflake(Protocol):
|
|
"""An ABC that details the common operations on a Discord model.
|
|
|
|
Almost all :ref:`Discord models <discord_model>` meet this
|
|
abstract base class.
|
|
|
|
If you want to create a snowflake on your own, consider using
|
|
:class:`.Object`.
|
|
|
|
Attributes
|
|
----------
|
|
id: :class:`int`
|
|
The model's unique ID.
|
|
"""
|
|
|
|
__slots__ = ()
|
|
id: int
|
|
|
|
|
|
@runtime_checkable
|
|
class User(Snowflake, Protocol):
|
|
"""An ABC that details the common operations on a Discord user.
|
|
|
|
The following classes implement this ABC:
|
|
|
|
- :class:`~disnake.User`
|
|
- :class:`~disnake.ClientUser`
|
|
- :class:`~disnake.Member`
|
|
|
|
This ABC must also implement :class:`~disnake.abc.Snowflake`.
|
|
|
|
Attributes
|
|
----------
|
|
name: :class:`str`
|
|
The user's username.
|
|
discriminator: :class:`str`
|
|
The user's discriminator.
|
|
|
|
.. note::
|
|
This is being phased out by Discord; the username system is moving away from ``username#discriminator``
|
|
to users having a globally unique username.
|
|
The value of a single zero (``"0"``) indicates that the user has been migrated to the new system.
|
|
See the `help article <https://dis.gd/app-usernames>`__ for details.
|
|
|
|
global_name: Optional[:class:`str`]
|
|
The user's global display name, if set.
|
|
This takes precedence over :attr:`.name` when shown.
|
|
|
|
.. versionadded:: 2.9
|
|
|
|
bot: :class:`bool`
|
|
Whether the user is a bot account.
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
name: str
|
|
discriminator: str
|
|
global_name: Optional[str]
|
|
bot: bool
|
|
|
|
@property
|
|
def display_name(self) -> str:
|
|
""":class:`str`: Returns the user's display name."""
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
def mention(self) -> str:
|
|
""":class:`str`: Returns a string that allows you to mention the given user."""
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
def avatar(self) -> Optional[Asset]:
|
|
"""Optional[:class:`~disnake.Asset`]: Returns an :class:`~disnake.Asset` for
|
|
the avatar the user has.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
|
|
@runtime_checkable
|
|
class PrivateChannel(Snowflake, Protocol):
|
|
"""An ABC that details the common operations on a private Discord channel.
|
|
|
|
The following classes implement this ABC:
|
|
|
|
- :class:`~disnake.DMChannel`
|
|
- :class:`~disnake.GroupChannel`
|
|
|
|
This ABC must also implement :class:`~disnake.abc.Snowflake`.
|
|
|
|
Attributes
|
|
----------
|
|
me: :class:`~disnake.ClientUser`
|
|
The user representing yourself.
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
me: ClientUser
|
|
|
|
|
|
class _Overwrites:
|
|
__slots__ = ("id", "allow", "deny", "type")
|
|
|
|
ROLE = 0
|
|
MEMBER = 1
|
|
|
|
def __init__(self, data: PermissionOverwritePayload) -> None:
|
|
self.id: int = int(data["id"])
|
|
self.allow: int = int(data.get("allow", 0))
|
|
self.deny: int = int(data.get("deny", 0))
|
|
self.type: OverwriteType = data["type"]
|
|
|
|
def _asdict(self) -> PermissionOverwritePayload:
|
|
return {
|
|
"id": self.id,
|
|
"allow": str(self.allow),
|
|
"deny": str(self.deny),
|
|
"type": self.type,
|
|
}
|
|
|
|
def is_role(self) -> bool:
|
|
return self.type == 0
|
|
|
|
def is_member(self) -> bool:
|
|
return self.type == 1
|
|
|
|
|
|
class GuildChannel(ABC):
|
|
"""An ABC that details the common operations on a Discord guild channel.
|
|
|
|
The following classes implement this ABC:
|
|
|
|
- :class:`.TextChannel`
|
|
- :class:`.VoiceChannel`
|
|
- :class:`.CategoryChannel`
|
|
- :class:`.StageChannel`
|
|
- :class:`.ForumChannel`
|
|
|
|
This ABC must also implement :class:`.abc.Snowflake`.
|
|
|
|
Attributes
|
|
----------
|
|
name: :class:`str`
|
|
The channel name.
|
|
guild: :class:`.Guild`
|
|
The guild the channel belongs to.
|
|
position: :class:`int`
|
|
The position in the channel list. This is a number that starts at 0.
|
|
e.g. the top channel is position 0.
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
id: int
|
|
name: str
|
|
guild: Guild
|
|
type: ChannelType
|
|
position: int
|
|
category_id: Optional[int]
|
|
_flags: int
|
|
_state: ConnectionState
|
|
_overwrites: List[_Overwrites]
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
def __init__(
|
|
self, *, state: ConnectionState, guild: Guild, data: Mapping[str, Any]
|
|
) -> None:
|
|
...
|
|
|
|
def __str__(self) -> str:
|
|
return self.name
|
|
|
|
@property
|
|
def _sorting_bucket(self) -> int:
|
|
raise NotImplementedError
|
|
|
|
def _update(self, guild: Guild, data: Dict[str, Any]) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def _move(
|
|
self,
|
|
position: int,
|
|
parent_id: Optional[int] = None,
|
|
lock_permissions: bool = False,
|
|
*,
|
|
reason: Optional[str],
|
|
) -> None:
|
|
if position < 0:
|
|
raise ValueError("Channel position cannot be less than 0.")
|
|
|
|
http = self._state.http
|
|
bucket = self._sorting_bucket
|
|
channels = [c for c in self.guild.channels if c._sorting_bucket == bucket]
|
|
channels = cast(List[GuildChannel], channels)
|
|
|
|
channels.sort(key=lambda c: c.position)
|
|
|
|
try:
|
|
# remove ourselves from the channel list
|
|
channels.remove(self)
|
|
except ValueError:
|
|
# not there somehow lol
|
|
return
|
|
else:
|
|
index = next(
|
|
(i for i, c in enumerate(channels) if c.position >= position), len(channels)
|
|
)
|
|
# add ourselves at our designated position
|
|
channels.insert(index, self)
|
|
|
|
payload = []
|
|
for index, c in enumerate(channels):
|
|
d: Dict[str, Any] = {"id": c.id, "position": index}
|
|
if parent_id is not MISSING and c.id == self.id:
|
|
d.update(parent_id=parent_id, lock_permissions=lock_permissions)
|
|
payload.append(d)
|
|
|
|
await http.bulk_channel_update(self.guild.id, payload, reason=reason)
|
|
|
|
async def _edit(
|
|
self,
|
|
*,
|
|
name: str = MISSING,
|
|
topic: Optional[str] = MISSING,
|
|
position: int = MISSING,
|
|
nsfw: bool = MISSING,
|
|
sync_permissions: bool = MISSING,
|
|
category: Optional[Snowflake] = MISSING,
|
|
slowmode_delay: Optional[int] = MISSING,
|
|
default_thread_slowmode_delay: Optional[int] = MISSING,
|
|
default_auto_archive_duration: Optional[AnyThreadArchiveDuration] = MISSING,
|
|
type: ChannelType = MISSING,
|
|
overwrites: Mapping[Union[Role, Member], PermissionOverwrite] = MISSING,
|
|
bitrate: int = MISSING,
|
|
user_limit: int = MISSING,
|
|
rtc_region: Optional[Union[str, VoiceRegion]] = MISSING,
|
|
video_quality_mode: VideoQualityMode = MISSING,
|
|
flags: ChannelFlags = MISSING,
|
|
available_tags: Sequence[ForumTag] = MISSING,
|
|
default_reaction: Optional[Union[str, Emoji, PartialEmoji]] = MISSING,
|
|
default_sort_order: Optional[ThreadSortOrder] = MISSING,
|
|
default_layout: ThreadLayout = MISSING,
|
|
reason: Optional[str] = None,
|
|
) -> Optional[ChannelPayload]:
|
|
parent_id: Optional[int]
|
|
if category is not MISSING:
|
|
# if category is given, it's either `None` (no parent) or a category channel
|
|
parent_id = category.id if category else None
|
|
else:
|
|
# if it's not given, don't change the category
|
|
parent_id = MISSING
|
|
|
|
rtc_region_payload: Optional[str]
|
|
if rtc_region is not MISSING:
|
|
rtc_region_payload = str(rtc_region) if rtc_region is not None else None
|
|
else:
|
|
rtc_region_payload = MISSING
|
|
|
|
video_quality_mode_payload: Optional[int]
|
|
if video_quality_mode is not MISSING:
|
|
video_quality_mode_payload = int(video_quality_mode)
|
|
else:
|
|
video_quality_mode_payload = MISSING
|
|
|
|
default_auto_archive_duration_payload: Optional[int]
|
|
if default_auto_archive_duration is not MISSING:
|
|
default_auto_archive_duration_payload = (
|
|
int(default_auto_archive_duration)
|
|
if default_auto_archive_duration is not None
|
|
else default_auto_archive_duration
|
|
)
|
|
else:
|
|
default_auto_archive_duration_payload = MISSING
|
|
|
|
lock_permissions: bool = bool(sync_permissions)
|
|
|
|
overwrites_payload: List[PermissionOverwritePayload] = MISSING
|
|
|
|
if position is not MISSING:
|
|
await self._move(
|
|
position, parent_id=parent_id, lock_permissions=lock_permissions, reason=reason
|
|
)
|
|
parent_id = MISSING # no need to change it again in the edit request below
|
|
elif lock_permissions:
|
|
if parent_id is not MISSING:
|
|
p_id = parent_id
|
|
else:
|
|
p_id = self.category_id
|
|
|
|
if p_id is not None and (parent := self.guild.get_channel(p_id)):
|
|
overwrites_payload = [c._asdict() for c in parent._overwrites]
|
|
|
|
if overwrites is not MISSING and overwrites is not None:
|
|
overwrites_payload = []
|
|
for target, perm in overwrites.items():
|
|
if not isinstance(perm, PermissionOverwrite):
|
|
raise TypeError(
|
|
f"Expected PermissionOverwrite, received {perm.__class__.__name__}"
|
|
)
|
|
|
|
allow, deny = perm.pair()
|
|
payload: PermissionOverwritePayload = {
|
|
"allow": str(allow.value),
|
|
"deny": str(deny.value),
|
|
"id": target.id,
|
|
"type": _Overwrites.ROLE if isinstance(target, Role) else _Overwrites.MEMBER,
|
|
}
|
|
overwrites_payload.append(payload)
|
|
|
|
type_payload: int
|
|
if type is not MISSING:
|
|
if not isinstance(type, ChannelType):
|
|
raise TypeError("type field must be of type ChannelType")
|
|
type_payload = type.value
|
|
else:
|
|
type_payload = MISSING
|
|
|
|
flags_payload: int
|
|
if flags is not MISSING:
|
|
if not isinstance(flags, ChannelFlags):
|
|
raise TypeError("flags field must be of type ChannelFlags")
|
|
flags_payload = flags.value
|
|
else:
|
|
flags_payload = MISSING
|
|
|
|
available_tags_payload: List[PartialForumTagPayload] = MISSING
|
|
if available_tags is not MISSING:
|
|
available_tags_payload = [tag.to_dict() for tag in available_tags]
|
|
|
|
default_reaction_emoji_payload: Optional[DefaultReactionPayload] = MISSING
|
|
if default_reaction is not MISSING:
|
|
if default_reaction is not None:
|
|
emoji_name, emoji_id = PartialEmoji._emoji_to_name_id(default_reaction)
|
|
default_reaction_emoji_payload = {
|
|
"emoji_name": emoji_name,
|
|
"emoji_id": emoji_id,
|
|
}
|
|
else:
|
|
default_reaction_emoji_payload = None
|
|
|
|
default_sort_order_payload: Optional[int] = MISSING
|
|
if default_sort_order is not MISSING:
|
|
default_sort_order_payload = (
|
|
try_enum_to_int(default_sort_order) if default_sort_order is not None else None
|
|
)
|
|
|
|
default_layout_payload: int = MISSING
|
|
if default_layout is not MISSING:
|
|
default_layout_payload = try_enum_to_int(default_layout)
|
|
|
|
options: Dict[str, Any] = {
|
|
"name": name,
|
|
"parent_id": parent_id,
|
|
"topic": topic,
|
|
"bitrate": bitrate,
|
|
"nsfw": nsfw,
|
|
"user_limit": user_limit,
|
|
# note: not passing `position` as it already got updated before, if passed
|
|
"permission_overwrites": overwrites_payload,
|
|
"rate_limit_per_user": slowmode_delay,
|
|
"default_thread_rate_limit_per_user": default_thread_slowmode_delay,
|
|
"type": type_payload,
|
|
"rtc_region": rtc_region_payload,
|
|
"video_quality_mode": video_quality_mode_payload,
|
|
"default_auto_archive_duration": default_auto_archive_duration_payload,
|
|
"flags": flags_payload,
|
|
"available_tags": available_tags_payload,
|
|
"default_reaction_emoji": default_reaction_emoji_payload,
|
|
"default_sort_order": default_sort_order_payload,
|
|
"default_forum_layout": default_layout_payload,
|
|
}
|
|
options = {k: v for k, v in options.items() if v is not MISSING}
|
|
|
|
if options:
|
|
return await self._state.http.edit_channel(self.id, reason=reason, **options)
|
|
|
|
def _fill_overwrites(self, data: GuildChannelPayload) -> None:
|
|
self._overwrites = []
|
|
everyone_index = 0
|
|
everyone_id = self.guild.id
|
|
|
|
for index, overridden in enumerate(data.get("permission_overwrites", [])):
|
|
overwrite = _Overwrites(overridden)
|
|
self._overwrites.append(overwrite)
|
|
|
|
if overwrite.type == _Overwrites.MEMBER:
|
|
continue
|
|
|
|
if overwrite.id == everyone_id:
|
|
# the @everyone role is not guaranteed to be the first one
|
|
# in the list of permission overwrites, however the permission
|
|
# resolution code kind of requires that it is the first one in
|
|
# the list since it is special. So we need the index so we can
|
|
# swap it to be the first one.
|
|
everyone_index = index
|
|
|
|
# do the swap
|
|
tmp = self._overwrites
|
|
if tmp:
|
|
tmp[everyone_index], tmp[0] = tmp[0], tmp[everyone_index]
|
|
|
|
@property
|
|
def changed_roles(self) -> List[Role]:
|
|
"""List[:class:`.Role`]: Returns a list of roles that have been overridden from
|
|
their default values in the :attr:`.Guild.roles` attribute.
|
|
"""
|
|
ret = []
|
|
g = self.guild
|
|
for overwrite in filter(lambda o: o.is_role(), self._overwrites):
|
|
role = g.get_role(overwrite.id)
|
|
if role is None:
|
|
continue
|
|
|
|
role = copy.copy(role)
|
|
role.permissions.handle_overwrite(overwrite.allow, overwrite.deny)
|
|
ret.append(role)
|
|
return ret
|
|
|
|
@property
|
|
def mention(self) -> str:
|
|
""":class:`str`: The string that allows you to mention the channel."""
|
|
return f"<#{self.id}>"
|
|
|
|
@property
|
|
def created_at(self) -> datetime:
|
|
""":class:`datetime.datetime`: Returns the channel's creation time in UTC."""
|
|
return utils.snowflake_time(self.id)
|
|
|
|
def overwrites_for(self, obj: Union[Role, User]) -> PermissionOverwrite:
|
|
"""Returns the channel-specific overwrites for a member or a role.
|
|
|
|
Parameters
|
|
----------
|
|
obj: Union[:class:`.Role`, :class:`.abc.User`]
|
|
The role or user denoting
|
|
whose overwrite to get.
|
|
|
|
Returns
|
|
-------
|
|
:class:`~disnake.PermissionOverwrite`
|
|
The permission overwrites for this object.
|
|
"""
|
|
predicate: Callable[[_Overwrites], bool]
|
|
if isinstance(obj, User):
|
|
predicate = lambda p: p.is_member()
|
|
elif isinstance(obj, Role):
|
|
predicate = lambda p: p.is_role()
|
|
else:
|
|
predicate = lambda p: True
|
|
|
|
for overwrite in filter(predicate, self._overwrites):
|
|
if overwrite.id == obj.id:
|
|
allow = Permissions(overwrite.allow)
|
|
deny = Permissions(overwrite.deny)
|
|
return PermissionOverwrite.from_pair(allow, deny)
|
|
|
|
return PermissionOverwrite()
|
|
|
|
@property
|
|
def overwrites(self) -> Dict[Union[Role, Member], PermissionOverwrite]:
|
|
"""Returns all of the channel's overwrites.
|
|
|
|
This is returned as a dictionary where the key contains the target which
|
|
can be either a :class:`~disnake.Role` or a :class:`~disnake.Member` and the value is the
|
|
overwrite as a :class:`~disnake.PermissionOverwrite`.
|
|
|
|
Returns
|
|
-------
|
|
Dict[Union[:class:`~disnake.Role`, :class:`~disnake.Member`], :class:`~disnake.PermissionOverwrite`]
|
|
The channel's permission overwrites.
|
|
"""
|
|
ret = {}
|
|
for ow in self._overwrites:
|
|
allow = Permissions(ow.allow)
|
|
deny = Permissions(ow.deny)
|
|
overwrite = PermissionOverwrite.from_pair(allow, deny)
|
|
target = None
|
|
|
|
if ow.is_role():
|
|
target = self.guild.get_role(ow.id)
|
|
elif ow.is_member():
|
|
target = self.guild.get_member(ow.id)
|
|
|
|
# TODO: There is potential data loss here in the non-chunked
|
|
# case, i.e. target is None because get_member returned nothing.
|
|
# This can be fixed with a slight breaking change to the return type,
|
|
# i.e. adding disnake.Object to the list of it
|
|
# However, for now this is an acceptable compromise.
|
|
if target is not None:
|
|
ret[target] = overwrite
|
|
return ret
|
|
|
|
@property
|
|
def category(self) -> Optional[CategoryChannel]:
|
|
"""Optional[:class:`~disnake.CategoryChannel`]: The category this channel belongs to.
|
|
|
|
If there is no category then this is ``None``.
|
|
"""
|
|
return self.guild.get_channel(self.category_id) # type: ignore
|
|
|
|
@property
|
|
def permissions_synced(self) -> bool:
|
|
""":class:`bool`: Whether or not the permissions for this channel are synced with the
|
|
category it belongs to.
|
|
|
|
If there is no category then this is ``False``.
|
|
|
|
.. versionadded:: 1.3
|
|
"""
|
|
if self.category_id is None:
|
|
return False
|
|
|
|
category = self.guild.get_channel(self.category_id)
|
|
return bool(category and category.overwrites == self.overwrites)
|
|
|
|
@property
|
|
def flags(self) -> ChannelFlags:
|
|
""":class:`.ChannelFlags`: The channel flags for this channel.
|
|
|
|
.. versionadded:: 2.6
|
|
"""
|
|
return ChannelFlags._from_value(self._flags)
|
|
|
|
@property
|
|
def jump_url(self) -> str:
|
|
"""A URL that can be used to jump to this channel.
|
|
|
|
.. versionadded:: 2.4
|
|
|
|
.. note::
|
|
|
|
This exists for all guild channels but may not be usable by the client for all guild channel types.
|
|
"""
|
|
return f"https://discord.com/channels/{self.guild.id}/{self.id}"
|
|
|
|
def _apply_implict_permissions(self, base: Permissions) -> None:
|
|
# if you can't send a message in a channel then you can't have certain
|
|
# permissions as well
|
|
if not base.send_messages:
|
|
base.send_tts_messages = False
|
|
base.send_voice_messages = False
|
|
base.mention_everyone = False
|
|
base.embed_links = False
|
|
base.attach_files = False
|
|
|
|
# if you can't view a channel then you have no permissions there
|
|
if not base.view_channel:
|
|
denied = Permissions.all_channel()
|
|
base.value &= ~denied.value
|
|
|
|
def permissions_for(
|
|
self,
|
|
obj: Union[Member, Role],
|
|
/,
|
|
*,
|
|
ignore_timeout: bool = MISSING,
|
|
) -> Permissions:
|
|
"""Handles permission resolution for the :class:`~disnake.Member`
|
|
or :class:`~disnake.Role`.
|
|
|
|
This function takes into consideration the following cases:
|
|
|
|
- Guild owner
|
|
- Guild roles
|
|
- Channel overrides
|
|
- Member overrides
|
|
- Timeouts
|
|
|
|
If a :class:`~disnake.Role` is passed, then it checks the permissions
|
|
someone with that role would have, which is essentially:
|
|
|
|
- The default role permissions
|
|
- The permissions of the role used as a parameter
|
|
- The default role permission overwrites
|
|
- The permission overwrites of the role used as a parameter
|
|
|
|
.. versionchanged:: 2.0
|
|
The object passed in can now be a role object.
|
|
|
|
Parameters
|
|
----------
|
|
obj: Union[:class:`~disnake.Member`, :class:`~disnake.Role`]
|
|
The object to resolve permissions for. This could be either
|
|
a member or a role. If it's a role then member overwrites
|
|
are not computed.
|
|
ignore_timeout: :class:`bool`
|
|
Whether or not to ignore the user's timeout.
|
|
Defaults to ``False``.
|
|
|
|
.. versionadded:: 2.4
|
|
|
|
.. note::
|
|
|
|
This only applies to :class:`~disnake.Member` objects.
|
|
|
|
.. versionchanged:: 2.6
|
|
|
|
The default was changed to ``False``.
|
|
|
|
Raises
|
|
------
|
|
TypeError
|
|
``ignore_timeout`` is only supported for :class:`~disnake.Member` objects.
|
|
|
|
Returns
|
|
-------
|
|
:class:`~disnake.Permissions`
|
|
The resolved permissions for the member or role.
|
|
"""
|
|
# The current cases can be explained as:
|
|
# Guild owner get all permissions -- no questions asked. Otherwise...
|
|
# The @everyone role gets the first application.
|
|
# After that, the applied roles that the user has in the channel
|
|
# (or otherwise) are then OR'd together.
|
|
# After the role permissions are resolved, the member permissions
|
|
# have to take into effect.
|
|
# After all that is done.. you have to do the following:
|
|
|
|
# The operation first takes into consideration the denied
|
|
# and then the allowed.
|
|
|
|
# Timeouted users have only view_channel and read_message_history
|
|
# if they already have them.
|
|
if ignore_timeout is not MISSING and isinstance(obj, Role):
|
|
raise TypeError("ignore_timeout is only supported for disnake.Member objects")
|
|
|
|
if ignore_timeout is MISSING:
|
|
ignore_timeout = False
|
|
|
|
if self.guild.owner_id == obj.id:
|
|
return Permissions.all()
|
|
|
|
default = self.guild.default_role
|
|
base = Permissions(default.permissions.value)
|
|
|
|
# Handle the role case first
|
|
if isinstance(obj, Role):
|
|
base.value |= obj._permissions
|
|
|
|
if base.administrator:
|
|
return Permissions.all()
|
|
|
|
# Apply @everyone allow/deny first since it's special
|
|
try:
|
|
maybe_everyone = self._overwrites[0]
|
|
if maybe_everyone.id == self.guild.id:
|
|
base.handle_overwrite(allow=maybe_everyone.allow, deny=maybe_everyone.deny)
|
|
except IndexError:
|
|
pass
|
|
|
|
if obj.is_default():
|
|
return base
|
|
|
|
overwrite = utils.get(self._overwrites, type=_Overwrites.ROLE, id=obj.id)
|
|
if overwrite is not None:
|
|
base.handle_overwrite(overwrite.allow, overwrite.deny)
|
|
|
|
return base
|
|
|
|
roles = obj._roles
|
|
get_role = self.guild.get_role
|
|
|
|
# Apply guild roles that the member has.
|
|
for role_id in roles:
|
|
role = get_role(role_id)
|
|
if role is not None:
|
|
base.value |= role._permissions
|
|
|
|
# Guild-wide Administrator -> True for everything
|
|
# Bypass all channel-specific overrides
|
|
if base.administrator:
|
|
return Permissions.all()
|
|
|
|
# Apply @everyone allow/deny first since it's special
|
|
try:
|
|
maybe_everyone = self._overwrites[0]
|
|
if maybe_everyone.id == self.guild.id:
|
|
base.handle_overwrite(allow=maybe_everyone.allow, deny=maybe_everyone.deny)
|
|
remaining_overwrites = self._overwrites[1:]
|
|
else:
|
|
remaining_overwrites = self._overwrites
|
|
except IndexError:
|
|
remaining_overwrites = self._overwrites
|
|
|
|
denies = 0
|
|
allows = 0
|
|
|
|
# Apply channel specific role permission overwrites
|
|
for overwrite in remaining_overwrites:
|
|
if overwrite.is_role() and roles.has(overwrite.id):
|
|
denies |= overwrite.deny
|
|
allows |= overwrite.allow
|
|
|
|
base.handle_overwrite(allow=allows, deny=denies)
|
|
|
|
# Apply member specific permission overwrites
|
|
for overwrite in remaining_overwrites:
|
|
if overwrite.is_member() and overwrite.id == obj.id:
|
|
base.handle_overwrite(allow=overwrite.allow, deny=overwrite.deny)
|
|
break
|
|
|
|
# if you have a timeout then you can't have any permissions
|
|
# except read messages and read message history
|
|
if not ignore_timeout and obj.current_timeout:
|
|
allowed = Permissions(view_channel=True, read_message_history=True)
|
|
base.value &= allowed.value
|
|
|
|
return base
|
|
|
|
async def delete(self, *, reason: Optional[str] = None) -> None:
|
|
"""|coro|
|
|
|
|
Deletes the channel.
|
|
|
|
You must have :attr:`.Permissions.manage_channels` permission to do this.
|
|
|
|
Parameters
|
|
----------
|
|
reason: Optional[:class:`str`]
|
|
The reason for deleting this channel. Shows up on the audit log.
|
|
|
|
Raises
|
|
------
|
|
Forbidden
|
|
You do not have proper permissions to delete the channel.
|
|
NotFound
|
|
The channel was not found or was already deleted.
|
|
HTTPException
|
|
Deleting the channel failed.
|
|
"""
|
|
await self._state.http.delete_channel(self.id, reason=reason)
|
|
|
|
@overload
|
|
async def set_permissions(
|
|
self,
|
|
target: Union[Member, Role],
|
|
*,
|
|
overwrite: Optional[PermissionOverwrite] = ...,
|
|
reason: Optional[str] = ...,
|
|
) -> None:
|
|
...
|
|
|
|
@overload
|
|
@_overload_with_permissions
|
|
async def set_permissions(
|
|
self,
|
|
target: Union[Member, Role],
|
|
*,
|
|
reason: Optional[str] = ...,
|
|
add_reactions: Optional[bool] = ...,
|
|
administrator: Optional[bool] = ...,
|
|
attach_files: Optional[bool] = ...,
|
|
ban_members: Optional[bool] = ...,
|
|
change_nickname: Optional[bool] = ...,
|
|
connect: Optional[bool] = ...,
|
|
create_forum_threads: Optional[bool] = ...,
|
|
create_instant_invite: Optional[bool] = ...,
|
|
create_private_threads: Optional[bool] = ...,
|
|
create_public_threads: Optional[bool] = ...,
|
|
deafen_members: Optional[bool] = ...,
|
|
embed_links: Optional[bool] = ...,
|
|
external_emojis: Optional[bool] = ...,
|
|
external_stickers: Optional[bool] = ...,
|
|
kick_members: Optional[bool] = ...,
|
|
manage_channels: Optional[bool] = ...,
|
|
manage_emojis: Optional[bool] = ...,
|
|
manage_emojis_and_stickers: Optional[bool] = ...,
|
|
manage_events: Optional[bool] = ...,
|
|
manage_guild: Optional[bool] = ...,
|
|
manage_guild_expressions: Optional[bool] = ...,
|
|
manage_messages: Optional[bool] = ...,
|
|
manage_nicknames: Optional[bool] = ...,
|
|
manage_permissions: Optional[bool] = ...,
|
|
manage_roles: Optional[bool] = ...,
|
|
manage_threads: Optional[bool] = ...,
|
|
manage_webhooks: Optional[bool] = ...,
|
|
mention_everyone: Optional[bool] = ...,
|
|
moderate_members: Optional[bool] = ...,
|
|
move_members: Optional[bool] = ...,
|
|
mute_members: Optional[bool] = ...,
|
|
priority_speaker: Optional[bool] = ...,
|
|
read_message_history: Optional[bool] = ...,
|
|
read_messages: Optional[bool] = ...,
|
|
request_to_speak: Optional[bool] = ...,
|
|
send_messages: Optional[bool] = ...,
|
|
send_messages_in_threads: Optional[bool] = ...,
|
|
send_tts_messages: Optional[bool] = ...,
|
|
send_voice_messages: Optional[bool] = ...,
|
|
speak: Optional[bool] = ...,
|
|
start_embedded_activities: Optional[bool] = ...,
|
|
stream: Optional[bool] = ...,
|
|
use_application_commands: Optional[bool] = ...,
|
|
use_embedded_activities: Optional[bool] = ...,
|
|
use_external_emojis: Optional[bool] = ...,
|
|
use_external_sounds: Optional[bool] = ...,
|
|
use_external_stickers: Optional[bool] = ...,
|
|
use_slash_commands: Optional[bool] = ...,
|
|
use_soundboard: Optional[bool] = ...,
|
|
use_voice_activation: Optional[bool] = ...,
|
|
view_audit_log: Optional[bool] = ...,
|
|
view_channel: Optional[bool] = ...,
|
|
view_creator_monetization_analytics: Optional[bool] = ...,
|
|
view_guild_insights: Optional[bool] = ...,
|
|
) -> None:
|
|
...
|
|
|
|
async def set_permissions(
|
|
self,
|
|
target,
|
|
*,
|
|
overwrite: Optional[PermissionOverwrite] = MISSING,
|
|
reason: Optional[str] = None,
|
|
**permissions,
|
|
) -> None:
|
|
"""|coro|
|
|
|
|
Sets the channel specific permission overwrites for a target in the
|
|
channel.
|
|
|
|
The ``target`` parameter should either be a :class:`.Member` or a
|
|
:class:`.Role` that belongs to guild.
|
|
|
|
The ``overwrite`` parameter, if given, must either be ``None`` or
|
|
:class:`.PermissionOverwrite`. For convenience, you can pass in
|
|
keyword arguments denoting :class:`.Permissions` attributes. If this is
|
|
done, then you cannot mix the keyword arguments with the ``overwrite``
|
|
parameter.
|
|
|
|
If the ``overwrite`` parameter is ``None``, then the permission
|
|
overwrites are deleted.
|
|
|
|
You must have :attr:`.Permissions.manage_roles` permission to do this.
|
|
|
|
.. note::
|
|
|
|
This method *replaces* the old overwrites with the ones given.
|
|
|
|
.. versionchanged:: 2.6
|
|
Raises :exc:`TypeError` instead of ``InvalidArgument``.
|
|
|
|
Examples
|
|
--------
|
|
Setting allow and deny: ::
|
|
|
|
await message.channel.set_permissions(message.author, view_channel=True,
|
|
send_messages=False)
|
|
|
|
Deleting overwrites ::
|
|
|
|
await channel.set_permissions(member, overwrite=None)
|
|
|
|
Using :class:`.PermissionOverwrite` ::
|
|
|
|
overwrite = disnake.PermissionOverwrite()
|
|
overwrite.send_messages = False
|
|
overwrite.view_channel = True
|
|
await channel.set_permissions(member, overwrite=overwrite)
|
|
|
|
Parameters
|
|
----------
|
|
target: Union[:class:`.Member`, :class:`.Role`]
|
|
The member or role to overwrite permissions for.
|
|
overwrite: Optional[:class:`.PermissionOverwrite`]
|
|
The permissions to allow and deny to the target, or ``None`` to
|
|
delete the overwrite.
|
|
**permissions
|
|
A keyword argument list of permissions to set for ease of use.
|
|
Cannot be mixed with ``overwrite``.
|
|
reason: Optional[:class:`str`]
|
|
The reason for doing this action. Shows up on the audit log.
|
|
|
|
Raises
|
|
------
|
|
Forbidden
|
|
You do not have permissions to edit channel specific permissions.
|
|
HTTPException
|
|
Editing channel specific permissions failed.
|
|
NotFound
|
|
The role or member being edited is not part of the guild.
|
|
TypeError
|
|
``overwrite`` is invalid,
|
|
the target type was not :class:`.Role` or :class:`.Member`,
|
|
both keyword arguments and ``overwrite`` were provided,
|
|
or invalid permissions were provided as keyword arguments.
|
|
"""
|
|
http = self._state.http
|
|
|
|
if isinstance(target, User):
|
|
perm_type = _Overwrites.MEMBER
|
|
elif isinstance(target, Role):
|
|
perm_type = _Overwrites.ROLE
|
|
else:
|
|
raise TypeError("target parameter must be either Member or Role")
|
|
|
|
if overwrite is MISSING:
|
|
if len(permissions) == 0:
|
|
raise TypeError("No overwrite provided.")
|
|
try:
|
|
overwrite = PermissionOverwrite(**permissions)
|
|
except (ValueError, TypeError) as e:
|
|
raise TypeError("Invalid permissions given to keyword arguments.") from e
|
|
else:
|
|
if len(permissions) > 0:
|
|
raise TypeError("Cannot mix overwrite and keyword arguments.")
|
|
|
|
# TODO: wait for event
|
|
|
|
if overwrite is None:
|
|
await http.delete_channel_permissions(self.id, target.id, reason=reason)
|
|
elif isinstance(overwrite, PermissionOverwrite):
|
|
(allow, deny) = overwrite.pair()
|
|
await http.edit_channel_permissions(
|
|
self.id, target.id, allow.value, deny.value, perm_type, reason=reason
|
|
)
|
|
else:
|
|
raise TypeError("Invalid overwrite type provided.")
|
|
|
|
async def _clone_impl(
|
|
self,
|
|
base_attrs: Dict[str, Any],
|
|
*,
|
|
name: Optional[str] = None,
|
|
category: Optional[Snowflake] = MISSING,
|
|
overwrites: Mapping[Union[Role, Member], PermissionOverwrite] = MISSING,
|
|
reason: Optional[str] = None,
|
|
) -> Self:
|
|
# if the overwrites are MISSING, defaults to the
|
|
# original permissions of the channel
|
|
overwrites_payload: List[PermissionOverwritePayload]
|
|
if overwrites is not MISSING:
|
|
if not isinstance(overwrites, dict):
|
|
raise TypeError("overwrites parameter expects a dict.")
|
|
|
|
overwrites_payload = []
|
|
for target, perm in overwrites.items():
|
|
if not isinstance(perm, PermissionOverwrite):
|
|
raise TypeError(
|
|
f"Expected PermissionOverwrite, received {perm.__class__.__name__}"
|
|
)
|
|
|
|
allow, deny = perm.pair()
|
|
payload: PermissionOverwritePayload = {
|
|
"allow": str(allow.value),
|
|
"deny": str(deny.value),
|
|
"id": target.id,
|
|
"type": (_Overwrites.ROLE if isinstance(target, Role) else _Overwrites.MEMBER),
|
|
}
|
|
overwrites_payload.append(payload)
|
|
else:
|
|
overwrites_payload = [x._asdict() for x in self._overwrites]
|
|
base_attrs["permission_overwrites"] = overwrites_payload
|
|
if category is not MISSING:
|
|
base_attrs["parent_id"] = category.id if category else None
|
|
else:
|
|
# if no category was given don't change the category
|
|
base_attrs["parent_id"] = self.category_id
|
|
base_attrs["name"] = name or self.name
|
|
channel_type = base_attrs.get("type") or self.type.value
|
|
guild_id = self.guild.id
|
|
cls = self.__class__
|
|
data = await self._state.http.create_channel(
|
|
guild_id, channel_type, reason=reason, **base_attrs
|
|
)
|
|
obj = cls(state=self._state, guild=self.guild, data=data)
|
|
|
|
# temporarily add it to the cache
|
|
self.guild._channels[obj.id] = obj # type: ignore
|
|
return obj
|
|
|
|
async def clone(self, *, name: Optional[str] = None, reason: Optional[str] = None) -> Self:
|
|
"""|coro|
|
|
|
|
Clones this channel. This creates a channel with the same properties
|
|
as this channel.
|
|
|
|
You must have :attr:`.Permissions.manage_channels` permission to
|
|
do this.
|
|
|
|
.. versionadded:: 1.1
|
|
|
|
Parameters
|
|
----------
|
|
name: Optional[:class:`str`]
|
|
The name of the new channel. If not provided, defaults to this channel name.
|
|
reason: Optional[:class:`str`]
|
|
The reason for cloning this channel. Shows up on the audit log.
|
|
|
|
Raises
|
|
------
|
|
Forbidden
|
|
You do not have the proper permissions to create this channel.
|
|
HTTPException
|
|
Creating the channel failed.
|
|
|
|
Returns
|
|
-------
|
|
:class:`.abc.GuildChannel`
|
|
The channel that was created.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@overload
|
|
async def move(
|
|
self,
|
|
*,
|
|
beginning: bool,
|
|
offset: int = ...,
|
|
category: Optional[Snowflake] = ...,
|
|
sync_permissions: bool = ...,
|
|
reason: Optional[str] = ...,
|
|
) -> None:
|
|
...
|
|
|
|
@overload
|
|
async def move(
|
|
self,
|
|
*,
|
|
end: bool,
|
|
offset: int = ...,
|
|
category: Optional[Snowflake] = ...,
|
|
sync_permissions: bool = ...,
|
|
reason: Optional[str] = ...,
|
|
) -> None:
|
|
...
|
|
|
|
@overload
|
|
async def move(
|
|
self,
|
|
*,
|
|
before: Snowflake,
|
|
offset: int = ...,
|
|
category: Optional[Snowflake] = ...,
|
|
sync_permissions: bool = ...,
|
|
reason: Optional[str] = ...,
|
|
) -> None:
|
|
...
|
|
|
|
@overload
|
|
async def move(
|
|
self,
|
|
*,
|
|
after: Snowflake,
|
|
offset: int = ...,
|
|
category: Optional[Snowflake] = ...,
|
|
sync_permissions: bool = ...,
|
|
reason: Optional[str] = ...,
|
|
) -> None:
|
|
...
|
|
|
|
async def move(self, **kwargs) -> None:
|
|
"""|coro|
|
|
|
|
A rich interface to help move a channel relative to other channels.
|
|
|
|
If exact position movement is required, ``edit`` should be used instead.
|
|
|
|
You must have :attr:`.Permissions.manage_channels` permission to
|
|
do this.
|
|
|
|
.. note::
|
|
|
|
Voice channels will always be sorted below text channels.
|
|
This is a Discord limitation.
|
|
|
|
.. versionadded:: 1.7
|
|
|
|
.. versionchanged:: 2.6
|
|
Raises :exc:`TypeError` or :exc:`ValueError` instead of ``InvalidArgument``.
|
|
|
|
Parameters
|
|
----------
|
|
beginning: :class:`bool`
|
|
Whether to move the channel to the beginning of the
|
|
channel list (or category if given).
|
|
This is mutually exclusive with ``end``, ``before``, and ``after``.
|
|
end: :class:`bool`
|
|
Whether to move the channel to the end of the
|
|
channel list (or category if given).
|
|
This is mutually exclusive with ``beginning``, ``before``, and ``after``.
|
|
before: :class:`.abc.Snowflake`
|
|
The channel that should be before our current channel.
|
|
This is mutually exclusive with ``beginning``, ``end``, and ``after``.
|
|
after: :class:`.abc.Snowflake`
|
|
The channel that should be after our current channel.
|
|
This is mutually exclusive with ``beginning``, ``end``, and ``before``.
|
|
offset: :class:`int`
|
|
The number of channels to offset the move by. For example,
|
|
an offset of ``2`` with ``beginning=True`` would move
|
|
it 2 after the beginning. A positive number moves it below
|
|
while a negative number moves it above. Note that this
|
|
number is relative and computed after the ``beginning``,
|
|
``end``, ``before``, and ``after`` parameters.
|
|
category: Optional[:class:`.abc.Snowflake`]
|
|
The category to move this channel under.
|
|
If ``None`` is given then it moves it out of the category.
|
|
This parameter is ignored if moving a category channel.
|
|
sync_permissions: :class:`bool`
|
|
Whether to sync the permissions with the category (if given).
|
|
reason: Optional[:class:`str`]
|
|
The reason for moving this channel. Shows up on the audit log.
|
|
|
|
Raises
|
|
------
|
|
Forbidden
|
|
You do not have permissions to move the channel.
|
|
HTTPException
|
|
Moving the channel failed.
|
|
TypeError
|
|
A bad mix of arguments were passed.
|
|
ValueError
|
|
An invalid position was given.
|
|
"""
|
|
if not kwargs:
|
|
return
|
|
|
|
beginning, end = kwargs.get("beginning"), kwargs.get("end")
|
|
before, after = kwargs.get("before"), kwargs.get("after")
|
|
offset = kwargs.get("offset", 0)
|
|
if sum(bool(a) for a in (beginning, end, before, after)) > 1:
|
|
raise TypeError("Only one of [before, after, end, beginning] can be used.")
|
|
|
|
bucket = self._sorting_bucket
|
|
parent_id = kwargs.get("category", MISSING)
|
|
if parent_id not in (MISSING, None):
|
|
parent_id = parent_id.id
|
|
channels = [
|
|
ch
|
|
for ch in self.guild.channels
|
|
if ch._sorting_bucket == bucket and ch.category_id == parent_id
|
|
]
|
|
else:
|
|
channels = [
|
|
ch
|
|
for ch in self.guild.channels
|
|
if ch._sorting_bucket == bucket and ch.category_id == self.category_id
|
|
]
|
|
|
|
channels.sort(key=lambda c: (c.position, c.id))
|
|
channels = cast(List[GuildChannel], channels)
|
|
|
|
try:
|
|
# Try to remove ourselves from the channel list
|
|
channels.remove(self)
|
|
except ValueError:
|
|
# If we're not there then it's probably due to not being in the category
|
|
pass
|
|
|
|
index = None
|
|
if beginning:
|
|
index = 0
|
|
elif end:
|
|
index = len(channels)
|
|
elif before:
|
|
index = next((i for i, c in enumerate(channels) if c.id == before.id), None)
|
|
elif after:
|
|
index = next((i + 1 for i, c in enumerate(channels) if c.id == after.id), None)
|
|
|
|
if index is None:
|
|
raise ValueError("Could not resolve appropriate move position")
|
|
|
|
channels.insert(max((index + offset), 0), self)
|
|
payload = []
|
|
lock_permissions = kwargs.get("sync_permissions", False)
|
|
reason = kwargs.get("reason")
|
|
for index, channel in enumerate(channels):
|
|
d = {"id": channel.id, "position": index}
|
|
if parent_id is not MISSING and channel.id == self.id:
|
|
d.update(parent_id=parent_id, lock_permissions=lock_permissions)
|
|
payload.append(d)
|
|
|
|
await self._state.http.bulk_channel_update(self.guild.id, payload, reason=reason)
|
|
|
|
async def create_invite(
|
|
self,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
max_age: int = 0,
|
|
max_uses: int = 0,
|
|
temporary: bool = False,
|
|
unique: bool = True,
|
|
target_type: Optional[InviteTarget] = None,
|
|
target_user: Optional[User] = None,
|
|
target_application: Optional[Union[Snowflake, PartyType]] = None,
|
|
guild_scheduled_event: Optional[GuildScheduledEvent] = None,
|
|
) -> Invite:
|
|
"""|coro|
|
|
|
|
Creates an instant invite from a text or voice channel.
|
|
|
|
You must have :attr:`.Permissions.create_instant_invite` permission to
|
|
do this.
|
|
|
|
Parameters
|
|
----------
|
|
max_age: :class:`int`
|
|
How long the invite should last in seconds. If set to ``0``, then the invite
|
|
doesn't expire. Defaults to ``0``.
|
|
|
|
.. warning::
|
|
|
|
If the guild is not a Community guild (has ``COMMUNITY`` in :attr:`.Guild.features`),
|
|
this must be set to a number between ``1`` and ``2592000`` seconds.
|
|
|
|
max_uses: :class:`int`
|
|
How many uses the invite could be used for. If it's 0 then there
|
|
are unlimited uses. Defaults to ``0``.
|
|
temporary: :class:`bool`
|
|
Whether the invite grants temporary membership
|
|
(i.e. they get kicked after they disconnect). Defaults to ``False``.
|
|
unique: :class:`bool`
|
|
Whether a unique invite URL should be created. Defaults to ``True``.
|
|
If this is set to ``False`` then it will return a previously created
|
|
invite.
|
|
target_type: Optional[:class:`.InviteTarget`]
|
|
The type of target for the voice channel invite, if any.
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
target_user: Optional[:class:`User`]
|
|
The user whose stream to display for this invite, required if ``target_type`` is :attr:`.InviteTarget.stream`.
|
|
The user must be streaming in the channel.
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
target_application: Optional[:class:`.Snowflake`]
|
|
The ID of the embedded application for the invite, required if ``target_type`` is :attr:`.InviteTarget.embedded_application`.
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
.. versionchanged:: 2.9
|
|
``PartyType`` is deprecated, and :class:`.Snowflake` should be used instead.
|
|
|
|
guild_scheduled_event: Optional[:class:`.GuildScheduledEvent`]
|
|
The guild scheduled event to include with the invite.
|
|
|
|
.. versionadded:: 2.3
|
|
|
|
reason: Optional[:class:`str`]
|
|
The reason for creating this invite. Shows up on the audit log.
|
|
|
|
Raises
|
|
------
|
|
HTTPException
|
|
Invite creation failed.
|
|
NotFound
|
|
The channel that was passed is a category or an invalid channel.
|
|
|
|
Returns
|
|
-------
|
|
:class:`.Invite`
|
|
The newly created invite.
|
|
"""
|
|
if isinstance(target_application, PartyType):
|
|
utils.warn_deprecated(
|
|
"PartyType is deprecated and will be removed in future version",
|
|
stacklevel=2,
|
|
)
|
|
target_application = Object(target_application.value)
|
|
data = await self._state.http.create_invite(
|
|
self.id,
|
|
reason=reason,
|
|
max_age=max_age,
|
|
max_uses=max_uses,
|
|
temporary=temporary,
|
|
unique=unique,
|
|
target_type=try_enum_to_int(target_type),
|
|
target_user_id=target_user.id if target_user else None,
|
|
target_application_id=target_application.id if target_application else None,
|
|
)
|
|
invite = Invite.from_incomplete(data=data, state=self._state)
|
|
invite.guild_scheduled_event = guild_scheduled_event
|
|
return invite
|
|
|
|
async def invites(self) -> List[Invite]:
|
|
"""|coro|
|
|
|
|
Returns a list of all active instant invites from this channel.
|
|
|
|
You must have :attr:`.Permissions.manage_channels` permission to use this.
|
|
|
|
Raises
|
|
------
|
|
Forbidden
|
|
You do not have proper permissions to get the information.
|
|
HTTPException
|
|
An error occurred while fetching the information.
|
|
|
|
Returns
|
|
-------
|
|
List[:class:`.Invite`]
|
|
The list of invites that are currently active.
|
|
"""
|
|
state = self._state
|
|
data = await state.http.invites_from_channel(self.id)
|
|
guild = self.guild
|
|
return [Invite(state=state, data=invite, channel=self, guild=guild) for invite in data]
|
|
|
|
|
|
class Messageable:
|
|
"""An ABC that details the common operations on a model that can send messages.
|
|
|
|
The following classes implement this ABC:
|
|
|
|
- :class:`~disnake.TextChannel`
|
|
- :class:`~disnake.DMChannel`
|
|
- :class:`~disnake.GroupChannel`
|
|
- :class:`~disnake.User`
|
|
- :class:`~disnake.Member`
|
|
- :class:`~disnake.Thread`
|
|
- :class:`~disnake.VoiceChannel`
|
|
- :class:`~disnake.StageChannel`
|
|
- :class:`~disnake.ext.commands.Context`
|
|
- :class:`~disnake.PartialMessageable`
|
|
"""
|
|
|
|
__slots__ = ()
|
|
_state: ConnectionState
|
|
|
|
async def _get_channel(self) -> MessageableChannel:
|
|
raise NotImplementedError
|
|
|
|
@overload
|
|
async def send(
|
|
self,
|
|
content: Optional[str] = ...,
|
|
*,
|
|
tts: bool = ...,
|
|
embed: Embed = ...,
|
|
file: File = ...,
|
|
stickers: Sequence[Union[GuildSticker, StickerItem]] = ...,
|
|
delete_after: float = ...,
|
|
nonce: Union[str, int] = ...,
|
|
suppress_embeds: bool = ...,
|
|
flags: MessageFlags = ...,
|
|
allowed_mentions: AllowedMentions = ...,
|
|
reference: Union[Message, MessageReference, PartialMessage] = ...,
|
|
mention_author: bool = ...,
|
|
view: View = ...,
|
|
components: Components[MessageUIComponent] = ...,
|
|
) -> Message:
|
|
...
|
|
|
|
@overload
|
|
async def send(
|
|
self,
|
|
content: Optional[str] = ...,
|
|
*,
|
|
tts: bool = ...,
|
|
embed: Embed = ...,
|
|
files: List[File] = ...,
|
|
stickers: Sequence[Union[GuildSticker, StickerItem]] = ...,
|
|
delete_after: float = ...,
|
|
nonce: Union[str, int] = ...,
|
|
suppress_embeds: bool = ...,
|
|
flags: MessageFlags = ...,
|
|
allowed_mentions: AllowedMentions = ...,
|
|
reference: Union[Message, MessageReference, PartialMessage] = ...,
|
|
mention_author: bool = ...,
|
|
view: View = ...,
|
|
components: Components[MessageUIComponent] = ...,
|
|
) -> Message:
|
|
...
|
|
|
|
@overload
|
|
async def send(
|
|
self,
|
|
content: Optional[str] = ...,
|
|
*,
|
|
tts: bool = ...,
|
|
embeds: List[Embed] = ...,
|
|
file: File = ...,
|
|
stickers: Sequence[Union[GuildSticker, StickerItem]] = ...,
|
|
delete_after: float = ...,
|
|
nonce: Union[str, int] = ...,
|
|
suppress_embeds: bool = ...,
|
|
flags: MessageFlags = ...,
|
|
allowed_mentions: AllowedMentions = ...,
|
|
reference: Union[Message, MessageReference, PartialMessage] = ...,
|
|
mention_author: bool = ...,
|
|
view: View = ...,
|
|
components: Components[MessageUIComponent] = ...,
|
|
) -> Message:
|
|
...
|
|
|
|
@overload
|
|
async def send(
|
|
self,
|
|
content: Optional[str] = ...,
|
|
*,
|
|
tts: bool = ...,
|
|
embeds: List[Embed] = ...,
|
|
files: List[File] = ...,
|
|
stickers: Sequence[Union[GuildSticker, StickerItem]] = ...,
|
|
delete_after: float = ...,
|
|
nonce: Union[str, int] = ...,
|
|
suppress_embeds: bool = ...,
|
|
flags: MessageFlags = ...,
|
|
allowed_mentions: AllowedMentions = ...,
|
|
reference: Union[Message, MessageReference, PartialMessage] = ...,
|
|
mention_author: bool = ...,
|
|
view: View = ...,
|
|
components: Components[MessageUIComponent] = ...,
|
|
) -> Message:
|
|
...
|
|
|
|
async def send(
|
|
self,
|
|
content: Optional[str] = None,
|
|
*,
|
|
tts: bool = False,
|
|
embed: Optional[Embed] = None,
|
|
embeds: Optional[List[Embed]] = None,
|
|
file: Optional[File] = None,
|
|
files: Optional[List[File]] = None,
|
|
stickers: Optional[Sequence[Union[GuildSticker, StickerItem]]] = None,
|
|
delete_after: Optional[float] = None,
|
|
nonce: Optional[Union[str, int]] = None,
|
|
suppress_embeds: Optional[bool] = None,
|
|
flags: Optional[MessageFlags] = None,
|
|
allowed_mentions: Optional[AllowedMentions] = None,
|
|
reference: Optional[Union[Message, MessageReference, PartialMessage]] = None,
|
|
mention_author: Optional[bool] = None,
|
|
view: Optional[View] = None,
|
|
components: Optional[Components[MessageUIComponent]] = None,
|
|
):
|
|
"""|coro|
|
|
|
|
Sends a message to the destination with the content given.
|
|
|
|
The content must be a type that can convert to a string through ``str(content)``.
|
|
|
|
At least one of ``content``, ``embed``/``embeds``, ``file``/``files``,
|
|
``stickers``, ``components``, or ``view`` must be provided.
|
|
|
|
To upload a single file, the ``file`` parameter should be used with a
|
|
single :class:`.File` object. To upload multiple files, the ``files``
|
|
parameter should be used with a :class:`list` of :class:`.File` objects.
|
|
**Specifying both parameters will lead to an exception**.
|
|
|
|
To upload a single embed, the ``embed`` parameter should be used with a
|
|
single :class:`.Embed` object. To upload multiple embeds, the ``embeds``
|
|
parameter should be used with a :class:`list` of :class:`.Embed` objects.
|
|
**Specifying both parameters will lead to an exception**.
|
|
|
|
.. versionchanged:: 2.6
|
|
Raises :exc:`TypeError` or :exc:`ValueError` instead of ``InvalidArgument``.
|
|
|
|
Parameters
|
|
----------
|
|
content: Optional[:class:`str`]
|
|
The content of the message to send.
|
|
tts: :class:`bool`
|
|
Whether the message should be sent using text-to-speech.
|
|
embed: :class:`.Embed`
|
|
The rich embed for the content to send. This cannot be mixed with the
|
|
``embeds`` parameter.
|
|
embeds: List[:class:`.Embed`]
|
|
A list of embeds to send with the content. Must be a maximum of 10.
|
|
This cannot be mixed with the ``embed`` parameter.
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
file: :class:`.File`
|
|
The file to upload. This cannot be mixed with the ``files`` parameter.
|
|
files: List[:class:`.File`]
|
|
A list of files to upload. Must be a maximum of 10.
|
|
This cannot be mixed with the ``file`` parameter.
|
|
stickers: Sequence[Union[:class:`.GuildSticker`, :class:`.StickerItem`]]
|
|
A list of stickers to upload. Must be a maximum of 3.
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
nonce: Union[:class:`str`, :class:`int`]
|
|
The nonce to use for sending this message. If the message was successfully sent,
|
|
then the message will have a nonce with this value.
|
|
delete_after: :class:`float`
|
|
If provided, the number of seconds to wait in the background
|
|
before deleting the message we just sent. If the deletion fails,
|
|
then it is silently ignored.
|
|
allowed_mentions: :class:`.AllowedMentions`
|
|
Controls the mentions being processed in this message. If this is
|
|
passed, then the object is merged with :attr:`.Client.allowed_mentions`.
|
|
The merging behaviour only overrides attributes that have been explicitly passed
|
|
to the object, otherwise it uses the attributes set in :attr:`.Client.allowed_mentions`.
|
|
If no object is passed at all then the defaults given by :attr:`.Client.allowed_mentions`
|
|
are used instead.
|
|
|
|
.. versionadded:: 1.4
|
|
|
|
reference: Union[:class:`.Message`, :class:`.MessageReference`, :class:`.PartialMessage`]
|
|
A reference to the :class:`.Message` to which you are replying, this can be created using
|
|
:meth:`.Message.to_reference` or passed directly as a :class:`.Message`. You can control
|
|
whether this mentions the author of the referenced message using the :attr:`.AllowedMentions.replied_user`
|
|
attribute of ``allowed_mentions`` or by setting ``mention_author``.
|
|
|
|
.. versionadded:: 1.6
|
|
|
|
mention_author: Optional[:class:`bool`]
|
|
If set, overrides the :attr:`.AllowedMentions.replied_user` attribute of ``allowed_mentions``.
|
|
|
|
.. versionadded:: 1.6
|
|
|
|
view: :class:`.ui.View`
|
|
A Discord UI View to add to the message. This cannot be mixed with ``components``.
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
components: |components_type|
|
|
A list of components to include in the message. This cannot be mixed with ``view``.
|
|
|
|
.. versionadded:: 2.4
|
|
|
|
suppress_embeds: :class:`bool`
|
|
Whether to suppress embeds for the message. This hides
|
|
all the embeds from the UI if set to ``True``.
|
|
|
|
.. versionadded:: 2.5
|
|
|
|
flags: :class:`.MessageFlags`
|
|
The flags to set for this message.
|
|
Only :attr:`~.MessageFlags.suppress_embeds` and :attr:`~.MessageFlags.suppress_notifications`
|
|
are supported.
|
|
|
|
If parameter ``suppress_embeds`` is provided,
|
|
that will override the setting of :attr:`.MessageFlags.suppress_embeds`.
|
|
|
|
.. versionadded:: 2.9
|
|
|
|
Raises
|
|
------
|
|
HTTPException
|
|
Sending the message failed.
|
|
Forbidden
|
|
You do not have the proper permissions to send the message.
|
|
TypeError
|
|
Specified both ``file`` and ``files``,
|
|
or you specified both ``embed`` and ``embeds``,
|
|
or you specified both ``view`` and ``components``,
|
|
or the ``reference`` object is not a :class:`.Message`,
|
|
:class:`.MessageReference` or :class:`.PartialMessage`.
|
|
ValueError
|
|
The ``files`` or ``embeds`` list is too large.
|
|
|
|
Returns
|
|
-------
|
|
:class:`.Message`
|
|
The message that was sent.
|
|
"""
|
|
channel = await self._get_channel()
|
|
state = self._state
|
|
content = str(content) if content is not None else None
|
|
|
|
if file is not None and files is not None:
|
|
raise TypeError("cannot pass both file and files parameter to send()")
|
|
|
|
if file is not None:
|
|
if not isinstance(file, File):
|
|
raise TypeError("file parameter must be File")
|
|
files = [file]
|
|
|
|
if embed is not None and embeds is not None:
|
|
raise TypeError("cannot pass both embed and embeds parameter to send()")
|
|
|
|
if embed is not None:
|
|
embeds = [embed]
|
|
|
|
embeds_payload = None
|
|
if embeds is not None:
|
|
if len(embeds) > 10:
|
|
raise ValueError("embeds parameter must be a list of up to 10 elements")
|
|
for embed in embeds:
|
|
if embed._files:
|
|
files = files or []
|
|
files.extend(embed._files.values())
|
|
embeds_payload = [embed.to_dict() for embed in embeds]
|
|
|
|
stickers_payload = None
|
|
if stickers is not None:
|
|
stickers_payload = [sticker.id for sticker in stickers]
|
|
|
|
allowed_mentions_payload = None
|
|
if allowed_mentions is None:
|
|
allowed_mentions_payload = state.allowed_mentions and state.allowed_mentions.to_dict()
|
|
elif state.allowed_mentions is not None:
|
|
allowed_mentions_payload = state.allowed_mentions.merge(allowed_mentions).to_dict()
|
|
else:
|
|
allowed_mentions_payload = allowed_mentions.to_dict()
|
|
|
|
if mention_author is not None:
|
|
allowed_mentions_payload = allowed_mentions_payload or AllowedMentions().to_dict()
|
|
allowed_mentions_payload["replied_user"] = bool(mention_author)
|
|
|
|
reference_payload = None
|
|
if reference is not None:
|
|
try:
|
|
reference_payload = reference.to_message_reference_dict()
|
|
except AttributeError:
|
|
raise TypeError(
|
|
"reference parameter must be Message, MessageReference, or PartialMessage"
|
|
) from None
|
|
|
|
if view is not None and components is not None:
|
|
raise TypeError("cannot pass both view and components parameter to send()")
|
|
|
|
elif view:
|
|
if not hasattr(view, "__discord_ui_view__"):
|
|
raise TypeError(f"view parameter must be View not {view.__class__!r}")
|
|
|
|
components_payload = view.to_components()
|
|
|
|
elif components:
|
|
components_payload = components_to_dict(components)
|
|
|
|
else:
|
|
components_payload = None
|
|
|
|
flags_payload = None
|
|
if suppress_embeds is not None:
|
|
flags = MessageFlags._from_value(0 if flags is None else flags.value)
|
|
flags.suppress_embeds = suppress_embeds
|
|
if flags is not None:
|
|
flags_payload = flags.value
|
|
|
|
if files is not None:
|
|
if len(files) > 10:
|
|
raise ValueError("files parameter must be a list of up to 10 elements")
|
|
elif not all(isinstance(file, File) for file in files):
|
|
raise TypeError("files parameter must be a list of File")
|
|
|
|
try:
|
|
data = await state.http.send_files(
|
|
channel.id,
|
|
files=files,
|
|
content=content,
|
|
tts=tts,
|
|
embeds=embeds_payload,
|
|
nonce=nonce,
|
|
allowed_mentions=allowed_mentions_payload,
|
|
message_reference=reference_payload,
|
|
stickers=stickers_payload,
|
|
components=components_payload,
|
|
flags=flags_payload,
|
|
)
|
|
finally:
|
|
for f in files:
|
|
f.close()
|
|
else:
|
|
data = await state.http.send_message(
|
|
channel.id,
|
|
content,
|
|
tts=tts,
|
|
embeds=embeds_payload,
|
|
nonce=nonce,
|
|
allowed_mentions=allowed_mentions_payload,
|
|
message_reference=reference_payload,
|
|
stickers=stickers_payload,
|
|
components=components_payload,
|
|
flags=flags_payload,
|
|
)
|
|
|
|
ret = state.create_message(channel=channel, data=data)
|
|
if view:
|
|
state.store_view(view, ret.id)
|
|
|
|
if delete_after is not None:
|
|
await ret.delete(delay=delete_after)
|
|
return ret
|
|
|
|
async def trigger_typing(self) -> None:
|
|
"""|coro|
|
|
|
|
Triggers a *typing* indicator to the destination.
|
|
|
|
*Typing* indicator will go away after 10 seconds, or after a message is sent.
|
|
"""
|
|
channel = await self._get_channel()
|
|
await self._state.http.send_typing(channel.id)
|
|
|
|
def typing(self) -> Typing:
|
|
"""Returns a context manager that allows you to type for an indefinite period of time.
|
|
|
|
This is useful for denoting long computations in your bot.
|
|
|
|
.. note::
|
|
|
|
This is both a regular context manager and an async context manager.
|
|
This means that both ``with`` and ``async with`` work with this.
|
|
|
|
Example Usage: ::
|
|
|
|
async with channel.typing():
|
|
# simulate something heavy
|
|
await asyncio.sleep(10)
|
|
|
|
await channel.send('done!')
|
|
|
|
"""
|
|
return Typing(self)
|
|
|
|
async def fetch_message(self, id: int, /) -> Message:
|
|
"""|coro|
|
|
|
|
Retrieves a single :class:`.Message` from the destination.
|
|
|
|
Parameters
|
|
----------
|
|
id: :class:`int`
|
|
The message ID to look for.
|
|
|
|
Raises
|
|
------
|
|
NotFound
|
|
The specified message was not found.
|
|
Forbidden
|
|
You do not have the permissions required to get a message.
|
|
HTTPException
|
|
Retrieving the message failed.
|
|
|
|
Returns
|
|
-------
|
|
:class:`.Message`
|
|
The message asked for.
|
|
"""
|
|
channel = await self._get_channel()
|
|
data = await self._state.http.get_message(channel.id, id)
|
|
return self._state.create_message(channel=channel, data=data)
|
|
|
|
async def pins(self) -> List[Message]:
|
|
"""|coro|
|
|
|
|
Retrieves all messages that are currently pinned in the channel.
|
|
|
|
.. note::
|
|
|
|
Due to a limitation with the Discord API, the :class:`.Message`
|
|
objects returned by this method do not contain complete
|
|
:attr:`.Message.reactions` data.
|
|
|
|
Raises
|
|
------
|
|
HTTPException
|
|
Retrieving the pinned messages failed.
|
|
|
|
Returns
|
|
-------
|
|
List[:class:`.Message`]
|
|
The messages that are currently pinned.
|
|
"""
|
|
channel = await self._get_channel()
|
|
state = self._state
|
|
data = await state.http.pins_from(channel.id)
|
|
return [state.create_message(channel=channel, data=m) for m in data]
|
|
|
|
def history(
|
|
self,
|
|
*,
|
|
limit: Optional[int] = 100,
|
|
before: Optional[SnowflakeTime] = None,
|
|
after: Optional[SnowflakeTime] = None,
|
|
around: Optional[SnowflakeTime] = None,
|
|
oldest_first: Optional[bool] = None,
|
|
) -> HistoryIterator:
|
|
"""Returns an :class:`.AsyncIterator` that enables receiving the destination's message history.
|
|
|
|
You must have :attr:`.Permissions.read_message_history` permission to use this.
|
|
|
|
Examples
|
|
--------
|
|
Usage ::
|
|
|
|
counter = 0
|
|
async for message in channel.history(limit=200):
|
|
if message.author == client.user:
|
|
counter += 1
|
|
|
|
Flattening into a list: ::
|
|
|
|
messages = await channel.history(limit=123).flatten()
|
|
# messages is now a list of Message...
|
|
|
|
All parameters are optional.
|
|
|
|
Parameters
|
|
----------
|
|
limit: Optional[:class:`int`]
|
|
The number of messages to retrieve.
|
|
If ``None``, retrieves every message in the channel. Note, however,
|
|
that this would make it a slow operation.
|
|
before: Optional[Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]]
|
|
Retrieve messages before this date or message.
|
|
If a datetime is provided, it is recommended to use a UTC aware datetime.
|
|
If the datetime is naive, it is assumed to be local time.
|
|
after: Optional[Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]]
|
|
Retrieve messages after this date or message.
|
|
If a datetime is provided, it is recommended to use a UTC aware datetime.
|
|
If the datetime is naive, it is assumed to be local time.
|
|
around: Optional[Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]]
|
|
Retrieve messages around this date or message.
|
|
If a datetime is provided, it is recommended to use a UTC aware datetime.
|
|
If the datetime is naive, it is assumed to be local time.
|
|
When using this argument, the maximum limit is 101. Note that if the limit is an
|
|
even number then this will return at most limit + 1 messages.
|
|
oldest_first: Optional[:class:`bool`]
|
|
If set to ``True``, return messages in oldest->newest order. Defaults to ``True`` if
|
|
``after`` is specified, otherwise ``False``.
|
|
|
|
Raises
|
|
------
|
|
Forbidden
|
|
You do not have permissions to get channel message history.
|
|
HTTPException
|
|
The request to get message history failed.
|
|
|
|
Yields
|
|
------
|
|
:class:`.Message`
|
|
The message with the message data parsed.
|
|
"""
|
|
from .iterators import HistoryIterator # cyclic import
|
|
|
|
return HistoryIterator(
|
|
self, limit=limit, before=before, after=after, around=around, oldest_first=oldest_first
|
|
)
|
|
|
|
|
|
class Connectable(Protocol):
|
|
"""An ABC that details the common operations on a channel that can
|
|
connect to a voice server.
|
|
|
|
The following classes implement this ABC:
|
|
|
|
- :class:`~disnake.VoiceChannel`
|
|
- :class:`~disnake.StageChannel`
|
|
|
|
Note
|
|
----
|
|
This ABC is not decorated with :func:`typing.runtime_checkable`, so will fail :func:`isinstance`/:func:`issubclass`
|
|
checks.
|
|
"""
|
|
|
|
__slots__ = ()
|
|
_state: ConnectionState
|
|
guild: Guild
|
|
id: int
|
|
|
|
def _get_voice_client_key(self) -> Tuple[int, str]:
|
|
raise NotImplementedError
|
|
|
|
def _get_voice_state_pair(self) -> Tuple[int, int]:
|
|
raise NotImplementedError
|
|
|
|
async def connect(
|
|
self,
|
|
*,
|
|
timeout: float = 60.0,
|
|
reconnect: bool = True,
|
|
cls: Callable[[Client, Connectable], VoiceProtocolT] = VoiceClient,
|
|
) -> VoiceProtocolT:
|
|
"""|coro|
|
|
|
|
Connects to voice and creates a :class:`VoiceClient` to establish
|
|
your connection to the voice server.
|
|
|
|
This requires :attr:`Intents.voice_states`.
|
|
|
|
Parameters
|
|
----------
|
|
timeout: :class:`float`
|
|
The timeout in seconds to wait for the voice endpoint.
|
|
reconnect: :class:`bool`
|
|
Whether the bot should automatically attempt
|
|
a reconnect if a part of the handshake fails
|
|
or the gateway goes down.
|
|
cls: Type[:class:`VoiceProtocol`]
|
|
A type that subclasses :class:`VoiceProtocol` to connect with.
|
|
Defaults to :class:`VoiceClient`.
|
|
|
|
Raises
|
|
------
|
|
asyncio.TimeoutError
|
|
Could not connect to the voice channel in time.
|
|
ClientException
|
|
You are already connected to a voice channel.
|
|
opus.OpusNotLoaded
|
|
The opus library has not been loaded.
|
|
|
|
Returns
|
|
-------
|
|
:class:`VoiceProtocol`
|
|
A voice client that is fully connected to the voice server.
|
|
"""
|
|
key_id, _ = self._get_voice_client_key()
|
|
state = self._state
|
|
|
|
if state._get_voice_client(key_id):
|
|
raise ClientException("Already connected to a voice channel.")
|
|
|
|
client = state._get_client()
|
|
voice = cls(client, self)
|
|
|
|
if not isinstance(voice, VoiceProtocol):
|
|
raise TypeError("Type must meet VoiceProtocol abstract base class.")
|
|
|
|
state._add_voice_client(key_id, voice)
|
|
|
|
try:
|
|
await voice.connect(timeout=timeout, reconnect=reconnect)
|
|
except asyncio.TimeoutError:
|
|
try:
|
|
await voice.disconnect(force=True)
|
|
except Exception:
|
|
# we don't care if disconnect failed because connection failed
|
|
pass
|
|
raise # re-raise
|
|
|
|
return voice
|