mirror
/
Rapptz.discord.py
mirror of https://github.com/Rapptz/discord.py
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

fix client presences/sessions; add timeout capability; remove GuildIterator; add proper team/application support

pull/10109/head
dolfies 3 years ago
parent
da7a69fa3d
commit
6420dfdc0f
12 changed files with 1263 additions and 451 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 82
      discord/activity.py
  2. 465
      discord/appinfo.py
  3. 542
      discord/client.py
  4. 17
      discord/gateway.py
  5. 48
      discord/http.py
  6. 114
      discord/iterators.py
  7. 116
      discord/member.py
  8. 120
      discord/state.py
  9. 193
      discord/team.py
  10. 9
      discord/types/appinfo.py
  11. 6
      discord/types/team.py
  12. 2
      requirements.txt

82
discord/activity.py
View File

@ -252,6 +252,22 @@ class Activity(BaseActivity):
inner = ' '.join('%s=%r' % t for t in attrs)
return f'<Activity {inner}>'
def __eq__(self, other):
return (
isinstance(other, Activity) and
other.type == self.type and
other.name == self.name and
other.url == self.url and
other.emoji == self.emoji and
other.state == self.state and
other.session_id == self.session_id and
other.sync_id == self.sync_id and
other.start == self.start
)
def __ne__(self, other):
return not self.__eq__(other)
def to_dict(self) -> Dict[str, Any]:
ret: Dict[str, Any] = {}
for attr in self.__slots__:
@ -730,31 +746,45 @@ class CustomActivity(BaseActivity):
.. versionadded:: 1.3
.. note::
Technically, the name of custom activities is hardcoded to "Custom Status",
and the state parameter has the actual custom text.
This is confusing, so here, the name represents the actual custom text.
However, the "correct" way still works.
Attributes
-----------
name: Optional[:class:`str`]
The custom activity's name.
emoji: Optional[:class:`PartialEmoji`]
The emoji to pass to the activity, if any.
expires_at: Optional[:class:`datetime.datetime`]
When the custom activity will expire. This is only available from :attr:`discord.Settings.custom_activity`
"""
__slots__ = ('name', 'emoji', 'state')
def __init__(self, name: Optional[str], *, emoji: Optional[PartialEmoji] = None, **extra: Any):
super().__init__(**extra)
__slots__ = ('name', 'emoji', 'expires_at')
def __init__(
self,
name: Optional[str],
*,
emoji: Optional[PartialEmoji] = None,
state: Optional[str] = None,
expires_at: Optional[datetime.datetime] = None,
**kwargs,
):
super().__init__(**kwargs)
if name == 'Custom Status':
name = state
self.name: Optional[str] = name
self.state = state = extra.pop('state', None)
if self.name == 'Custom Status':
self.name = state
self.expires_at = expires_at
self.emoji: Optional[PartialEmoji]
if emoji is None:
self.emoji = emoji
elif isinstance(emoji, dict):
if isinstance(emoji, dict):
self.emoji = PartialEmoji.from_dict(emoji)
elif isinstance(emoji, str):
self.emoji = PartialEmoji(name=emoji)
elif isinstance(emoji, PartialEmoji):
elif isinstance(emoji, PartialEmoji) or emoji is None:
self.emoji = emoji
else:
raise TypeError(f'Expected str, PartialEmoji, or None, received {type(emoji)!r} instead.')
@ -767,16 +797,29 @@ class CustomActivity(BaseActivity):
"""
return ActivityType.custom
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> Dict[str, Union[str, int]]:
o = {
'type': ActivityType.custom.value,
'state': self.name,
'name': 'Custom Status',
'name': 'Custom Status', # Not a confusing API at all
}
if self.emoji:
o['emoji'] = self.emoji.to_dict()
return o
def to_settings_dict(self) -> Dict[str, Any]:
o: Dict[str, Optional[Union[str, int]]] = {}
if (text := self.name):
o['text'] = text
if (emoji := self.emoji):
o['emoji_name'] = emoji.name
if emoji.id:
o['emoji_id'] = emoji.id
if (expiry := self.expires_at) is not None:
o['expires_at'] = expiry.isoformat()
return o
def __eq__(self, other: Any) -> bool:
return isinstance(other, CustomActivity) and other.name == self.name and other.emoji == self.emoji
@ -833,3 +876,16 @@ def create_activity(data: Optional[ActivityPayload]) -> Optional[ActivityTypes]:
elif game_type is ActivityType.listening and 'sync_id' in data and 'session_id' in data:
return Spotify(**data)
return Activity(**data)
def create_settings_activity(*, data, state):
if not data:
return
emoji = None
if (emoji_id := _get_as_snowflake(data, 'emoji_id')) is not None:
emoji = state.get_emoji(emoji_id)
emoji = emoji and emoji._to_partial()
elif (emoji_name := data.get('emoji_name')) is not None:
emoji = PartialEmoji(name=emoji_name)
return CustomActivity(name=data.get('text'), emoji=emoji, expires_at=data.get('expires_at'))

465
discord/appinfo.py
View File

@ -28,25 +28,93 @@ from typing import List, TYPE_CHECKING, Optional
from . import utils
from .asset import Asset
from .enums import ApplicationVerificationState, RPCApplicationState, StoreApplicationState, try_enum
from .flags import ApplicationFlags
from .user import User
if TYPE_CHECKING:
from .abc import Snowflake
from .guild import Guild
from .types.appinfo import (
AppInfo as AppInfoPayload,
PartialAppInfo as PartialAppInfoPayload,
Team as TeamPayload,
)
from .user import User
from .state import ConnectionState
from .user import BaseUser
__all__ = (
'AppInfo',
'PartialAppInfo',
'Application',
'PartialApplication',
)
MISSING = utils.MISSING
class AppInfo:
"""Represents application info for an application/bot.
class ApplicationBot(User):
__slots__ = ('token', 'public', 'require_code_grant')
def __init__(self, *, data, state: ConnectionState, application: Application):
super().__init__(state=state, data=data)
self.application = application
self.token: str = data['token']
self.public: bool = data['public']
self.require_code_grant: bool = data['require_code_grant']
async def reset_token(self) -> None:
"""|coro|
Resets the bot's token.
Raises
------
HTTPException
Resetting the token failed.
"""
data = await self._state.http.reset_token(self.application.id)
self.token = data['token']
self._update(data)
async def edit(
self,
*,
public: bool = MISSING,
require_code_grant: bool = MISSING,
) -> None:
"""|coro|
Edits the bot.
Parameters
-----------
public: :class:`bool`
Whether the bot is public or not.
require_code_grant: :class:`bool`
Whether the bot requires a code grant or not.
Raises
------
Forbidden
You are not allowed to edit this bot.
HTTPException
Editing the bot failed.
"""
payload = {}
if public is not MISSING:
payload['bot_public'] = public
if require_code_grant is not MISSING:
payload['bot_require_code_grant'] = require_code_grant
data = await self._state.http.edit_application(self.application.id, payload=payload)
self.public = data['bot_public']
self.require_code_grant = data['bot_require_code_grant']
self.application._update(data)
class PartialApplication:
"""Represents a partial Application.
.. versionadded:: 2.0
Attributes
-------------
@ -54,119 +122,82 @@ class AppInfo:
The application ID.
name: :class:`str`
The application name.
owner: :class:`User`
The application owner.
team: Optional[:class:`Team`]
The application's team.
.. versionadded:: 1.3
description: :class:`str`
The application description.
bot_public: :class:`bool`
Whether the bot can be invited by anyone or if it is locked
to the application owner.
bot_require_code_grant: :class:`bool`
Whether the bot requires the completion of the full oauth2 code
grant flow to join.
rpc_origins: Optional[List[:class:`str`]]
A list of RPC origin URLs, if RPC is enabled.
summary: :class:`str`
If this application is a game sold on Discord,
this field will be the summary field for the store page of its primary SKU.
.. versionadded:: 1.3
verify_key: :class:`str`
The hex encoded key for verification in interactions and the
GameSDK's `GetTicket <https://discord.com/developers/docs/game-sdk/applications#getticket>`_.
.. versionadded:: 1.3
guild_id: Optional[:class:`int`]
If this application is a game sold on Discord,
this field will be the guild to which it has been linked to.
.. versionadded:: 1.3
primary_sku_id: Optional[:class:`int`]
If this application is a game sold on Discord,
this field will be the id of the "Game SKU" that is created,
if it exists.
.. versionadded:: 1.3
slug: Optional[:class:`str`]
If this application is a game sold on Discord,
this field will be the URL slug that links to the store page.
.. versionadded:: 1.3
terms_of_service_url: Optional[:class:`str`]
The application's terms of service URL, if set.
.. versionadded:: 2.0
privacy_policy_url: Optional[:class:`str`]
The application's privacy policy URL, if set.
.. versionadded:: 2.0
public: :class:`bool`
Whether the integration can be invited by anyone or if it is locked
to the application owner.
require_code_grant: :class:`bool`
Whether the integration requires the completion of the full OAuth2 code
grant flow to join
max_participants: Optional[:class:`int`]
The max number of people that can participate in the activity.
Only available for embedded activities.
premium_tier_level: Optional[:class:`int`]
The required premium tier level to launch the activity.
Only available for embedded activities.
"""
__slots__ = (
'_state',
'description',
'id',
'name',
'description',
'rpc_origins',
'bot_public',
'bot_require_code_grant',
'owner',
'_icon',
'summary',
'verify_key',
'team',
'guild_id',
'primary_sku_id',
'slug',
'_cover_image',
'terms_of_service_url',
'privacy_policy_url',
'_icon',
'_flags'
'_cover_image',
'public',
'require_code_grant',
'type',
'hook',
'premium_tier_level',
)
def __init__(self, state: ConnectionState, data: AppInfoPayload):
from .team import Team
def __init__(self, *, state: ConnectionState, data: PartialAppInfoPayload):
self._state: ConnectionState = state
self._update(data)
def _update(self, data: PartialAppInfoPayload) -> None:
self.id: int = int(data['id'])
self.name: str = data['name']
self.description: str = data['description']
self._icon: Optional[str] = data['icon']
self.rpc_origins: List[str] = data['rpc_origins']
self.bot_public: bool = data['bot_public']
self.bot_require_code_grant: bool = data['bot_require_code_grant']
self.owner: User = state.create_user(data['owner'])
team: Optional[TeamPayload] = data.get('team')
self.team: Optional[Team] = Team(state, team) if team else None
self.rpc_origins: Optional[List[str]] = data.get('rpc_origins')
self.summary: str = data['summary']
self.verify_key: str = data['verify_key']
self.guild_id: Optional[int] = utils._get_as_snowflake(data, 'guild_id')
self.primary_sku_id: Optional[int] = utils._get_as_snowflake(data, 'primary_sku_id')
self.slug: Optional[str] = data.get('slug')
self._icon: Optional[str] = data.get('icon')
self._cover_image: Optional[str] = data.get('cover_image')
self.terms_of_service_url: Optional[str] = data.get('terms_of_service_url')
self.privacy_policy_url: Optional[str] = data.get('privacy_policy_url')
self._flags: int = data.get('flags', 0)
self.type: Optional[int] = data.get('type')
self.hook: bool = data.get('hook', False)
self.max_participants: Optional[int] = data.get('max_participants')
self.premium_tier_level: Optional[int] = data.get('embedded_activity_config', {}).get('activity_premium_tier_level')
self.public: bool = data.get('integration_public', data.get('bot_public')) # The two seem to be used interchangeably?
self.require_code_grant: bool = data.get('integration_require_code_grant', data.get('bot_require_code_grant')) # Same here
def __repr__(self) -> str:
return (
f'<{self.__class__.__name__} id={self.id} name={self.name!r} '
f'description={self.description!r} public={self.bot_public} '
f'owner={self.owner!r}>'
)
return f'<{self.__class__.__name__} id={self.id} name={self.name!r} description={self.description!r}>'
@property
def icon(self) -> Optional[Asset]:
@ -186,61 +217,251 @@ class AppInfo:
return Asset._from_cover_image(self._state, self.id, self._cover_image)
@property
def guild(self) -> Optional[Guild]:
"""Optional[:class:`Guild`]: If this application is a game sold on Discord,
this field will be the guild to which it has been linked
def flags(self) -> ApplicationFlags:
""":class:`ApplicationFlags`: The flags of this application."""
return ApplicationFlags._from_value(self._flags)
.. versionadded:: 1.3
"""
return self._state._get_guild(self.guild_id)
class PartialAppInfo:
"""Represents a partial AppInfo given by :func:`~discord.abc.GuildChannel.create_invite`
class Application(PartialApplication):
"""Represents application info for an application you own.
.. versionadded:: 2.0
Attributes
-------------
id: :class:`int`
The application ID.
name: :class:`str`
The application name.
description: :class:`str`
The application description.
rpc_origins: Optional[List[:class:`str`]]
A list of RPC origin URLs, if RPC is enabled.
summary: :class:`str`
owner: :class:`BaseUser`
The application owner.
team: Optional[:class:`Team`]
The application's team.
bot: Optional[:class:`ApplicationBot`]
The bot attached to the application, if any.
guild_id: Optional[:class:`int`]
If this application is a game sold on Discord,
this field will be the summary field for the store page of its primary SKU.
verify_key: :class:`str`
The hex encoded key for verification in interactions and the
GameSDK's `GetTicket <https://discord.com/developers/docs/game-sdk/applications#getticket>`_.
terms_of_service_url: Optional[:class:`str`]
The application's terms of service URL, if set.
privacy_policy_url: Optional[:class:`str`]
The application's privacy policy URL, if set.
this field will be the guild to which it has been linked to.
primary_sku_id: Optional[:class:`int`]
If this application is a game sold on Discord,
this field will be the id of the "Game SKU" that is created,
if it exists.
slug: Optional[:class:`str`]
If this application is a game sold on Discord,
this field will be the URL slug that links to the store page.
interactions_endpoint_url: Optional[:class:`str`]
The URL interactions will be sent to, if set.
secret: :class:`str`
The application's secret key.
redirect_uris: List[:class:`str`]
A list of redirect URIs authorized for this application.
tags: List[:class:`str`]
A list of tags that describe the application.
verification_state: :class:`ApplicationVerificationState`
The verification state of the application.
store_application_state: :class:`StoreApplicationState`
The approval state of the commerce application.
rpc_application_state: :class:`RPCApplicationState`
The approval state of the RPC usage application.
"""
__slots__ = ('_state', 'id', 'name', 'description', 'rpc_origins', 'summary', 'verify_key', 'terms_of_service_url', 'privacy_policy_url', '_icon')
__slots__ = (
'owner',
'team',
'guild_id',
'primary_sku_id',
'slug',
'secret',
'redirect_uris',
'bot',
'tags',
'verification_state',
'store_application_state',
'rpc_application_state',
'interactions_endpoint_url',
)
def __init__(self, *, state: ConnectionState, data: PartialAppInfoPayload):
self._state: ConnectionState = state
self.id: int = int(data['id'])
self.name: str = data['name']
self._icon: Optional[str] = data.get('icon')
self.description: str = data['description']
self.rpc_origins: Optional[List[str]] = data.get('rpc_origins')
self.summary: str = data['summary']
self.verify_key: str = data['verify_key']
self.terms_of_service_url: Optional[str] = data.get('terms_of_service_url')
self.privacy_policy_url: Optional[str] = data.get('privacy_policy_url')
def _update(self, data: AppInfoPayload) -> None:
super()._update(data)
from .team import Team
self.secret: str = data['secret']
self.redirect_uris: List[str] = data.get('redirect_uris', [])
self.tags: List[str] = data.get('tags', [])
self.guild_id: Optional[int] = utils._get_as_snowflake(data, 'guild_id')
self.verification_state = try_enum(ApplicationVerificationState, data['verification_state'])
self.store_application_state = try_enum(StoreApplicationState, data['store_application_state'])
self.rpc_application_state = try_enum(RPCApplicationState, data['rpc_application_state'])
self.primary_sku_id: Optional[int] = utils._get_as_snowflake(data, 'primary_sku_id')
self.slug: Optional[str] = data.get('slug')
self.interactions_endpoint_url: Optional[str] = data['interactions_endpoint_url']
state = self._state
team: Optional[TeamPayload] = data.get('team')
self.team: Optional[Team] = Team(state, team) if team else None
if (bot := data.get('bot')):
bot['public'] = data.get('bot_public', self.public)
bot['require_code_grant'] = data.get('bot_require_code_grant', self.require_code_grant)
self.bot: Optional[ApplicationBot] = ApplicationBot(data=bot, state=state, application=self) if bot else None
owner = data.get('owner')
if owner is not None and int(owner['id']) != state.self_id: # Consistency
self.owner: BaseUser = state.create_user(owner)
else:
self.owner: BaseUser = state.user # type: ignore
def __repr__(self) -> str:
return f'<{self.__class__.__name__} id={self.id} name={self.name!r} description={self.description!r}>'
return (
f'<{self.__class__.__name__} id={self.id} name={self.name!r} '
f'description={self.description!r} public={self.public} '
f'owner={self.owner!r}>'
)
@property
def icon(self) -> Optional[Asset]:
"""Optional[:class:`.Asset`]: Retrieves the application's icon asset, if any."""
if self._icon is None:
return None
return Asset._from_icon(self._state, self.id, self._icon, path='app')
def guild(self) -> Optional[Guild]:
"""Optional[:class:`Guild`]: If this application is a game sold on Discord,
this field will be the guild to which it has been linked.
"""
return self._state._get_guild(self.guild_id)
async def edit(
self,
*,
name: str = MISSING,
description: Optional[str] = MISSING,
icon: Optional[bytes] = MISSING,
cover_image: Optional[bytes] = MISSING,
tags: List[str] = MISSING,
terms_of_service_url: Optional[str] = MISSING,
privacy_policy_url: Optional[str] = MISSING,
interactions_endpoint_url: Optional[str] = MISSING,
redirect_uris: List[str] = MISSING,
rpc_origins: List[str] = MISSING,
public: bool = MISSING,
require_code_grant: bool = MISSING,
flags: ApplicationFlags = MISSING,
team: Snowflake = MISSING,
) -> None:
"""|coro|
Edits the application.
Parameters
-----------
name: :class:`str`
The name of the application.
description: :class:`str`
The description of the application.
icon: Optional[:class:`bytes`]
The icon of the application.
cover_image: Optional[:class:`bytes`]
The cover image of the application.
tags: List[:class:`str`]
A list of tags that describe the application.
terms_of_service_url: Optional[:class:`str`]
The URL to the terms of service of the application.
privacy_policy_url: Optional[:class:`str`]
The URL to the privacy policy of the application.
interactions_endpoint_url: Optional[:class:`str`]
The URL interactions will be sent to, if set.
redirect_uris: List[:class:`str`]
A list of redirect URIs authorized for this application.
rpc_origins: List[:class:`str`]
A list of RPC origins authorized for this application.
public: :class:`bool`
Whether the application is public or not.
require_code_grant: :class:`bool`
Whether the application requires a code grant or not.
flags: :class:`ApplicationFlags`
The flags of the application.
team: :class:`Snowflake`
The team to transfer the application to.
Raises
-------
Forbidden
You do not have permissions to edit this application.
HTTPException
Editing the application failed.
"""
payload = {}
if name is not MISSING:
payload['name'] = name or ''
if description is not MISSING:
payload['description'] = description or ''
if icon is not MISSING:
if icon is not None:
payload['icon'] = utils._bytes_to_base64_data(icon)
else:
payload['icon'] = ''
if cover_image is not MISSING:
if cover_image is not None:
payload['cover_image'] = utils._bytes_to_base64_data(cover_image)
else:
payload['cover_image'] = ''
if tags is not MISSING:
payload['tags'] = tags
if terms_of_service_url is not MISSING:
payload['terms_of_service_url'] = terms_of_service_url or ''
if privacy_policy_url is not MISSING:
payload['privacy_policy_url'] = privacy_policy_url or ''
if interactions_endpoint_url is not MISSING:
payload['interactions_endpoint_url'] = interactions_endpoint_url or ''
if redirect_uris is not MISSING:
payload['redirect_uris'] = redirect_uris
if rpc_origins is not MISSING:
payload['rpc_origins'] = rpc_origins
if public is not MISSING:
payload['integration_public'] = public
if require_code_grant is not MISSING:
payload['integration_require_code_grant'] = require_code_grant
if flags is not MISSING:
payload['flags'] = flags.value
data = await self._state.http.edit_application(self.id, payload)
if team is not MISSING:
data = await self._state.http.transfer_application(self.id, team.id)
self._update(data)
async def reset_secret(self) -> None:
"""|coro|
Resets the application's secret.
Raises
------
Forbidden
You do not have permissions to reset the secret.
HTTPException
Resetting the secret failed.
"""
data = await self._state.http.reset_secret(self.id)
self._update(data)
async def create_bot(self) -> ApplicationBot:
"""|coro|
Creates a bot attached to this application.
Raises
------
Forbidden
You do not have permissions to create bots.
HTTPException
Creating the bot failed.
Returns
-------
:class:`ApplicationBot`
The newly created bot.
"""
state = self._state
data = await state.http.botify_app(self.id)
data['public'] = self.public
data['require_code_grant'] = self.require_code_grant
bot = ApplicationBot(data=data, state=state, application=self)
self.bot = bot
return bot

542
discord/client.py
View File

@ -29,18 +29,18 @@ import logging
import signal
import sys
import traceback
from typing import Any, Callable, Coroutine, Dict, Generator, List, Optional, Sequence, TYPE_CHECKING, Tuple, TypeVar, Union
from typing import Any, Callable, Coroutine, Dict, Generator, List, Optional, overload, Sequence, TYPE_CHECKING, Tuple, TypeVar, Union
import aiohttp
from .user import User, ClientUser, Note
from .user import BaseUser, User, ClientUser, Note
from .invite import Invite
from .template import Template
from .widget import Widget
from .guild import Guild
from .emoji import Emoji
from .channel import _private_channel_factory, _threaded_channel_factory, GroupChannel, PartialMessageable
from .enums import ChannelType, Status, VoiceRegion, try_enum
from .enums import ActivityType, ChannelType, Status, VoiceRegion, InviteType, try_enum
from .mentions import AllowedMentions
from .errors import *
from .gateway import *
@ -54,19 +54,20 @@ from .utils import MISSING
from .object import Object
from .backoff import ExponentialBackoff
from .webhook import Webhook
from .iterators import GuildIterator
from .appinfo import AppInfo
from .appinfo import Application
from .stage_instance import StageInstance
from .threads import Thread
from .sticker import GuildSticker, StandardSticker, StickerPack, _sticker_factory
from .profile import UserProfile
from .connections import Connection
from .team import Team
if TYPE_CHECKING:
from .abc import SnowflakeTime, PrivateChannel, GuildChannel, Snowflake
from .abc import PrivateChannel, GuildChannel, Snowflake
from .channel import DMChannel
from .message import Message
from .member import Member
from .relationship import Relationship
from .voice_client import VoiceProtocol
__all__ = (
@ -109,6 +110,7 @@ def _cleanup_loop(loop: asyncio.AbstractEventLoop) -> None:
_log.info('Closing the event loop.')
loop.close()
class Client:
r"""Represents a client connection that connects to Discord.
This class is used to interact with the Discord WebSocket and API.
@ -183,6 +185,9 @@ class Client:
To enable these events, this must be set to ``True``. Defaults to ``False``.
.. versionadded:: 2.0
sync_presence: :class:`bool`
Whether to keep presences up-to-date across clients.
The default behavior is ``True`` (what the client does).
Attributes
-----------
@ -218,10 +223,21 @@ class Client:
}
self._enable_debug_events: bool = options.pop('enable_debug_events', False)
self._sync_presences: bool = options.pop('sync_presence', True)
self._connection: ConnectionState = self._get_state(**options)
self._closed: bool = False
self._ready: asyncio.Event = asyncio.Event()
self._client_status: Dict[Optional[str], str] = {
None: 'offline',
'this': 'offline',
}
self._client_activities: Dict[Optional[str], Tuple[ActivityTypes, ...]] = {
None: None,
'this': None,
}
self._session_count = 1
if VoiceClient.warn_nacl:
VoiceClient.warn_nacl = False
_log.warning('PyNaCl is not installed, voice will NOT be supported.')
@ -238,10 +254,11 @@ class Client:
def _handle_connect(self) -> None:
state = self._connection
activity = create_activity(state._activity)
status = state._status and try_enum(Status, state._status)
if status is not None or activity is not None:
self.loop.create_task(self.change_presence(activity=activity, status=status))
activities = self.initial_activities
status = self.initial_status
if status is None:
status = getattr(state.settings, 'status', None)
self.loop.create_task(self.change_presence(activities=activities, status=status))
@property
def latency(self) -> float:
@ -379,6 +396,20 @@ class Client:
print(f'Ignoring exception in {event_method}', file=sys.stderr)
traceback.print_exc()
async def on_internal_settings_update(self, old_settings, new_settings):
if not self._sync_presences:
return
if old_settings._status == new_settings._status and old_settings._custom_status == new_settings._custom_status:
return # Nothing changed
status = new_settings.status
activities = [a for a in self.activities if a.type != ActivityType.custom]
if (activity := new_settings.custom_activity) is not None:
activities.append(activity)
await self.change_presence(status=status, activities=activities, edit_settings=False)
# Hooks
async def _call_before_identify_hook(self, *, initial: bool = False) -> None:
@ -439,7 +470,7 @@ class Client:
state = self._connection
data = await state.http.static_login(token.strip())
state.analytics_token = data.get('analytics_token', '')
self._connection.user = ClientUser(state=state, data=data)
state.user = ClientUser(state=state, data=data)
async def connect(self, *, reconnect: bool = True) -> None:
"""|coro|
@ -507,8 +538,8 @@ class Client:
# We should only get this when an unhandled close code happens,
# such as a clean disconnect (1000) or a bad state (bad token, etc)
# Sometimes, discord sends us 1000 for unknown reasons so we should reconnect
# regardless and rely on is_closed instead
# Sometimes, Discord sends us 1000 for unknown reasons so we should
# reconnect regardless and rely on is_closed instead
if isinstance(exc, ConnectionClosed):
if exc.code != 1000:
await self.close()
@ -634,38 +665,55 @@ class Client:
@property
def voice_client(self) -> Optional[VoiceProtocol]:
"""Optional[:class:`VoiceProtocol`]: Returns the :class:`VoiceProtocol` associated with private calls, if any."""
return self._connection._get_voice_client(self.user.id)
return self._connection._get_voice_client(self._connection.self_id)
@property
def activity(self) -> Optional[ActivityTypes]:
"""Optional[:class:`.BaseActivity`]: The activity being used upon
logging in.
def initial_activity(self) -> Optional[ActivityTypes]:
"""Optional[:class:`.BaseActivity`]: The primary activity set upon logging in.
.. note::
The client may be setting multiple activities, these can be accessed under :attr:`initial_activities`.
"""
return create_activity(self._connection._activity)
return create_activity(self._connection._activities[0]) if self._connection._activities else None
@activity.setter
def activity(self, value: Optional[ActivityTypes]) -> None:
@initial_activity.setter
def initial_activity(self, value: Optional[ActivityTypes]) -> None:
if value is None:
self._connection._activity = None
self._connection._activities = []
elif isinstance(value, BaseActivity):
# ConnectionState._activity is typehinted as ActivityPayload, we're passing Dict[str, Any]
self._connection._activity = value.to_dict() # type: ignore
# ConnectionState._activities is typehinted as List[ActivityPayload], we're passing List[Dict[str, Any]]
self._connection._activities = [value.to_dict()] # type: ignore
else:
raise TypeError('activity must derive from BaseActivity')
@property
def initial_activities(self) -> List[ActivityTypes]:
"""List[:class:`.BaseActivity`]: The activities set upon logging in."""
return [create_activity(activity) for activity in self._connection._activities]
@initial_activities.setter
def initial_activities(self, values: List[ActivityTypes]) -> None:
if not values:
self._connection._activities = []
elif all(isinstance(value, BaseActivity) for value in values):
# ConnectionState._activities is typehinted as List[ActivityPayload], we're passing List[Dict[str, Any]]
self._connection._activities = [value.to_dict() for value in values] # type: ignore
else:
raise TypeError('activity must derive from BaseActivity')
@property
def status(self):
""":class:`.Status`:
The status being used upon logging on to Discord.
def initial_status(self):
"""Optional[:class:`.Status`]: The status set upon logging in.
.. versionadded: 2.0
.. versionadded:: 2.0
"""
if self._connection._status in set(state.value for state in Status):
if self._connection._status in {state.value for state in Status}:
return Status(self._connection._status)
return Status.online
return
@status.setter
def status(self, value):
@initial_status.setter
def initial_status(self, value):
if value is Status.offline:
self._connection._status = 'invisible'
elif isinstance(value, Status):
@ -673,6 +721,151 @@ class Client:
else:
raise TypeError('status must derive from Status')
@property
def status(self) -> Status:
""":class:`Status`: The user's overall status.
.. versionadded:: 2.0
"""
status = try_enum(Status, self._client_status[None])
if status is Status.offline and not self.is_closed():
status = getattr(self._connection.settings, 'status', status)
return status
@property
def raw_status(self) -> str:
""":class:`str`: The user's overall status as a string value.
.. versionadded:: 2.0
"""
return str(self.status)
@status.setter
def status(self, value: Status) -> None:
# Internal use only
self._client_status[None] = str(value)
@property
def mobile_status(self) -> Status:
""":class:`Status`: The user's status on a mobile device, if applicable.
.. versionadded:: 2.0
"""
return try_enum(Status, self._client_status.get('mobile', 'offline'))
@property
def desktop_status(self) -> Status:
""":class:`Status`: The user's status on the desktop client, if applicable.
.. versionadded:: 2.0
"""
return try_enum(Status, self._client_status.get('desktop', 'offline'))
@property
def web_status(self) -> Status:
""":class:`Status`: The user's status on the web client, if applicable.
.. versionadded:: 2.0
"""
return try_enum(Status, self._client_status.get('web', 'offline'))
@property
def client_status(self) -> Status:
""":class:`Status`: The library's status.
.. versionadded:: 2.0
"""
status = try_enum(Status, self._client_status['this'])
if status is Status.offline and not self.is_closed():
status = getattr(self._connection.settings, 'status', status)
return status
def is_on_mobile(self) -> bool:
""":class:`bool`: A helper function that determines if a member is active on a mobile device.
.. versionadded:: 2.0
"""
return 'mobile' in self._client_status
@property
def activities(self) -> Tuple[ActivityTypes]:
"""Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the activities
the client is currently doing.
.. versionadded:: 2.0
.. note::
Due to a Discord API limitation, this may be ``None`` if
the user is listening to a song on Spotify with a title longer
than 128 characters. See :issue:`1738` for more information.
"""
activities = tuple(map(create_activity, self._client_activities[None]))
if activities is None and not self.is_closed():
activities = getattr(self._connection.settings, 'custom_activity', [])
activities = [activities] if activities else activities
return activities
@property
def activity(self) -> Optional[ActivityTypes]:
"""Optional[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the primary
activity the client is currently doing. Could be ``None`` if no activity is being done.
.. versionadded:: 2.0
.. note::
Due to a Discord API limitation, this may be ``None`` if
the user is listening to a song on Spotify with a title longer
than 128 characters. See :issue:`1738` for more information.
.. note::
The client may have multiple activities, these can be accessed under :attr:`activities`.
"""
if (activities := self.activities):
return activities[0]
@property
def mobile_activities(self) -> Tuple[ActivityTypes]:
"""Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the activities
the client is currently doing on a mobile device, if applicable.
.. versionadded:: 2.0
"""
return tuple(map(create_activity, self._client_activities.get('mobile', [])))
@property
def desktop_activities(self) -> Tuple[ActivityTypes]:
"""Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the activities
the client is currently doing on the desktop client, if applicable.
.. versionadded:: 2.0
"""
return tuple(map(create_activity, self._client_activities.get('desktop', [])))
@property
def web_activities(self) -> Tuple[ActivityTypes]:
"""Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the activities
the client is currently doing on the web client, if applicable.
.. versionadded:: 2.0
"""
return tuple(map(create_activity, self._client_activities.get('web', [])))
@property
def client_activities(self) -> Tuple[ActivityTypes]:
"""Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the activities
the client is currently doing through this library, if applicable.
.. versionadded:: 2.0
"""
activities = tuple(map(create_activity, self._client_activities.get('this', [])))
if activities is None and not self.is_closed():
activities = getattr(self._connection.settings, 'custom_activity', [])
activities = [activities] if activities else activities
return activities
@property
def allowed_mentions(self) -> Optional[AllowedMentions]:
"""Optional[:class:`~discord.AllowedMentions`]: The allowed mention configuration.
@ -1005,13 +1198,19 @@ class Client:
self,
*,
activity: Optional[BaseActivity] = None,
activities: Optional[List[BaseActivity]] = None,
status: Optional[Status] = None,
afk: bool = False
afk: bool = False,
edit_settings: bool = True,
):
"""|coro|
Changes the client's presence.
.. versionchanged:: 2.0
Edits are no longer in place most of the time.
Added option to update settings.
Example
---------
@ -1023,51 +1222,62 @@ class Client:
Parameters
----------
activity: Optional[:class:`.BaseActivity`]
The activity being done. ``None`` if no currently active activity is done.
The activity being done. ``None`` if no activity is done.
activities: Optional[List[:class:`BaseActivity`]]
A list of the activities being done. ``None`` if no activities
are done. Cannot be sent with ``activity``.
status: Optional[:class:`.Status`]
Indicates what status to change to. If ``None``, then
:attr:`.Status.online` is used.
afk: Optional[:class:`bool`]
afk: :class:`bool`
Indicates if you are going AFK. This allows the Discord
client to know how to handle push notifications better
for you in case you are actually idle and not lying.
edit_settings: :class:`bool`
Whether to update the settings with the new status and/or
custom activity. This will broadcast the change and cause
all connected (official) clients to change presence as well.
Defaults to ``True``. Required for setting/editing expires_at
for custom activities.
It's not recommended to change this.
Raises
------
:exc:`.InvalidArgument`
If the ``activity`` parameter is not the proper type.
The ``activity`` parameter is not the proper type.
Both ``activity`` and ``activities`` were passed.
"""
if activity and activities:
raise InvalidArgument('Cannot pass both activity and activities')
activities = activities or activity and [activity]
if activities is None:
activities = []
if status is None:
status_str = 'online'
status = Status.online
elif status is Status.offline:
status_str = 'invisible'
status = Status.offline
else:
breakpoint()
status_str = str(status)
status = Status.invisible
await self.ws.change_presence(activity=activity, status=status_str, afk=afk)
await self.ws.change_presence(status=status, activities=activities, afk=afk)
# TODO: do the same for custom status and check which comes first
if status:
try:
await self._connection.user.edit_settings(status=status)
except Exception: # Not essential to actually changing status...
pass
if edit_settings:
custom_activity = None
for guild in self._connection.guilds:
me = guild.me
if me is None:
continue
for activity in activities:
if getattr(activity, 'type', None) is ActivityType.custom:
custom_activity = activity
if activity is not None:
me.activities = (activity,)
else:
me.activities = ()
payload: Dict[str, Any] = {'status': status}
payload['custom_activity'] = custom_activity
await self.user.edit_settings(**payload)
me.status = status
status_str = str(status)
activities_tuple = tuple(a.to_dict() for a in activities)
self._client_status['this'] = str(status)
self._client_activities['this'] = activities_tuple
if self._session_count <= 1:
self._client_status[None] = status_str
self._client_activities[None] = self._client_activities['this'] = activities_tuple
async def change_voice_state(
self,
@ -1080,9 +1290,9 @@ class Client:
) -> None:
"""|coro|
Changes client's voice state in the guild.
Changes client's private channel voice state.
.. versionadded:: 1.4
.. versionadded:: 1.10
Parameters
-----------
@ -1111,14 +1321,12 @@ class Client:
# Guild stuff
def fetch_guilds(
async def fetch_guilds(
self,
*,
limit: Optional[int] = None,
before: SnowflakeTime = None,
after: SnowflakeTime = None
) -> GuildIterator:
"""Retrieves an :class:`.AsyncIterator` that enables receiving your guilds.
with_counts: bool = True
) -> List[Guild]:
"""Retrieves all your your guilds.
.. note::
@ -1129,47 +1337,25 @@ class Client:
This method is an API call. For general usage, consider :attr:`guilds` instead.
Examples
---------
Usage ::
async for guild in client.fetch_guilds():
print(guild.name)
Flattening into a list ::
guilds = await client.fetch_guilds().flatten()
# guilds is now a list of Guild...
All parameters are optional.
Parameters
-----------
limit: Optional[:class:`int`]
The number of guilds to retrieve.
If ``None``, it retrieves every guild you have access to.
Defaults to ``None``.
before: Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]
Retrieves guilds before this date or object.
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: Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]
Retrieve guilds after this date or object.
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.
with_counts: :class:`bool`
Whether to return approximate :attr:`.Guild.member_count` and :attr:`.Guild.presence_count`.
Defaults to ``True``.
Raises
------
:exc:`.HTTPException`
Getting the guilds failed.
Yields
Returns
--------
:class:`.Guild`
The guild with the guild data parsed.
List[:class:`.Guild`]
A list of all your guilds.
"""
return GuildIterator(self, limit=limit, before=before, after=after)
state = self._connection
guilds = await state.http.get_guilds(with_counts)
return [Guild(data=data, state=state) for data in guilds]
async def fetch_template(self, code: Union[Template, str]) -> Template:
"""|coro|
@ -1197,7 +1383,7 @@ class Client:
data = await self.http.get_template(code)
return Template(data=data, state=self._connection) # type: ignore
async def fetch_guild(self, guild_id: int, /) -> Guild:
async def fetch_guild(self, guild_id: int, /, *, with_counts: bool = True) -> Guild:
"""|coro|
Retrieves a :class:`.Guild` from an ID.
@ -1227,13 +1413,13 @@ class Client:
:class:`.Guild`
The guild from the ID.
"""
data = await self.http.get_guild(guild_id)
data = await self.http.get_guild(guild_id, with_counts)
return Guild(data=data, state=self._connection)
async def create_guild(
self,
*,
name: str,
*,
icon: bytes = MISSING,
code: str = MISSING,
) -> Guild:
@ -1245,7 +1431,7 @@ class Client:
----------
name: :class:`str`
The name of the guild.
icon: Optional[:class:`bytes`]
icon: :class:`bytes`
The :term:`py:bytes-like object` representing the icon. See :meth:`.ClientUser.edit`
for more details on what is expected.
code: :class:`str`
@ -1280,7 +1466,7 @@ class Client:
async def fetch_stage_instance(self, channel_id: int, /) -> StageInstance:
"""|coro|
Gets a :class:`.StageInstance` for a stage channel id.
Gets a :class:`.StageInstance` for a stage channel ID.
.. versionadded:: 2.0
@ -1375,10 +1561,11 @@ class Client:
invite_id = utils.resolve_invite(invite)
await self.http.delete_invite(invite_id)
async def accept_invite(self, invite: Union[Invite, str]) -> Guild:
async def accept_invite(self, invite: Union[Invite, str]) -> Union[Guild, User, GroupChannel]:
"""|coro|
Accepts an invite and joins a guild.
Uses an invite.
Either joins a guild, joins a group DM, or adds a friend.
.. versionadded:: 1.9
@ -1390,7 +1577,7 @@ class Client:
Raises
------
:exc:`.HTTPException`
Joining the guild failed.
Using the invite failed.
Returns
-------
@ -1402,10 +1589,23 @@ class Client:
if not isinstance(invite, Invite):
invite = await self.fetch_invite(invite, with_counts=False, with_expiration=False)
data = await self.http.accept_invite(invite.code, guild_id=invite.guild.id, channel_id=invite.channel.id, channel_type=invite.channel.type.value)
return Guild(data=data['guild'], state=self._connection)
use_invite = accept_invite
state = self._connection
type = invite.type
if (message := invite._message):
kwargs = {'message': message}
else:
kwargs = {
'guild_id': getattr(invite.guild, 'id', MISSING),
'channel_id': getattr(invite.channel, 'id', MISSING),
'channel_type': getattr(invite.channel, 'type', MISSING),
}
data = await state.http.accept_invite(invite.code, type, **kwargs)
if type is InviteType.guild:
return Guild(data=data['guild'], state=state)
elif type is InviteType.group_dm:
return GroupChannel(data=data['channel'], state=state, me=state.user) # type: ignore
else:
return User(data=data['inviter'], state=state)
# Miscellaneous stuff
@ -1449,6 +1649,11 @@ class Client:
This method is an API call. If you have member cache enabled, consider :meth:`get_user` instead.
.. warning::
This API route is not used by the Discord client and may increase your chances at getting detected.
Consider :meth:`fetch_user_profile` if you share a guild/relationship with the user.
Parameters
-----------
user_id: :class:`int`
@ -1609,7 +1814,7 @@ class Client:
return cls(state=self._connection, data=data) # type: ignore
async def fetch_sticker_packs(
self, *, country='US', locale='en-US', payment_source_id: int = MISSING
self, *, country: str = 'US', locale: str = 'en-US', payment_source_id: int = MISSING
) -> List[StickerPack]:
"""|coro|
@ -1797,3 +2002,120 @@ class Client:
state = self._connection
data = await state.http.start_group(users)
return GroupChannel(me=self.user, data=data, state=state)
@overload
async def send_friend_request(self, user: BaseUser) -> Relationship:
...
@overload
async def send_friend_request(self, user: str) -> Relationship:
...
@overload
async def send_friend_request(self, username: str, discriminator: Union[int, str]) -> Relationship:
...
async def send_friend_request(self, *args: Union[BaseUser, int, str]) -> Relationship:
"""|coro|
Sends a friend request to another user.
This function can be used in multiple ways.
.. code-block:: python
# Passing a user object:
await client.send_friend_request(user)
# Passing a stringified user:
await client.send_friend_request('Jake#0001')
# Passing a username and discriminator:
await client.send_friend_request('Jake', '0001')
Parameters
-----------
user: Union[:class:`User`, :class:`str`]
The user to send the friend request to.
username: :class:`str`
The username of the user to send the friend request to.
discriminator: :class:`str`
The discriminator of the user to send the friend request to.
More than 2 parameters or less than 1 parameter raises a :exc:`TypeError`.
Raises
-------
:exc:`.Forbidden`
Not allowed to send a friend request to this user.
:exc:`.HTTPException`
Sending the friend request failed.
Returns
-------
:class:`.Relationship`
The new relationship.
"""
username: str
discrim: Union[str, int]
if len(args) == 1:
user = args[0]
if isinstance(user, BaseUser):
user = str(user)
username, discrim = user.split('#') # type: ignore
elif len(args) == 2:
username, discrim = args # type: ignore
else:
raise TypeError(f'send_friend_request() takes 1 or 2 arguments but {len(args)} were given')
state = self._connection
data = await state.http.send_friend_request(username, discrim)
return Relationship(state=state, data=data)
async def create_application(self, name: str):
"""|coro|
Creates an application.
Parameters
----------
name: :class:`str`
The name of the application.
Raises
-------
:exc:`.HTTPException`
Failed to create the application.
Returns
-------
:class:`.Application`
The newly-created application.
"""
state = self._connection
data = await state.http.create_app(name)
return Application(state=state, data=data)
async def create_team(self, name: str):
"""|coro|
Creates a team.
Parameters
----------
name: :class:`str`
The name of the team.
Raises
-------
:exc:`.HTTPException`
Failed to create the team.
Returns
-------
:class:`.Team`
The newly-created team.
"""
state = self._connection
data = await state.http.create_team(name)
return Team(state=state, data=data)

17
discord/gateway.py
View File

@ -377,11 +377,12 @@ class DiscordWebSocket:
async def identify(self):
"""Sends the IDENTIFY packet."""
state = self._connection
payload = {
'op': self.IDENTIFY,
'd': {
'token': self.token,
'capabilities': 125,
'capabilities': 253,
'properties': self._super_properties,
'presence': {
'status': 'online',
@ -606,13 +607,13 @@ class DiscordWebSocket:
if not self._can_handle_close():
raise ConnectionClosed(self.socket) from exc
async def change_presence(self, *, activity=None, status=None, since=0.0, afk=False):
if activity is not None:
if not isinstance(activity, BaseActivity):
async def change_presence(self, *, activities=None, status=None, since=0, afk=False):
if activities is not None:
if not all(isinstance(activity, BaseActivity) for activity in activities):
raise InvalidArgument('activity must derive from BaseActivity')
activity = [activity.to_dict()]
activities = [activity.to_dict() for activity in activities]
else:
activity = []
activities = []
if status == 'idle':
since = int(time.time() * 1000)
@ -620,10 +621,10 @@ class DiscordWebSocket:
payload = {
'op': self.PRESENCE,
'd': {
'activities': activity,
'activities': activities,
'afk': afk,
'since': since,
'status': status
'status': str(status)
}
}

48
discord/http.py
View File

@ -76,6 +76,7 @@ if TYPE_CHECKING:
user,
webhook,
widget,
team,
threads,
sticker,
welcome_screen,
@ -1883,10 +1884,17 @@ class HTTPClient:
def edit_application(self, app_id: Snowflake, payload) -> Response[appinfo.AppInfo]:
return self.request(Route('PATCH', '/applications/{app_id}', app_id=app_id), super_properties_to_track=True, json=payload)
def delete_application(self, app_id: Snowflake) -> Response[appinfo.AppInfo]:
def delete_application(self, app_id: Snowflake) -> Response[None]:
return self.request(Route('POST', '/applications/{app_id}/delete', app_id=app_id), super_properties_to_track=True)
def get_partial_application(self, app_id: Snowflake):
def transfer_application(self, app_id: Snowflake, team_id: Snowflake) -> Response[appinfo.AppInfo]:
payload = {
'team_id': team_id
}
return self.request(Route('POST', '/applications/{app_id}/transfer', app_id=app_id), json=payload, super_properties_to_track=True)
def get_partial_application(self, app_id: Snowflake) -> Response[appinfo.PartialAppInfo]:
return self.request(Route('GET', '/applications/{app_id}/rpc', app_id=app_id), auth=False)
def create_app(self, name: str):
@ -1894,7 +1902,7 @@ class HTTPClient:
'name': name
}
return self.request(Route('POST', '/applications'), json=payload)
return self.request(Route('POST', '/applications'), json=payload, super_properties_to_track=True)
def get_app_entitlements(self, app_id: Snowflake): # TODO: return type
r = Route('GET', '/users/@me/applications/{app_id}/entitlements', app_id=app_id)
@ -1914,12 +1922,42 @@ class HTTPClient:
def get_app_whitelist(self, app_id):
return self.request(Route('GET', '/oauth2/applications/{app_id}/allowlist', app_id=app_id), super_properties_to_track=True)
def get_teams(self): # TODO: return type
def create_team(self, name: str):
payload = {
'name': name
}
return self.request(Route('POST', '/teams'), json=payload, super_properties_to_track=True)
def get_teams(self) -> Response[List[team.Team]]:
return self.request(Route('GET', '/teams'), super_properties_to_track=True)
def get_team(self, team_id: Snowflake): # TODO: return type
def get_team(self, team_id: Snowflake) -> Response[team.Team]:
return self.request(Route('GET', '/teams/{team_id}', team_id=team_id), super_properties_to_track=True)
def edit_team(self, team_id: Snowflake, payload) -> Response[team.Team]:
return self.request(Route('PATCH', '/teams/{team_id}', team_id=team_id), json=payload, super_properties_to_track=True)
def delete_application(self, team_id: Snowflake) -> Response[None]:
return self.request(Route('POST', '/teams/{app_id}/delete', team_id=team_id), super_properties_to_track=True)
def get_team_applications(self, team_id: Snowflake) -> Response[List[appinfo.AppInfo]]:
return self.request(Route('GET', '/teams/{team_id}/applications', team_id=team_id), super_properties_to_track=True)
def get_team_members(self, team_id: Snowflake) -> Response[List[team.TeamMember]]:
return self.request(Route('GET', '/teams/{team_id}/members', team_id=team_id), super_properties_to_track=True)
def invite_team_member(self, team_id: Snowflake, username: str, discriminator: Snowflake):
payload = {
'username': username,
'discriminator': str(discriminator)
}
return self.request(Route('POST', '/teams/{team_id}/members', team_id=team_id), json=payload, super_properties_to_track=True)
def remove_team_member(self, team_id: Snowflake, user_id: Snowflake):
return self.request(Route('DELETE', '/teams/{team_id}/members/{user_id}', team_id=team_id, user_id=user_id), super_properties_to_track=True)
def botify_app(self, app_id: Snowflake):
return self.request(Route('POST', '/applications/{app_id}/bot', app_id=app_id), super_properties_to_track=True)

114
discord/iterators.py
View File

@ -40,7 +40,8 @@ __all__ = (
'ReactionIterator',
'HistoryIterator',
'AuditLogIterator',
'GuildIterator',
'CommandIterator',
'FakeCommandIterator',
)
if TYPE_CHECKING:
@ -496,117 +497,6 @@ class AuditLogIterator(_AsyncIterator['AuditLogEntry']):
await self.entries.put(AuditLogEntry(data=element, users=self._users, guild=self.guild))
class GuildIterator(_AsyncIterator['Guild']):
"""Iterator for receiving the client's guilds.
The guilds endpoint has the same two behaviours as described
in :class:`HistoryIterator`:
If ``before`` is specified, the guilds endpoint returns the ``limit``
newest guilds before ``before``, sorted with newest first. For filling over
100 guilds, update the ``before`` parameter to the oldest guild received.
Guilds will be returned in order by time.
If `after` is specified, it returns the ``limit`` oldest guilds after ``after``,
sorted with newest first.
Not that if both ``before`` and ``after`` are specified, ``before`` is ignored by the
guilds endpoint.
Parameters
-----------
bot: :class:`discord.Client`
The client to retrieve the guilds from.
limit: :class:`int`
Maximum number of guilds to retrieve.
before: Optional[Union[:class:`abc.Snowflake`, :class:`datetime.datetime`]]
Object before which all guilds must be.
after: Optional[Union[:class:`abc.Snowflake`, :class:`datetime.datetime`]]
Object after which all guilds must be.
"""
def __init__(self, bot, limit, before=None, after=None):
if isinstance(before, datetime.datetime):
before = Object(id=time_snowflake(before, high=False))
if isinstance(after, datetime.datetime):
after = Object(id=time_snowflake(after, high=True))
self.bot = bot
self.limit = limit
self.before = before
self.after = after
self._filter = None
self.state = self.bot._connection
self.get_guilds = self.bot.http.get_guilds
self.guilds = asyncio.Queue()
if self.before and self.after:
self._retrieve_guilds = self._retrieve_guilds_before_strategy # type: ignore
self._filter = lambda m: int(m['id']) > self.after.id
elif self.after:
self._retrieve_guilds = self._retrieve_guilds_after_strategy # type: ignore
else:
self._retrieve_guilds = self._retrieve_guilds_before_strategy # type: ignore
async def next(self) -> Guild:
if self.guilds.empty():
await self.fill_guilds()
try:
return self.guilds.get_nowait()
except asyncio.QueueEmpty:
raise NoMoreItems()
def _get_retrieve(self):
l = self.limit
if l is None or l > 200:
r = 200
else:
r = l
self.retrieve = r
return r > 0
def create_guild(self, data):
from .guild import Guild
return Guild(state=self.state, data=data)
async def fill_guilds(self):
if self._get_retrieve():
data = await self._retrieve_guilds(self.retrieve)
self.limit = 0 # Max amount of guilds a user can be in is 200
if self._filter:
data = filter(self._filter, data)
for element in data:
await self.guilds.put(self.create_guild(element))
async def _retrieve_guilds(self, retrieve) -> List[Guild]:
"""Retrieve guilds and update next parameters."""
raise NotImplementedError
async def _retrieve_guilds_before_strategy(self, retrieve):
"""Retrieve guilds using before parameter."""
before = self.before.id if self.before else None
data: List[GuildPayload] = await self.get_guilds(retrieve, before=before)
if len(data):
if self.limit is not None:
self.limit -= retrieve
self.before = Object(id=int(data[-1]['id']))
return data
async def _retrieve_guilds_after_strategy(self, retrieve):
"""Retrieve guilds using after parameter."""
after = self.after.id if self.after else None
data: List[GuildPayload] = await self.get_guilds(retrieve, after=after)
if len(data):
if self.limit is not None:
self.limit -= retrieve
self.after = Object(id=int(data[0]['id']))
return data
class ArchivedThreadIterator(_AsyncIterator['Thread']):
def __init__(
self,

116
discord/member.py
View File

@ -236,15 +236,6 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
joined_at: Optional[:class:`datetime.datetime`]
An aware datetime object that specifies the date and time in UTC that the member joined the guild.
If the member left and rejoined the guild, this will be the latest date. In certain cases, this can be ``None``.
activities: Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]
The activities that the user is currently doing.
.. note::
Due to a Discord API limitation, a user's Spotify activity may not appear
if they are listening to a song with a title longer
than 128 characters. See :issue:`1738` for more information.
guild: :class:`Guild`
The guild that the member belongs to.
nick: Optional[:class:`str`]
@ -262,7 +253,7 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
'_roles',
'joined_at',
'premium_since',
'activities',
'_activities',
'guild',
'pending',
'nick',
@ -271,6 +262,7 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
'_state',
'_avatar',
'_index', # Member list index
'_communication_disabled_until',
)
if TYPE_CHECKING:
@ -298,10 +290,11 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
self.premium_since: Optional[datetime.datetime] = utils.parse_time(data.get('premium_since'))
self._roles: utils.SnowflakeList = utils.SnowflakeList(map(int, data['roles']))
self._client_status: Dict[Optional[str], str] = {None: 'offline'}
self.activities: Tuple[ActivityTypes, ...] = tuple()
self._activities: Tuple[ActivityTypes, ...] = tuple()
self.nick: Optional[str] = data.get('nick', None)
self.pending: bool = data.get('pending', False)
self._avatar: Optional[str] = data.get('avatar')
self._communication_disabled_until: Optional[datetime.datetime] = utils.parse_time(data.get('communication_disabled_until'))
def __str__(self) -> str:
return str(self._user)
@ -333,6 +326,7 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
self._roles = utils.SnowflakeList(map(int, data['roles']))
self.nick = data.get('nick', None)
self.pending = data.get('pending', False)
self._communication_disabled_until = utils.parse_time(data.get('communication_disabled_until'))
@classmethod
def _try_upgrade(cls: Type[M], *, data: UserWithMemberPayload, guild: Guild, state: ConnectionState) -> Union[User, M]:
@ -356,9 +350,10 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
self.guild = member.guild
self.nick = member.nick
self.pending = member.pending
self.activities = member.activities
self._activities = member._activities
self._state = member._state
self._avatar = member._avatar
self._communication_disabled_until = member._communication_disabled_until
# Reference will not be copied unless necessary by PRESENCE_UPDATE
# See below
@ -366,8 +361,8 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
return self
def _update(self, data: MemberPayload) -> None:
# the nickname change is optional,
# if it isn't in the payload then it didn't change
# The nickname change is optional
# If it isn't in the payload then it didn't change
try:
self.nick = data['nick']
except KeyError:
@ -381,9 +376,13 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
self.premium_since = utils.parse_time(data.get('premium_since'))
self._roles = utils.SnowflakeList(map(int, data['roles']))
self._avatar = data.get('avatar')
self._communication_disabled_until = utils.parse_time(data.get('communication_disabled_until'))
def _presence_update(self, data: PartialPresenceUpdate, user: UserPayload) -> Optional[Tuple[User, User]]:
self.activities = tuple(map(create_activity, data['activities']))
if self._self:
return
self._activities = tuple(map(create_activity, data['activities']))
self._client_status = {
sys.intern(key): sys.intern(value) for key, value in data.get('client_status', {}).items() # type: ignore
}
@ -391,7 +390,6 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
if len(user) > 1:
return self._update_inner_user(user)
return
def _update_inner_user(self, user: UserPayload) -> Optional[Tuple[User, User]]:
u = self._user
@ -407,7 +405,8 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
@property
def status(self) -> Status:
""":class:`Status`: The member's overall status. If the value is unknown, then it will be a :class:`str` instead."""
return try_enum(Status, self._client_status[None])
client_status = self._client_status if not self._self else self._state.client._client_status
return try_enum(Status, client_status[None])
@property
def raw_status(self) -> str:
@ -415,31 +414,37 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
.. versionadded:: 1.5
"""
return self._client_status[None]
client_status = self._client_status if not self._self else self._state.client._client_status
return client_status[None]
@status.setter
def status(self, value: Status) -> None:
# Internal use only
self._client_status[None] = str(value)
client_status = self._client_status if not self._self else self._state.client._client_status
client_status[None] = str(value)
@property
def mobile_status(self) -> Status:
""":class:`Status`: The member's status on a mobile device, if applicable."""
return try_enum(Status, self._client_status.get('mobile', 'offline'))
client_status = self._client_status if not self._self else self._state.client._client_status
return try_enum(Status, client_status.get('mobile', 'offline'))
@property
def desktop_status(self) -> Status:
""":class:`Status`: The member's status on the desktop client, if applicable."""
return try_enum(Status, self._client_status.get('desktop', 'offline'))
client_status = self._client_status if not self._self else self._state.client._client_status
return try_enum(Status, client_status.get('desktop', 'offline'))
@property
def web_status(self) -> Status:
""":class:`Status`: The member's status on the web client, if applicable."""
return try_enum(Status, self._client_status.get('web', 'offline'))
client_status = self._client_status if not self._self else self._state.client._client_status
return try_enum(Status, client_status.get('web', 'offline'))
def is_on_mobile(self) -> bool:
""":class:`bool`: A helper function that determines if a member is active on a mobile device."""
return 'mobile' in self._client_status
client_status = self._client_status if not self._self else self._state.client._client_status
return 'mobile' in client_status
@property
def colour(self) -> Colour:
@ -526,6 +531,22 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
return None
return Asset._from_guild_avatar(self._state, self.guild.id, self.id, self._avatar)
@property
def activities(self) -> Tuple[ActivityTypes, ...]:
"""Tuple[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the activities that
the user is currently doing.
.. note::
Due to a Discord API limitation, a user's Spotify activity may not appear
if they are listening to a song with a title longer
than 128 characters. See :issue:`1738` for more information.
"""
if self._self:
return self._state.client.activities
return self._activities
@property
def activity(self) -> Optional[ActivityTypes]:
"""Optional[Union[:class:`BaseActivity`, :class:`Spotify`]]: Returns the primary
@ -608,6 +629,43 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
"""Optional[:class:`VoiceState`]: Returns the member's current voice state."""
return self.guild._voice_state_for(self._user.id)
@property
def timed_out(self) -> bool:
""":class:`bool`: Returns whether the member is timed out.
.. versionadded:: 2.0
"""
return bool(self.timed_out_until)
@property
def timed_out_until(self) -> Optional[datetime.datetime]:
"""Optional[:class:`datetime.datetime`]: Returns an aware datetime object that
specifies the date and time in UTC until the member is timed out.
There is an alias for this called :attr:`timeout_until`.
.. versionadded:: 2.0
"""
until = self._communication_disabled_until
if until is None:
return
return until if until > utils.utcnow() else None
@property
def timeout_until(self) -> Optional[datetime.datetime]:
"""Optional[:class:`datetime.datetime`]: Returns an aware datetime object that
specifies the date and time in UTC until the member is timed out.
This is an alias of :attr:`timed_out_until`.
.. versionadded:: 2.0
"""
return self.timed_out_until
@property
def _self(self) -> bool:
return self._user.id == self._state.self_id
async def ban(
self,
*,
@ -643,8 +701,9 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
suppress: bool = MISSING,
roles: List[discord.abc.Snowflake] = MISSING,
voice_channel: Optional[VocalGuildChannel] = MISSING,
reason: Optional[str] = None,
avatar: Optional[bytes] = MISSING,
timeout_until: Optional[datetime.datetime] = MISSING,
reason: Optional[str] = None,
) -> Optional[Member]:
"""|coro|
@ -665,6 +724,8 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
+---------------+--------------------------------------+
| voice_channel | :attr:`Permissions.move_members` |
+---------------+--------------------------------------+
| timeout_until | :attr:`Permissions.moderate_members` |
+---------------+--------------------------------------+
All parameters are optional.
@ -702,6 +763,8 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
avatar: Optional[:class:`bytes`]
The member's new guild avatar. Pass ``None`` to remove the avatar.
You can only change your own guild avatar.
timeout_until: Optional[:class:`datetime.datetime`]
A datetime object denoting how long this member should be in timeout for.
reason: Optional[:class:`str`]
The reason for editing this member. Shows up on the audit log.
@ -728,7 +791,7 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
payload['nick'] = nick
if avatar is not MISSING:
payload['avatar'] = utils._bytes_to_base64_data(avatar) # type: ignore
payload['avatar'] = utils._bytes_to_base64_data(avatar) if avatar is not None else None
if me and payload:
data = await http.edit_me(**payload)
@ -762,6 +825,9 @@ class Member(discord.abc.Messageable, discord.abc.Connectable, _UserTag):
if roles is not MISSING:
payload['roles'] = tuple(r.id for r in roles)
if timeout_until is not MISSING:
payload['communication_disabled_until'] = timeout_until.isoformat() if timeout_until is not None else None
if payload:
data = await http.edit_member(guild_id, self.id, reason=reason, **payload)

120
discord/state.py
View File

@ -34,17 +34,18 @@ import inspect
import time
import os
import random
from sys import intern
from .errors import NotFound
from .guild import CommandCounts, Guild
from .activity import BaseActivity
from .activity import BaseActivity, create_activity
from .user import User, ClientUser
from .emoji import Emoji
from .mentions import AllowedMentions
from .partial_emoji import PartialEmoji
from .message import Message
from .channel import *
from .channel import _channel_factory
from .channel import _channel_factory, _private_channel_factory
from .raw_models import *
from .member import Member
from .relationship import Relationship
@ -57,11 +58,10 @@ from .integrations import _integration_factory
from .stage_instance import StageInstance
from .threads import Thread, ThreadMember
from .sticker import GuildSticker
from .settings import UserSettings
from .settings import UserSettings, GuildSettings
from .tracking import Tracking
from .interactions import Interaction
if TYPE_CHECKING:
from .abc import PrivateChannel
from .message import MessageableChannel
@ -86,6 +86,8 @@ if TYPE_CHECKING:
CS = TypeVar('CS', bound='ConnectionState')
Channel = Union[GuildChannel, VocalGuildChannel, PrivateChannel, PartialMessageable]
MISSING = utils.MISSING
class ChunkRequest:
def __init__(
@ -100,7 +102,7 @@ class ChunkRequest:
self.resolver: Callable[[int], Any] = resolver
self.loop: asyncio.AbstractEventLoop = loop
self.cache: bool = cache
self.nonce: str = os.urandom(16).hex()
self.nonce: str = str(utils.time_snowflake(utils.utcnow()))
self.buffer: List[Member] = []
self.waiters: List[asyncio.Future[List[Member]]] = []
@ -146,11 +148,6 @@ async def logging_coroutine(coroutine: Coroutine[Any, Any, T], *, info: str) ->
class ConnectionState:
if TYPE_CHECKING:
_get_websocket: Callable[..., DiscordWebSocket]
_get_client: Callable[..., Client]
_parsers: Dict[str, Callable[[Dict[str, Any]], None]]
def __init__(
self,
*,
@ -176,19 +173,21 @@ class ConnectionState:
self.heartbeat_timeout: float = options.get('heartbeat_timeout', 60.0)
allowed_mentions = options.get('allowed_mentions')
if allowed_mentions is not None and not isinstance(allowed_mentions, AllowedMentions):
raise TypeError('allowed_mentions parameter must be AllowedMentions')
self.allowed_mentions: Optional[AllowedMentions] = allowed_mentions
self._chunk_requests: Dict[Union[int, str], ChunkRequest] = {}
activity = options.get('activity', None)
if activity:
if not isinstance(activity, BaseActivity):
raise TypeError('activity parameter must derive from BaseActivity.')
activities = options.get('activities', [])
if not activities:
activity = options.get('activity')
if activity is not None:
activities = [activity]
activity = activity.to_dict()
if not all(isinstance(activity, BaseActivity) for activity in activities):
raise TypeError('activity parameter must derive from BaseActivity.')
activities = [activity.to_dict() for activity in activities]
status = options.get('status', None)
if status:
@ -217,17 +216,18 @@ class ConnectionState:
raise TypeError(f'member_cache_flags parameter must be MemberCacheFlags not {type(cache_flags)!r}')
self.member_cache_flags: MemberCacheFlags = cache_flags
self._activity: Optional[ActivityPayload] = activity
self._activities: List[ActivityPayload] = activities
self._status: Optional[str] = status
if cache_flags._empty:
self.store_user = self.create_user # type: ignore
self.deref_user = self.deref_user_no_intents # type: ignore
self.deref_user = lambda _: None # type: ignore
self.parsers = parsers = {}
parsers = {}
for attr, func in inspect.getmembers(self):
if attr.startswith('parse_'):
parsers[attr[6:].upper()] = func
self.parsers: Dict[str, Callable[[Dict[str, Any]], None]] = parsers
self.clear()
@ -237,7 +237,6 @@ class ConnectionState:
self.consents: Optional[Tracking] = None
self.analytics_token: Optional[str] = None
self.session_id: Optional[str] = None
self.connected_accounts: Optional[List[dict]] = None
self.preferred_region: Optional[VoiceRegion] = None
# Originally, this code used WeakValueDictionary to maintain references to the
# global user mapping
@ -379,9 +378,6 @@ class ConnectionState:
def create_user(self, data: UserPayload) -> User:
return User(state=self, data=data)
def deref_user_no_intents(self, user_id: int) -> None:
pass
def get_user(self, id: Optional[int]) -> Optional[User]:
# The keys of self._users are ints
return self._users.get(id) # type: ignore
@ -539,7 +535,7 @@ class ConnectionState:
except NotFound:
pass
def request_guild(self, guild_id: int) -> None:
def request_guild(self, guild_id: int) -> Coroutine:
return self.ws.request_lazy_guild(guild_id, typing=True, activities=True, threads=True)
def chunker(
@ -659,7 +655,7 @@ class ConnectionState:
# Private channel parsing
for pm in data.get('private_channels', []):
factory, _ = _channel_factory(pm['type'])
factory, _ = _private_channel_factory(pm['type'])
if 'recipients' not in pm:
pm['recipients'] = [temp_users[int(u_id)] for u_id in pm.pop('recipient_ids')]
self._add_private_channel(factory(me=user, data=pm, state=self))
@ -670,7 +666,7 @@ class ConnectionState:
region = data.get('geo_ordered_rtc_regions', ['us-west'])[0]
self.preferred_region = try_enum(VoiceRegion, region)
self.settings = UserSettings(data=data.get('user_settings', {}), state=self)
self.consents = Tracking(data.get('consents', {}))
self.consents = Tracking(data=data.get('consents', {}), state=self)
if 'required_action' in data: # Locked more than likely
self.parse_user_required_action_update(data)
@ -845,20 +841,80 @@ class ConnectionState:
def parse_user_settings_update(self, data) -> None:
new_settings = self.settings
old_settings = copy.copy(new_settings)
new_settings._update(data)
new_settings._update(data) # type: ignore
self.dispatch('settings_update', old_settings, new_settings)
self.dispatch('internal_settings_update', old_settings, new_settings)
def parse_user_guild_settings_update(self, data) -> None:
guild = self.get_guild(int(data['guild_id']))
new_settings = guild.notification_settings
old_settings = copy.copy(new_settings)
new_settings._update(data)
self.dispatch('guild_settings_update', old_settings, new_settings)
guild_id = int(data['guild_id'])
guild = self._get_guild(guild_id)
if guild is None:
_log.debug('USER_GUILD_SETTINGS_UPDATE referencing an unknown guild ID: %s. Discarding.', guild_id)
return
settings = guild.notification_settings
if settings is not None:
old_settings = copy.copy(settings)
settings._update(data)
else:
old_settings = None
settings = GuildSettings(data=data, state=self)
self.dispatch('guild_settings_update', old_settings, settings)
def parse_user_required_action_update(self, data) -> None:
required_action = try_enum(RequiredActionType, data['required_action'])
self.dispatch('required_action_update', required_action)
def parse_sessions_replace(self, data: List[Dict[str, Any]]) -> None:
overall = MISSING
this = MISSING
client_status = {}
client_activities = {}
if len(data) == 1:
overall = this = data[0]
def parse_key(key):
index = 0
while True:
if key not in client_status:
return key
if not index:
key += f'-{str(index + 1)}'
else:
key = key.replace(str(index), str(index + 1))
index += 1
for session in data:
if session['session_id'] == 'all':
overall = session
data.remove(session)
continue
elif session['session_id'] == self.session_id:
this = session
continue
key = parse_key(intern(session['client_info']['client']))
client_status[key] = intern(session['status'])
client_activities[key] = tuple(session['activities'])
if overall is MISSING and this is MISSING:
_log.debug('SESSIONS_REPLACE has weird data: %s.', data)
return # ._.
elif overall is MISSING:
overall = this
elif this is MISSING:
this = overall
client_status[None] = overall['status']
client_activities[None] = tuple(overall['activities'])
client_activities['this'] = tuple(this['activities'])
client_status['this'] = this['status']
client = self.client
client._client_status = client_status
client._client_activities = client_activities
client._session_count = len(data)
def parse_invite_create(self, data) -> None:
invite = Invite.from_gateway(state=self, data=data)
self.dispatch('invite_create', invite)

193
discord/team.py
View File

@ -29,15 +29,19 @@ from .user import BaseUser
from .asset import Asset
from .enums import TeamMembershipState, try_enum
from typing import TYPE_CHECKING, Optional, List
from typing import TYPE_CHECKING, Optional, overload, List, Union
if TYPE_CHECKING:
from .abc import Snowflake
from .state import ConnectionState
from .types.team import (
Team as TeamPayload,
TeamMember as TeamMemberPayload,
)
from .types.user import User as UserPayload
MISSING = utils.MISSING
__all__ = (
'Team',
@ -46,7 +50,7 @@ __all__ = (
class Team:
"""Represents an application team for a bot provided by Discord.
"""Represents an application team.
Attributes
-------------
@ -57,21 +61,35 @@ class Team:
owner_id: :class:`int`
The team's owner ID.
members: List[:class:`TeamMember`]
A list of the members in the team
.. versionadded:: 1.3
A list of the members in the team.
A call to :meth:`fetch_members` may be required to populate this past the owner.
"""
if TYPE_CHECKING:
owner_id: int
members: List[TeamMember]
__slots__ = ('_state', 'id', 'name', '_icon', 'owner_id', 'members')
def __init__(self, state: ConnectionState, data: TeamPayload):
self._state: ConnectionState = state
self._update(data)
def _update(self, data: TeamPayload):
self.id: int = int(data['id'])
self.name: str = data['name']
self._icon: Optional[str] = data['icon']
self.owner_id: Optional[int] = utils._get_as_snowflake(data, 'owner_user_id')
self.members: List[TeamMember] = [TeamMember(self, self._state, member) for member in data['members']]
self.owner_id = owner_id = int(data['owner_user_id'])
self.members = members = [TeamMember(self, self._state, member) for member in data.get('members', [])]
if owner_id not in members and owner_id == self._state.self_id: # Discord moment
user: UserPayload = self._state.user._to_minimal_user_json() # type: ignore
member: TeamMemberPayload = {
'user': user,
'team_id': self.id,
'membership_state': 2,
'permissions': ['*'],
}
members.append(TeamMember(self, self._state, member))
def __repr__(self) -> str:
return f'<{self.__class__.__name__} id={self.id} name={self.name}>'
@ -88,6 +106,141 @@ class Team:
"""Optional[:class:`TeamMember`]: The team's owner."""
return utils.get(self.members, id=self.owner_id)
async def edit(
self,
*,
name: str = MISSING,
icon: Optional[bytes] = MISSING,
owner: Snowflake = MISSING,
) -> None:
"""|coro|
Edits the team.
Parameters
-----------
name: :class:`str`
The name of the team.
icon: Optional[:class:`bytes`]
The icon of the team.
owner: :class:`Snowflake`
The team's owner.
Raises
-------
Forbidden
You do not have permissions to edit the team.
HTTPException
Editing the team failed.
"""
payload = {}
if name is not MISSING:
payload['name'] = name
if icon is not MISSING:
if icon is not None:
payload['icon'] = utils._bytes_to_base64_data(icon)
else:
payload['icon'] = ''
if owner is not MISSING:
payload['owner_user_id'] = owner.id
await self._state.http.edit_team(self.id, payload)
await self._state.http.edit_team(self.id, payload)
self._update(payload)
async def fetch_members(self) -> List[TeamMember]:
"""|coro|
Retrieves the team's members.
Returns
--------
List[:class:`TeamMember`]
The team's members.
Raises
-------
Forbidden
You do not have permissions to fetch the team's members.
HTTPException
Retrieving the team members failed.
"""
data = await self._state.http.get_team_members(self.id)
members = [TeamMember(self, self._state, member) for member in data]
self.members = members
return members
@overload
async def invite_member(self, user: BaseUser) -> TeamMember:
...
@overload
async def invite_member(self, user: str) -> TeamMember:
...
@overload
async def invite_member(self, username: str, discriminator: Union[int, str]) -> TeamMember:
...
async def invite_member(self, *args: Union[BaseUser, int, str]) -> TeamMember:
"""|coro|
Invites a member to the team.
This function can be used in multiple ways.
.. code-block:: python
# Passing a user object:
await team.invite_member(user)
# Passing a stringified user:
await team.invite_member('Jake#0001')
# Passing a username and discriminator:
await team.invite_member('Jake', '0001')
Parameters
-----------
user: Union[:class:`User`, :class:`str`]
The user to invite.
username: :class:`str`
The username of the user to invite.
discriminator: :class:`str`
The discriminator of the user to invite.
More than 2 parameters or less than 1 parameter raises a :exc:`TypeError`.
Raises
-------
Forbidden
You do not have permissions to invite the user.
:exc:`.HTTPException`
Inviting the user failed.
Returns
-------
:class:`.TeamMember`
The new member.
"""
username: str
discrim: Union[str, int]
if len(args) == 1:
user = args[0]
if isinstance(user, BaseUser):
user = str(user)
username, discrim = user.split('#') # type: ignore
elif len(args) == 2:
username, discrim = args # type: ignore
else:
raise TypeError(f'invite_member() takes 1 or 2 arguments but {len(args)} were given')
state = self._state
data = await state.http.invite_team_member(self.id, username, discrim)
member = TeamMember(self, state, data)
self.members.append(member)
return member
class TeamMember(BaseUser):
"""Represents a team member in a team.
@ -114,20 +267,10 @@ class TeamMember(BaseUser):
Attributes
-------------
name: :class:`str`
The team member's username.
id: :class:`int`
The team member's unique ID.
discriminator: :class:`str`
The team member's discriminator. This is given when the username has conflicts.
avatar: Optional[:class:`str`]
The avatar hash the team member has. Could be None.
bot: :class:`bool`
Specifies if the user is a bot account.
team: :class:`Team`
The team that the member is from.
membership_state: :class:`TeamMembershipState`
The membership state of the member (e.g. invited or accepted)
The membership state of the member (i.e. invited or accepted)
"""
__slots__ = ('team', 'membership_state', 'permissions')
@ -143,3 +286,17 @@ class TeamMember(BaseUser):
f'<{self.__class__.__name__} id={self.id} name={self.name!r} '
f'discriminator={self.discriminator!r} membership_state={self.membership_state!r}>'
)
async def remove(self) -> None:
"""|coro|
Removes the member from the team.
Raises
-------
Forbidden
You do not have permissions to remove the member.
HTTPException
Removing the member failed.
"""
await self._state.http.remove_team_member(self.team.id, self.id)

9
discord/types/appinfo.py
View File

@ -51,8 +51,13 @@ class _AppInfoOptional(TypedDict, total=False):
class AppInfo(BaseAppInfo, _AppInfoOptional):
rpc_origins: List[str]
owner: User
bot_public: bool
bot_require_code_grant: bool
integration_public: bool
integration_require_code_grant: bool
secret: str
verification_state: int
store_application_state: int
rpc_application_state: int
interactions_endpoint_url: str
class _PartialAppInfoOptional(TypedDict, total=False):
rpc_origins: List[str]

6
discord/types/team.py
View File

@ -26,11 +26,11 @@ from __future__ import annotations
from typing import TypedDict, List, Optional
from .user import PartialUser
from .user import User
from .snowflake import Snowflake
class TeamMember(TypedDict):
user: PartialUser
user: User
membership_state: int
permissions: List[str]
team_id: Snowflake
@ -38,6 +38,6 @@ class TeamMember(TypedDict):
class Team(TypedDict):
id: Snowflake
name: str
owner_id: Snowflake
owner_user_id: Snowflake
members: List[TeamMember]
icon: Optional[str]

2
requirements.txt
View File

@ -1 +1 @@
aiohttp>=3.6.0,<3.8.0
aiohttp>=3.6.0,<3.9.0

Write Preview
Loading…
Cancel
Save