diff options
Diffstat (limited to 'discord')
| -rw-r--r-- | discord/http.py | 5 | ||||
| -rw-r--r-- | discord/webhook.py | 1216 | ||||
| -rw-r--r-- | discord/webhook/__init__.py | 13 | ||||
| -rw-r--r-- | discord/webhook/async_.py | 1388 | ||||
| -rw-r--r-- | discord/webhook/sync.py | 961 |
5 files changed, 2365 insertions, 1218 deletions
diff --git a/discord/http.py b/discord/http.py index b32cc917..6a045f2f 100644 --- a/discord/http.py +++ b/discord/http.py @@ -65,12 +65,13 @@ class Route: # major parameters: self.channel_id = parameters.get('channel_id') self.guild_id = parameters.get('guild_id') - self.interaction_token = parameters.get('interaction_token') + self.webhook_id = parameters.get('webhook_id') + self.webhook_token = parameters.get('webhook_token') @property def bucket(self): # the bucket is just method + path w/ major parameters - return f'{self.channel_id}:{self.guild_id}:{self.interaction_token}:{self.path}' + return f'{self.channel_id}:{self.guild_id}:{self.path}' class MaybeUnlock: diff --git a/discord/webhook.py b/discord/webhook.py deleted file mode 100644 index fd4cd54f..00000000 --- a/discord/webhook.py +++ /dev/null @@ -1,1216 +0,0 @@ -""" -The MIT License (MIT) - -Copyright (c) 2015-present Rapptz - -Permission is hereby granted, free of charge, to any person obtaining a -copy of this software and associated documentation files (the "Software"), -to deal in the Software without restriction, including without limitation -the rights to use, copy, modify, merge, publish, distribute, sublicense, -and/or sell copies of the Software, and to permit persons to whom the -Software is furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in -all copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -DEALINGS IN THE SOFTWARE. -""" - -import logging -import asyncio -import json -import time -import re -from urllib.parse import quote as _uriquote - -import aiohttp - -from . import utils -from .errors import InvalidArgument, HTTPException, Forbidden, NotFound, DiscordServerError -from .message import Message -from .enums import try_enum, WebhookType -from .user import BaseUser, User -from .asset import Asset -from .mixins import Hashable - -__all__ = ( - 'WebhookAdapter', - 'AsyncWebhookAdapter', - 'RequestsWebhookAdapter', - 'Webhook', - 'WebhookMessage', - 'PartialWebhookChannel', - 'PartialWebhookGuild' -) - -log = logging.getLogger(__name__) - -class PartialWebhookChannel(Hashable): - """Represents a partial channel for webhooks. - - These are typically given for channel follower webhooks. - - .. versionadded:: 2.0 - - Attributes - ----------- - id: :class:`int` - The partial channel's ID. - name: :class:`str` - The partial channel's name. - """ - - __slots__ = ('id', 'name') - - def __init__(self, *, data): - self.id = int(data['id']) - self.name = data['name'] - - def __repr__(self): - return f'<PartialWebhookChannel name={self.name!r} id={self.id}>' - -class PartialWebhookGuild(Hashable): - """Represents a partial guild for webhooks. - - These are typically given for channel follower webhooks. - - .. versionadded:: 2.0 - - Attributes - ----------- - id: :class:`int` - The partial guild's ID. - name: :class:`str` - The partial guild's name. - icon: :class:`str` - The partial guild's icon - """ - - __slots__ = ('id', 'name', 'icon', '_state') - - def __init__(self, *, data, state): - self._state = state - self.id = int(data['id']) - self.name = data['name'] - self.icon = data['icon'] - - def __repr__(self): - return f'<PartialWebhookGuild name={self.name!r} id={self.id}>' - - @property - def icon_url(self): - """:class:`Asset`: Returns the guild's icon asset.""" - return self.icon_url_as() - - def is_icon_animated(self): - """:class:`bool`: Returns True if the guild has an animated icon.""" - return bool(self.icon and self.icon.startswith('a_')) - - def icon_url_as(self, *, format=None, static_format='webp', size=1024): - """Returns an :class:`Asset` for the guild's icon. - - The format must be one of 'webp', 'jpeg', 'jpg', 'png' or 'gif', and - 'gif' is only valid for animated avatars. The size must be a power of 2 - between 16 and 4096. - - Parameters - ----------- - format: Optional[:class:`str`] - The format to attempt to convert the icon to. - If the format is ``None``, then it is automatically - detected into either 'gif' or static_format depending on the - icon being animated or not. - static_format: Optional[:class:`str`] - Format to attempt to convert only non-animated icons to. - size: :class:`int` - The size of the image to display. - - Raises - ------ - InvalidArgument - Bad image format passed to ``format`` or invalid ``size``. - - Returns - -------- - :class:`Asset` - The resulting CDN asset. - """ - return Asset._from_guild_icon(self._state, self, format=format, static_format=static_format, size=size) - -class WebhookAdapter: - """Base class for all webhook adapters. - - Attributes - ------------ - webhook: :class:`Webhook` - The webhook that owns this adapter. - """ - - BASE = 'https://discord.com/api/v8' - - def _prepare(self, webhook): - self._webhook_id = webhook.id - self._webhook_token = webhook.token - self._request_url = f'{self.BASE}/webhooks/{webhook.id}/{webhook.token}' - self.webhook = webhook - - def is_async(self): - return False - - def request(self, verb, url, payload=None, multipart=None): - """Actually does the request. - - Subclasses must implement this. - - Parameters - ----------- - verb: :class:`str` - The HTTP verb to use for the request. - url: :class:`str` - The URL to send the request to. This will have - the query parameters already added to it, if any. - multipart: Optional[:class:`dict`] - A dict containing multipart form data to send with - the request. If a filename is being uploaded, then it will - be under a ``file`` key which will have a 3-element :class:`tuple` - denoting ``(filename, file, content_type)``. - payload: Optional[:class:`dict`] - The JSON to send with the request, if any. - """ - raise NotImplementedError() - - def delete_webhook(self, *, reason=None): - return self.request('DELETE', self._request_url, reason=reason) - - def edit_webhook(self, *, reason=None, **payload): - return self.request('PATCH', self._request_url, payload=payload, reason=reason) - - def edit_webhook_message(self, message_id, payload): - return self.request('PATCH', f'{self._request_url}/messages/{message_id}', payload=payload) - - def delete_webhook_message(self, message_id): - return self.request('DELETE', f'{self._request_url}/messages/{message_id}') - - def handle_execution_response(self, data, *, wait): - """Transforms the webhook execution response into something - more meaningful. - - This is mainly used to convert the data into a :class:`Message` - if necessary. - - Subclasses must implement this. - - Parameters - ------------ - data - The data that was returned from the request. - wait: :class:`bool` - Whether the webhook execution was asked to wait or not. - """ - raise NotImplementedError() - - async def _wrap_coroutine_and_cleanup(self, coro, cleanup): - try: - return await coro - finally: - cleanup() - - def execute_webhook(self, *, payload, wait=False, file=None, files=None): - cleanup = None - if file is not None: - multipart = { - 'file': (file.filename, file.fp, 'application/octet-stream'), - 'payload_json': utils.to_json(payload) - } - data = None - cleanup = file.close - files_to_pass = [file] - elif files is not None: - multipart = { - 'payload_json': utils.to_json(payload) - } - for i, file in enumerate(files): - multipart['file%i' % i] = (file.filename, file.fp, 'application/octet-stream') - data = None - - def _anon(): - for f in files: - f.close() - - cleanup = _anon - files_to_pass = files - else: - data = payload - multipart = None - files_to_pass = None - - url = f'{self._request_url}?wait={int(wait)}' - maybe_coro = None - try: - maybe_coro = self.request('POST', url, multipart=multipart, payload=data, files=files_to_pass) - finally: - if maybe_coro is not None and cleanup is not None: - if not asyncio.iscoroutine(maybe_coro): - cleanup() - else: - maybe_coro = self._wrap_coroutine_and_cleanup(maybe_coro, cleanup) - - # if request raises up there then this should never be `None` - return self.handle_execution_response(maybe_coro, wait=wait) - -class AsyncWebhookAdapter(WebhookAdapter): - """A webhook adapter suited for use with aiohttp. - - .. note:: - - You are responsible for cleaning up the client session. - - Parameters - ----------- - session: :class:`aiohttp.ClientSession` - The session to use to send requests. - """ - - def __init__(self, session): - self.session = session - self.loop = asyncio.get_event_loop() - - def is_async(self): - return True - - async def request(self, verb, url, payload=None, multipart=None, *, files=None, reason=None): - headers = {} - data = None - files = files or [] - if payload: - headers['Content-Type'] = 'application/json' - data = utils.to_json(payload) - - if reason: - headers['X-Audit-Log-Reason'] = _uriquote(reason, safe='/ ') - - - base_url = url.replace(self._request_url, '/') or '/' - _id = self._webhook_id - for tries in range(5): - for file in files: - file.reset(seek=tries) - - if multipart: - data = aiohttp.FormData() - for key, value in multipart.items(): - if key.startswith('file'): - data.add_field(key, value[1], filename=value[0], content_type=value[2]) - else: - data.add_field(key, value) - - async with self.session.request(verb, url, headers=headers, data=data) as r: - log.debug('Webhook ID %s with %s %s has returned status code %s', _id, verb, base_url, r.status) - # Coerce empty strings to return None for hygiene purposes - response = (await r.text(encoding='utf-8')) or None - if r.headers['Content-Type'] == 'application/json': - response = json.loads(response) - - # check if we have rate limit header information - remaining = r.headers.get('X-Ratelimit-Remaining') - if remaining == '0' and r.status != 429: - delta = utils._parse_ratelimit_header(r) - log.debug('Webhook ID %s has been pre-emptively rate limited, waiting %.2f seconds', _id, delta) - await asyncio.sleep(delta) - - if 300 > r.status >= 200: - return response - - # we are being rate limited - if r.status == 429: - if not r.headers.get('Via'): - # Banned by Cloudflare more than likely. - raise HTTPException(r, data) - - retry_after = response['retry_after'] / 1000.0 - log.warning('Webhook ID %s is rate limited. Retrying in %.2f seconds', _id, retry_after) - await asyncio.sleep(retry_after) - continue - - if r.status in (500, 502): - await asyncio.sleep(1 + tries * 2) - continue - - if r.status == 403: - raise Forbidden(r, response) - elif r.status == 404: - raise NotFound(r, response) - else: - raise HTTPException(r, response) - - # no more retries - if r.status >= 500: - raise DiscordServerError(r, response) - raise HTTPException(r, response) - - async def handle_execution_response(self, response, *, wait): - data = await response - if not wait: - return data - - # transform into Message object - # Make sure to coerce the state to the partial one to allow message edits/delete - state = _PartialWebhookState(self, self.webhook, parent=self.webhook._state) - return WebhookMessage(data=data, state=state, channel=self.webhook.channel) - -class RequestsWebhookAdapter(WebhookAdapter): - """A webhook adapter suited for use with ``requests``. - - Only versions of :doc:`req:index` higher than 2.13.0 are supported. - - Parameters - ----------- - session: Optional[`requests.Session <http://docs.python-requests.org/en/latest/api/#requests.Session>`_] - The requests session to use for sending requests. If not given then - each request will create a new session. Note if a session is given, - the webhook adapter **will not** clean it up for you. You must close - the session yourself. - sleep: :class:`bool` - Whether to sleep the thread when encountering a 429 or pre-emptive - rate limit or a 5xx status code. Defaults to ``True``. If set to - ``False`` then this will raise an :exc:`HTTPException` instead. - """ - - def __init__(self, session=None, *, sleep=True): - import requests - self.session = session or requests - self.sleep = sleep - - def request(self, verb, url, payload=None, multipart=None, *, files=None, reason=None): - headers = {} - data = None - files = files or [] - if payload: - headers['Content-Type'] = 'application/json' - data = utils.to_json(payload) - - if reason: - headers['X-Audit-Log-Reason'] = _uriquote(reason, safe='/ ') - - if multipart is not None: - data = {'payload_json': multipart.pop('payload_json')} - - base_url = url.replace(self._request_url, '/') or '/' - _id = self._webhook_id - for tries in range(5): - for file in files: - file.reset(seek=tries) - - r = self.session.request(verb, url, headers=headers, data=data, files=multipart) - r.encoding = 'utf-8' - # Coerce empty responses to return None for hygiene purposes - response = r.text or None - - # compatibility with aiohttp - r.status = r.status_code - - log.debug('Webhook ID %s with %s %s has returned status code %s', _id, verb, base_url, r.status) - if r.headers['Content-Type'] == 'application/json': - response = json.loads(response) - - # check if we have rate limit header information - remaining = r.headers.get('X-Ratelimit-Remaining') - if remaining == '0' and r.status != 429 and self.sleep: - delta = utils._parse_ratelimit_header(r) - log.debug('Webhook ID %s has been pre-emptively rate limited, waiting %.2f seconds', _id, delta) - time.sleep(delta) - - if 300 > r.status >= 200: - return response - - # we are being rate limited - if r.status == 429: - if self.sleep: - if not r.headers.get('Via'): - # Banned by Cloudflare more than likely. - raise HTTPException(r, data) - - retry_after = response['retry_after'] / 1000.0 - log.warning('Webhook ID %s is rate limited. Retrying in %.2f seconds', _id, retry_after) - time.sleep(retry_after) - continue - else: - raise HTTPException(r, response) - - if self.sleep and r.status in (500, 502): - time.sleep(1 + tries * 2) - continue - - if r.status == 403: - raise Forbidden(r, response) - elif r.status == 404: - raise NotFound(r, response) - else: - raise HTTPException(r, response) - - # no more retries - if r.status >= 500: - raise DiscordServerError(r, response) - raise HTTPException(r, response) - - def handle_execution_response(self, response, *, wait): - if not wait: - return response - - # transform into Message object - # Make sure to coerce the state to the partial one to allow message edits/delete - state = _PartialWebhookState(self, self.webhook, parent=self.webhook._state) - return WebhookMessage(data=response, state=state, channel=self.webhook.channel) - -class _FriendlyHttpAttributeErrorHelper: - __slots__ = () - - def __getattr__(self, attr): - raise AttributeError('PartialWebhookState does not support http methods.') - -class _PartialWebhookState: - __slots__ = ('loop', 'parent', '_webhook') - - def __init__(self, adapter, webhook, parent): - self._webhook = webhook - - if isinstance(parent, self.__class__): - self.parent = None - else: - self.parent = parent - - # Fetch the loop from the adapter if it's there - try: - self.loop = adapter.loop - except AttributeError: - self.loop = None - - def _get_guild(self, guild_id): - return None - - def store_user(self, data): - return BaseUser(state=self, data=data) - - @property - def http(self): - if self.parent is not None: - return self.parent.http - - # Some data classes assign state.http and that should be kosher - # however, using it should result in a late-binding error. - return _FriendlyHttpAttributeErrorHelper() - - def __getattr__(self, attr): - if self.parent is not None: - return getattr(self.parent, attr) - - raise AttributeError(f'PartialWebhookState does not support {attr!r}.') - -class WebhookMessage(Message): - """Represents a message sent from your webhook. - - This allows you to edit or delete a message sent by your - webhook. - - This inherits from :class:`discord.Message` with changes to - :meth:`edit` and :meth:`delete` to work. - - .. versionadded:: 1.6 - """ - - def edit(self, **fields): - """|maybecoro| - - Edits the message. - - The content must be able to be transformed into a string via ``str(content)``. - - .. versionadded:: 1.6 - - Parameters - ------------ - content: Optional[:class:`str`] - The content to edit the message with or ``None`` to clear it. - embeds: List[:class:`Embed`] - A list of embeds to edit the message with. - embed: Optional[:class:`Embed`] - The embed to edit the message with. ``None`` suppresses the embeds. - This should not be mixed with the ``embeds`` parameter. - allowed_mentions: :class:`AllowedMentions` - Controls the mentions being processed in this message. - See :meth:`.abc.Messageable.send` for more information. - - Raises - ------- - HTTPException - Editing the message failed. - Forbidden - Edited a message that is not yours. - InvalidArgument - You specified both ``embed`` and ``embeds`` or the length of - ``embeds`` was invalid or there was no token associated with - this webhook. - """ - return self._state._webhook.edit_message(self.id, **fields) - - def _delete_delay_sync(self, delay): - time.sleep(delay) - return self._state._webhook.delete_message(self.id) - - async def _delete_delay_async(self, delay): - async def inner_call(): - await asyncio.sleep(delay) - try: - await self._state._webhook.delete_message(self.id) - except HTTPException: - pass - - asyncio.create_task(inner_call()) - return await asyncio.sleep(0) - - def delete(self, *, delay=None): - """|coro| - - Deletes the message. - - Parameters - ----------- - delay: Optional[:class:`float`] - If provided, the number of seconds to wait before deleting the message. - If this is a coroutine, the waiting is done in the background and deletion failures - are ignored. If this is not a coroutine then the delay blocks the thread. - - Raises - ------ - Forbidden - You do not have proper permissions to delete the message. - NotFound - The message was deleted already. - HTTPException - Deleting the message failed. - """ - - if delay is not None: - if self._state.parent._adapter.is_async(): - return self._delete_delay_async(delay) - else: - return self._delete_delay_sync(delay) - - return self._state._webhook.delete_message(self.id) - -class Webhook(Hashable): - """Represents a Discord webhook. - - Webhooks are a form to send messages to channels in Discord without a - bot user or authentication. - - There are two main ways to use Webhooks. The first is through the ones - received by the library such as :meth:`.Guild.webhooks` and - :meth:`.TextChannel.webhooks`. The ones received by the library will - automatically have an adapter bound using the library's HTTP session. - Those webhooks will have :meth:`~.Webhook.send`, :meth:`~.Webhook.delete` and - :meth:`~.Webhook.edit` as coroutines. - - The second form involves creating a webhook object manually without having - it bound to a websocket connection using the :meth:`~.Webhook.from_url` or - :meth:`~.Webhook.partial` classmethods. This form allows finer grained control - over how requests are done, allowing you to mix async and sync code using either - :doc:`aiohttp <aio:index>` or :doc:`req:index`. - - For example, creating a webhook from a URL and using :doc:`aiohttp <aio:index>`: - - .. code-block:: python3 - - from discord import Webhook, AsyncWebhookAdapter - import aiohttp - - async def foo(): - async with aiohttp.ClientSession() as session: - webhook = Webhook.from_url('url-here', adapter=AsyncWebhookAdapter(session)) - await webhook.send('Hello World', username='Foo') - - Or creating a webhook from an ID and token and using :doc:`req:index`: - - .. code-block:: python3 - - import requests - from discord import Webhook, RequestsWebhookAdapter - - webhook = Webhook.partial(123456, 'abcdefg', adapter=RequestsWebhookAdapter()) - webhook.send('Hello World', username='Foo') - - .. container:: operations - - .. describe:: x == y - - Checks if two webhooks are equal. - - .. describe:: x != y - - Checks if two webhooks are not equal. - - .. describe:: hash(x) - - Returns the webhooks's hash. - - .. versionchanged:: 1.4 - Webhooks are now comparable and hashable. - - Attributes - ------------ - id: :class:`int` - The webhook's ID - type: :class:`WebhookType` - The type of the webhook. - - .. versionadded:: 1.3 - - token: Optional[:class:`str`] - The authentication token of the webhook. If this is ``None`` - then the webhook cannot be used to make requests. - guild_id: Optional[:class:`int`] - The guild ID this webhook is for. - channel_id: Optional[:class:`int`] - The channel ID this webhook is for. - user: Optional[:class:`abc.User`] - The user this webhook was created by. If the webhook was - received without authentication then this will be ``None``. - name: Optional[:class:`str`] - The default name of the webhook. - avatar: Optional[:class:`str`] - The default avatar of the webhook. - source_guild: Optional[:class:`PartialWebhookGuild`] - The guild of the channel that this webhook is following. - Only given if :attr:`type` is :attr:`WebhookType.channel_follower`. - - .. versionadded:: 2.0 - - source_channel: Optional[:class:`PartialWebhookChannel`] - The channel that this webhook is following. - Only given if :attr:`type` is :attr:`WebhookType.channel_follower`. - - .. versionadded:: 2.0 - """ - - __slots__ = ('id', 'type', 'guild_id', 'channel_id', 'user', 'name', - 'avatar', 'token', '_state', '_adapter', 'source_channel', 'source_guild') - - def __init__(self, data, *, adapter, state=None): - self.id = int(data['id']) - self.type = try_enum(WebhookType, int(data['type'])) - self.channel_id = utils._get_as_snowflake(data, 'channel_id') - self.guild_id = utils._get_as_snowflake(data, 'guild_id') - self.name = data.get('name') - self.avatar = data.get('avatar') - self.token = data.get('token') - self._state = state or _PartialWebhookState(adapter, self, parent=state) - self._adapter = adapter - self._adapter._prepare(self) - - user = data.get('user') - if user is None: - self.user = None - elif state is None: - self.user = BaseUser(state=None, data=user) - else: - self.user = User(state=state, data=user) - - source_channel = data.get('source_channel') - if source_channel: - source_channel = PartialWebhookChannel(data=source_channel) - - self.source_channel = source_channel - - source_guild = data.get('source_guild') - if source_guild: - source_guild = PartialWebhookGuild(data=source_guild, state=state) - - self.source_guild = source_guild - - def __repr__(self): - return f'<Webhook id={self.id!r}>' - - @property - def url(self): - """:class:`str` : Returns the webhook's url.""" - return f'https://discord.com/api/webhooks/{self.id}/{self.token}' - - @classmethod - def partial(cls, id, token, *, adapter): - """Creates a partial :class:`Webhook`. - - Parameters - ----------- - id: :class:`int` - The ID of the webhook. - token: :class:`str` - The authentication token of the webhook. - adapter: :class:`WebhookAdapter` - The webhook adapter to use when sending requests. This is - typically :class:`AsyncWebhookAdapter` for :doc:`aiohttp <aio:index>` or - :class:`RequestsWebhookAdapter` for :doc:`req:index`. - - Returns - -------- - :class:`Webhook` - A partial :class:`Webhook`. - A partial webhook is just a webhook object with an ID and a token. - """ - - if not isinstance(adapter, WebhookAdapter): - raise TypeError('adapter must be a subclass of WebhookAdapter') - - data = { - 'id': id, - 'type': 1, - 'token': token - } - - return cls(data, adapter=adapter) - - @classmethod - def from_url(cls, url, *, adapter): - """Creates a partial :class:`Webhook` from a webhook URL. - - Parameters - ------------ - url: :class:`str` - The URL of the webhook. - adapter: :class:`WebhookAdapter` - The webhook adapter to use when sending requests. This is - typically :class:`AsyncWebhookAdapter` for :doc:`aiohttp <aio:index>` or - :class:`RequestsWebhookAdapter` for :doc:`req:index`. - - Raises - ------- - InvalidArgument - The URL is invalid. - - Returns - -------- - :class:`Webhook` - A partial :class:`Webhook`. - A partial webhook is just a webhook object with an ID and a token. - """ - - m = re.search(r'discord(?:app)?.com/api/webhooks/(?P<id>[0-9]{17,20})/(?P<token>[A-Za-z0-9\.\-\_]{60,68})', url) - if m is None: - raise InvalidArgument('Invalid webhook URL given.') - data = m.groupdict() - data['type'] = 1 - return cls(data, adapter=adapter) - - @classmethod - def _as_follower(cls, data, *, channel, user): - name = f"{channel.guild} #{channel}" - feed = { - 'id': data['webhook_id'], - 'type': 2, - 'name': name, - 'channel_id': channel.id, - 'guild_id': channel.guild.id, - 'user': { - 'username': user.name, - 'discriminator': user.discriminator, - 'id': user.id, - 'avatar': user.avatar - } - } - - session = channel._state.http._HTTPClient__session - return cls(feed, adapter=AsyncWebhookAdapter(session=session)) - - @classmethod - def from_state(cls, data, state): - session = state.http._HTTPClient__session - return cls(data, adapter=AsyncWebhookAdapter(session=session), state=state) - - @property - def guild(self): - """Optional[:class:`Guild`]: The guild this webhook belongs to. - - If this is a partial webhook, then this will always return ``None``. - """ - return self._state._get_guild(self.guild_id) - - @property - def channel(self): - """Optional[:class:`TextChannel`]: The text channel this webhook belongs to. - - If this is a partial webhook, then this will always return ``None``. - """ - guild = self.guild - return guild and guild.get_channel(self.channel_id) - - @property - def created_at(self): - """:class:`datetime.datetime`: Returns the webhook's creation time in UTC.""" - return utils.snowflake_time(self.id) - - @property - def avatar_url(self): - """:class:`Asset`: Returns an :class:`Asset` for the avatar the webhook has. - - If the webhook does not have a traditional avatar, an asset for - the default avatar is returned instead. - - This is equivalent to calling :meth:`avatar_url_as` with the - default parameters. - """ - return self.avatar_url_as() - - def avatar_url_as(self, *, format=None, size=1024): - """Returns an :class:`Asset` for the avatar the webhook has. - - If the webhook does not have a traditional avatar, an asset for - the default avatar is returned instead. - - The format must be one of 'jpeg', 'jpg', or 'png'. - The size must be a power of 2 between 16 and 1024. - - Parameters - ----------- - format: Optional[:class:`str`] - The format to attempt to convert the avatar to. - If the format is ``None``, then it is equivalent to png. - size: :class:`int` - The size of the image to display. - - Raises - ------ - InvalidArgument - Bad image format passed to ``format`` or invalid ``size``. - - Returns - -------- - :class:`Asset` - The resulting CDN asset. - """ - if self.avatar is None: - # Default is always blurple apparently - return Asset(self._state, '/embed/avatars/0.png') - - if not utils.valid_icon_size(size): - raise InvalidArgument("size must be a power of 2 between 16 and 1024") - - format = format or 'png' - - if format not in ('png', 'jpg', 'jpeg'): - raise InvalidArgument("format must be one of 'png', 'jpg', or 'jpeg'.") - - url = f'/avatars/{self.id}/{self.avatar}.{format}?size={size}' - return Asset(self._state, url) - - def delete(self, *, reason=None): - """|maybecoro| - - Deletes this Webhook. - - If the webhook is constructed with a :class:`RequestsWebhookAdapter` then this is - not a coroutine. - - Parameters - ------------ - reason: Optional[:class:`str`] - The reason for deleting this webhook. Shows up on the audit log. - - .. versionadded:: 1.4 - - Raises - ------- - HTTPException - Deleting the webhook failed. - NotFound - This webhook does not exist. - Forbidden - You do not have permissions to delete this webhook. - InvalidArgument - This webhook does not have a token associated with it. - """ - if self.token is None: - raise InvalidArgument('This webhook does not have a token associated with it') - - return self._adapter.delete_webhook(reason=reason) - - def edit(self, *, reason=None, **kwargs): - """|maybecoro| - - Edits this Webhook. - - If the webhook is constructed with a :class:`RequestsWebhookAdapter` then this is - not a coroutine. - - Parameters - ------------ - name: Optional[:class:`str`] - The webhook's new default name. - avatar: Optional[:class:`bytes`] - A :term:`py:bytes-like object` representing the webhook's new default avatar. - reason: Optional[:class:`str`] - The reason for editing this webhook. Shows up on the audit log. - - .. versionadded:: 1.4 - - Raises - ------- - HTTPException - Editing the webhook failed. - NotFound - This webhook does not exist. - InvalidArgument - This webhook does not have a token associated with it. - """ - if self.token is None: - raise InvalidArgument('This webhook does not have a token associated with it') - - payload = {} - - try: - name = kwargs['name'] - except KeyError: - pass - else: - if name is not None: - payload['name'] = str(name) - else: - payload['name'] = None - - try: - avatar = kwargs['avatar'] - except KeyError: - pass - else: - if avatar is not None: - payload['avatar'] = utils._bytes_to_base64_data(avatar) - else: - payload['avatar'] = None - - return self._adapter.edit_webhook(reason=reason, **payload) - - def send(self, content=None, *, wait=False, username=None, avatar_url=None, tts=False, - file=None, files=None, embed=None, embeds=None, allowed_mentions=None): - """|maybecoro| - - Sends a message using the webhook. - - If the webhook is constructed with a :class:`RequestsWebhookAdapter` then this is - not a coroutine. - - The content must be a type that can convert to a string through ``str(content)``. - - To upload a single file, the ``file`` parameter should be used with a - single :class:`File` object. - - If the ``embed`` parameter is provided, it must be of type :class:`Embed` and - it must be a rich embed type. You cannot mix the ``embed`` parameter with the - ``embeds`` parameter, which must be a :class:`list` of :class:`Embed` objects to send. - - Parameters - ------------ - content: :class:`str` - The content of the message to send. - wait: :class:`bool` - Whether the server should wait before sending a response. This essentially - means that the return type of this function changes from ``None`` to - a :class:`WebhookMessage` if set to ``True``. - username: :class:`str` - The username to send with this message. If no username is provided - then the default username for the webhook is used. - avatar_url: Union[:class:`str`, :class:`Asset`] - The avatar URL to send with this message. If no avatar URL is provided - then the default avatar for the webhook is used. - tts: :class:`bool` - Indicates if the message should be sent using text-to-speech. - file: :class:`File` - The file to upload. This cannot be mixed with ``files`` parameter. - files: List[:class:`File`] - A list of files to send with the content. This cannot be mixed with the - ``file`` parameter. - embed: :class:`Embed` - The rich embed for the content to send. This cannot be mixed with - ``embeds`` parameter. - embeds: List[:class:`Embed`] - A list of embeds to send with the content. Maximum of 10. This cannot - be mixed with the ``embed`` parameter. - allowed_mentions: :class:`AllowedMentions` - Controls the mentions being processed in this message. - - .. versionadded:: 1.4 - - Raises - -------- - HTTPException - Sending the message failed. - NotFound - This webhook was not found. - Forbidden - The authorization token for the webhook is incorrect. - InvalidArgument - You specified both ``embed`` and ``embeds`` or the length of - ``embeds`` was invalid or there was no token associated with - this webhook. - - Returns - --------- - Optional[:class:`WebhookMessage`] - The message that was sent. - """ - - payload = {} - if self.token is None: - raise InvalidArgument('This webhook does not have a token associated with it') - if files is not None and file is not None: - raise InvalidArgument('Cannot mix file and files keyword arguments.') - if embeds is not None and embed is not None: - raise InvalidArgument('Cannot mix embed and embeds keyword arguments.') - - if embeds is not None: - if len(embeds) > 10: - raise InvalidArgument('embeds has a maximum of 10 elements.') - payload['embeds'] = [e.to_dict() for e in embeds] - - if embed is not None: - payload['embeds'] = [embed.to_dict()] - - if content is not None: - payload['content'] = str(content) - - payload['tts'] = tts - if avatar_url: - payload['avatar_url'] = str(avatar_url) - if username: - payload['username'] = username - - previous_mentions = getattr(self._state, 'allowed_mentions', None) - - if allowed_mentions: - if previous_mentions is not None: - payload['allowed_mentions'] = previous_mentions.merge(allowed_mentions).to_dict() - else: - payload['allowed_mentions'] = allowed_mentions.to_dict() - elif previous_mentions is not None: - payload['allowed_mentions'] = previous_mentions.to_dict() - - return self._adapter.execute_webhook(wait=wait, file=file, files=files, payload=payload) - - def execute(self, *args, **kwargs): - """An alias for :meth:`~.Webhook.send`.""" - return self.send(*args, **kwargs) - - def edit_message(self, message_id, **fields): - """|maybecoro| - - Edits a message owned by this webhook. - - This is a lower level interface to :meth:`WebhookMessage.edit` in case - you only have an ID. - - .. versionadded:: 1.6 - - Parameters - ------------ - message_id: :class:`int` - The message ID to edit. - content: Optional[:class:`str`] - The content to edit the message with or ``None`` to clear it. - embeds: List[:class:`Embed`] - A list of embeds to edit the message with. - embed: Optional[:class:`Embed`] - The embed to edit the message with. ``None`` suppresses the embeds. - This should not be mixed with the ``embeds`` parameter. - allowed_mentions: :class:`AllowedMentions` - Controls the mentions being processed in this message. - See :meth:`.abc.Messageable.send` for more information. - - Raises - ------- - HTTPException - Editing the message failed. - Forbidden - Edited a message that is not yours. - InvalidArgument - You specified both ``embed`` and ``embeds`` or the length of - ``embeds`` was invalid or there was no token associated with - this webhook. - """ - - payload = {} - - if self.token is None: - raise InvalidArgument('This webhook does not have a token associated with it') - - try: - content = fields['content'] - except KeyError: - pass - else: - if content is not None: - content = str(content) - payload['content'] = content - - # Check if the embeds interface is being used - try: - embeds = fields['embeds'] - except KeyError: - # Nope - pass - else: - if embeds is None or len(embeds) > 10: - raise InvalidArgument('embeds has a maximum of 10 elements') - payload['embeds'] = [e.to_dict() for e in embeds] - - try: - embed = fields['embed'] - except KeyError: - pass - else: - if 'embeds' in payload: - raise InvalidArgument('Cannot mix embed and embeds keyword arguments') - - if embed is None: - payload['embeds'] = [] - else: - payload['embeds'] = [embed.to_dict()] - - allowed_mentions = fields.pop('allowed_mentions', None) - previous_mentions = getattr(self._state, 'allowed_mentions', None) - - if allowed_mentions: - if previous_mentions is not None: - payload['allowed_mentions'] = previous_mentions.merge(allowed_mentions).to_dict() - else: - payload['allowed_mentions'] = allowed_mentions.to_dict() - elif previous_mentions is not None: - payload['allowed_mentions'] = previous_mentions.to_dict() - - return self._adapter.edit_webhook_message(message_id, payload=payload) - - def delete_message(self, message_id): - """|maybecoro| - - Deletes a message owned by this webhook. - - This is a lower level interface to :meth:`WebhookMessage.delete` in case - you only have an ID. - - .. versionadded:: 1.6 - - Parameters - ------------ - message_id: :class:`int` - The message ID to delete. - - Raises - ------- - HTTPException - Deleting the message failed. - Forbidden - Deleted a message that is not yours. - """ - return self._adapter.delete_webhook_message(message_id) diff --git a/discord/webhook/__init__.py b/discord/webhook/__init__.py new file mode 100644 index 00000000..8b0e4568 --- /dev/null +++ b/discord/webhook/__init__.py @@ -0,0 +1,13 @@ +""" +discord.webhook +~~~~~~~~~~~~~~ + +Webhook support + +:copyright: (c) 2015-present Rapptz +:license: MIT, see LICENSE for more details. + +""" + +from .async_ import * +from .sync import * diff --git a/discord/webhook/async_.py b/discord/webhook/async_.py new file mode 100644 index 00000000..6a68aede --- /dev/null +++ b/discord/webhook/async_.py @@ -0,0 +1,1388 @@ +""" +The MIT License (MIT) + +Copyright (c) 2015-present Rapptz + +Permission is hereby granted, free of charge, to any person obtaining a +copy of this software and associated documentation files (the "Software"), +to deal in the Software without restriction, including without limitation +the rights to use, copy, modify, merge, publish, distribute, sublicense, +and/or sell copies of the Software, and to permit persons to whom the +Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. +""" + +from __future__ import annotations + +import contextvars +import logging +import asyncio +import json +import time +import re + +from urllib.parse import quote as urlquote +from typing import Any, Dict, List, Literal, NamedTuple, Optional, TYPE_CHECKING, Tuple, Union, overload + +import aiohttp + +from .. import utils +from ..errors import InvalidArgument, HTTPException, Forbidden, NotFound, DiscordServerError +from ..message import Message +from ..enums import try_enum, WebhookType +from ..user import BaseUser, User +from ..asset import Asset +from ..http import Route +from ..mixins import Hashable +from ..object import Object + +__all__ = ( + 'Webhook', + 'WebhookMessage', + 'PartialWebhookChannel', + 'PartialWebhookGuild', +) + +log = logging.getLogger(__name__) + +if TYPE_CHECKING: + from ..file import File + from ..embeds import Embed + from ..mentions import AllowedMentions + from ..types.webhook import ( + Webhook as WebhookPayload, + ) + from ..guild import Guild + from ..channel import TextChannel + from ..abc import Snowflake + import datetime + + +class _Missing: + def __bool__(self): + return False + + def __repr__(self): + return '...' + + +MISSING: Any = _Missing() + + +class AsyncDeferredLock: + def __init__(self, lock: asyncio.Lock): + self.lock = lock + self.delta: Optional[float] = None + + async def __aenter__(self): + await self.lock.acquire() + return self + + def delay_by(self, delta: float) -> None: + self.delta = delta + + async def __aexit__(self, type, value, traceback): + if self.delta: + await asyncio.sleep(self.delta) + self.lock.release() + + +class AsyncWebhookAdapter: + def __init__(self): + self._locks: Dict[Any, asyncio.Lock] = {} + + async def request( + self, + route: Route, + session: aiohttp.ClientSession, + *, + payload: Optional[Dict[str, Any]] = None, + multipart: Optional[List[Dict[str, Any]]] = None, + files: Optional[List[File]] = None, + reason: Optional[str] = None, + auth_token: Optional[str] = None, + params: Optional[Dict[str, Any]] = None, + ) -> Any: + headers: Dict[str, str] = {} + files = files or [] + to_send: Optional[Union[str, aiohttp.FormData]] = None + bucket = (route.webhook_id, route.webhook_token) + + try: + lock = self._locks[bucket] + except KeyError: + self._locks[bucket] = lock = asyncio.Lock() + + if payload is not None: + headers['Content-Type'] = 'application/json' + to_send = utils.to_json(payload) + + if auth_token is not None: + headers['Authorization'] = f'Bot {auth_token}' + + if reason is not None: + headers['X-Audit-Log-Reason'] = urlquote(reason, safe='/ ') + + response: Optional[aiohttp.ClientResponse] = None + data: Optional[Union[Dict[str, Any], str]] = None + method = route.method + url = route.url + webhook_id = route.webhook_id + + async with AsyncDeferredLock(lock) as lock: + for attempt in range(5): + for file in files: + file.reset(seek=attempt) + + if multipart: + form_data = aiohttp.FormData() + for p in multipart: + form_data.add_field(**p) + to_send = form_data + + try: + async with session.request(method, url, data=to_send, headers=headers, params=params) as response: + log.debug( + 'Webhook ID %s with %s %s has returned status code %s', + webhook_id, + method, + url, + response.status, + ) + data = (await response.text(encoding='utf-8')) or None + if data and response.headers['Content-Type'] == 'application/json': + data = json.loads(data) + + remaining = response.headers.get('X-Ratelimit-Remaining') + if remaining == '0' and response.status != 429: + delta = utils._parse_ratelimit_header(response.headers) + log.debug( + 'Webhook ID %s has been pre-emptively rate limited, waiting %.2f seconds', webhook_id, delta + ) + lock.delay_by(delta) + + if 300 > response.status >= 200: + return data + + if response.status == 429: + if not response.headers.get('Via'): + raise HTTPException(response, data) + + retry_after = data['retry_after'] / 1000.0 # type: ignore + log.warning('Webhook ID %s is rate limited. Retrying in %.2f seconds', webhook_id, retry_after) + await asyncio.sleep(retry_after) + + if response.status >= 500: + await asyncio.sleep(1 + attempt * 2) + continue + + if response.status == 403: + raise Forbidden(response, data) + elif response.status == 404: + raise NotFound(response, data) + else: + raise HTTPException(response, data) + + except OSError as e: + if attempt < 4 and e.errno in (54, 10054): + continue + raise + + if response: + if response.status >= 500: + raise DiscordServerError(response, data) + raise HTTPException(response, data) + + raise RuntimeError('Unreachable code in HTTP handling.') + + def delete_webhook( + self, + webhook_id: int, + *, + token: Optional[str] = None, + session: aiohttp.ClientSession, + reason: Optional[str] = None, + ): + route = Route('DELETE', '/webhooks/{webhook_id}', webhook_id=webhook_id) + return self.request(route, session, reason=reason, auth_token=token) + + def delete_webhook_with_token( + self, + webhook_id: int, + token: str, + *, + session: aiohttp.ClientSession, + reason: Optional[str] = None, + ): + route = Route('DELETE', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session, reason=reason) + + def edit_webhook( + self, + webhook_id: int, + token: str, + payload: Dict[str, Any], + *, + session: aiohttp.ClientSession, + reason: Optional[str] = None, + ): + route = Route('PATCH', '/webhooks/{webhook_id}', webhook_id=webhook_id) + return self.request(route, session, reason=reason, payload=payload, auth_token=token) + + def edit_webhook_with_token( + self, + webhook_id: int, + token: str, + payload: Dict[str, Any], + *, + session: aiohttp.ClientSession, + reason: Optional[str] = None, + ): + route = Route('PATCH', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session, reason=reason, payload=payload) + + def execute_webhook( + self, + webhook_id: int, + token: str, + *, + session: aiohttp.ClientSession, + payload: Optional[Dict[str, Any]] = None, + multipart: Optional[List[Dict[str, Any]]] = None, + files: Optional[List[File]] = None, + wait: bool = False, + ): + params = {'wait': int(wait)} + route = Route('POST', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session, payload=payload, multipart=multipart, files=files, params=params) + + def edit_webhook_message( + self, + webhook_id: int, + token: str, + message_id: int, + *, + session: aiohttp.ClientSession, + payload: Optional[Dict[str, Any]] = None, + multipart: Optional[List[Dict[str, Any]]] = None, + files: Optional[List[File]] = None, + ): + route = Route( + 'PATCH', + '/webhooks/{webhook_id}/{webhook_token}/messages/{message_id}', + webhook_id=webhook_id, + webhook_token=token, + message_id=message_id, + ) + return self.request(route, session, payload=payload, multipart=multipart, files=files) + + def delete_webhook_message( + self, + webhook_id: int, + token: str, + message_id: int, + *, + session: aiohttp.ClientSession, + ): + route = Route( + 'DELETE', + '/webhooks/{webhook_id}/{webhook_token}/messages/{message_id}', + webhook_id=webhook_id, + webhook_token=token, + message_id=message_id, + ) + return self.request(route, session) + + def fetch_webhook( + self, + webhook_id: int, + token: str, + *, + session: aiohttp.ClientSession, + ): + route = Route('GET', '/webhooks/{webhook_id}', webhook_id=webhook_id) + return self.request(route, session=session, auth_token=token) + + def fetch_webhook_with_token( + self, + webhook_id: int, + token: str, + *, + session: aiohttp.ClientSession, + ): + route = Route('GET', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session=session) + + +class ExecuteWebhookParameters(NamedTuple): + payload: Optional[Dict[str, Any]] + multipart: Optional[List[Dict[str, Any]]] + files: Optional[List[File]] + + +def handle_message_parameters( + content: Optional[str] = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = False, + file: File = MISSING, + files: List[File] = MISSING, + embed: Optional[Embed] = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: Optional[AllowedMentions] = MISSING, + previous_allowed_mentions: Optional[AllowedMentions] = None, +) -> ExecuteWebhookParameters: + if files is not MISSING and file is not MISSING: + raise TypeError('Cannot mix file and files keyword arguments.') + if embeds is not MISSING and embed is not MISSING: + raise TypeError('Cannot mix embed and embeds keyword arguments.') + + payload = {} + if embeds is not MISSING: + if len(embeds) > 10: + raise InvalidArgument('embeds has a maximum of 10 elements.') + payload['embeds'] = [e.to_dict() for e in embeds] + + if embed is not MISSING: + if embed is None: + payload['embeds'] = [] + else: + payload['embeds'] = [embed.to_dict()] + + if content is not MISSING: + if content is not None: + payload['content'] = str(content) + else: + payload['content'] = None + + payload['tts'] = tts + if avatar_url: + payload['avatar_url'] = str(avatar_url) + if username: + payload['username'] = username + + if allowed_mentions: + if previous_allowed_mentions is not None: + payload['allowed_mentions'] = previous_allowed_mentions.merge(allowed_mentions).to_dict() + else: + payload['allowed_mentions'] = allowed_mentions.to_dict() + elif previous_allowed_mentions is not None: + payload['allowed_mentions'] = previous_allowed_mentions.to_dict() + + multipart = [] + if file is not MISSING: + files = [file] + + if files: + multipart.append({'name': 'payload_json', 'value': utils.to_json(payload)}) + payload = None + if len(files) == 1: + file = files[0] + multipart.append( + { + 'name': 'file', + 'value': file.fp, + 'filename': file.filename, + 'content_type': 'application/octet-stream', + } + ) + else: + for index, file in enumerate(files): + multipart.append( + { + 'name': f'file{index}', + 'value': file.fp, + 'filename': file.filename, + 'content_type': 'application/octet-stream', + } + ) + + return ExecuteWebhookParameters(payload=payload, multipart=multipart, files=files) + + +async_context = contextvars.ContextVar('async_webhook_context', default=AsyncWebhookAdapter()) + + +class PartialWebhookChannel(Hashable): + """Represents a partial channel for webhooks. + + These are typically given for channel follower webhooks. + + .. versionadded:: 2.0 + + Attributes + ----------- + id: :class:`int` + The partial channel's ID. + name: :class:`str` + The partial channel's name. + """ + + __slots__ = ('id', 'name') + + def __init__(self, *, data): + self.id = int(data['id']) + self.name = data['name'] + + def __repr__(self): + return f'<PartialWebhookChannel name={self.name!r} id={self.id}>' + + +class PartialWebhookGuild(Hashable): + """Represents a partial guild for webhooks. + + These are typically given for channel follower webhooks. + + .. versionadded:: 2.0 + + Attributes + ----------- + id: :class:`int` + The partial guild's ID. + name: :class:`str` + The partial guild's name. + icon: :class:`str` + The partial guild's icon + """ + + __slots__ = ('id', 'name', 'icon', '_state') + + def __init__(self, *, data, state): + self._state = state + self.id = int(data['id']) + self.name = data['name'] + self.icon = data['icon'] + + def __repr__(self): + return f'<PartialWebhookGuild name={self.name!r} id={self.id}>' + + @property + def icon_url(self) -> Asset: + """:class:`Asset`: Returns the guild's icon asset.""" + return self.icon_url_as() + + def is_icon_animated(self) -> bool: + """:class:`bool`: Returns True if the guild has an animated icon.""" + return bool(self.icon and self.icon.startswith('a_')) + + def icon_url_as(self, *, format=None, static_format='webp', size=1024): + """Returns an :class:`Asset` for the guild's icon. + + The format must be one of 'webp', 'jpeg', 'jpg', 'png' or 'gif', and + 'gif' is only valid for animated avatars. The size must be a power of 2 + between 16 and 4096. + + Parameters + ----------- + format: Optional[:class:`str`] + The format to attempt to convert the icon to. + If the format is ``None``, then it is automatically + detected into either 'gif' or static_format depending on the + icon being animated or not. + static_format: Optional[:class:`str`] + Format to attempt to convert only non-animated icons to. + size: :class:`int` + The size of the image to display. + + Raises + ------ + InvalidArgument + Bad image format passed to ``format`` or invalid ``size``. + + Returns + -------- + :class:`Asset` + The resulting CDN asset. + """ + return Asset._from_guild_icon(self._state, self, format=format, static_format=static_format, size=size) + + +class _FriendlyHttpAttributeErrorHelper: + __slots__ = () + + def __getattr__(self, attr): + raise AttributeError('PartialWebhookState does not support http methods.') + + +class _WebhookState: + __slots__ = ('_parent', '_webhook') + + def __init__(self, webhook, parent): + self._webhook = webhook + + if isinstance(parent, self.__class__): + self._parent = None + else: + self._parent = parent + + def _get_guild(self, guild_id): + if self._parent is not None: + return self._parent._get_guild(guild_id) + return None + + def store_user(self, data): + if self._parent is not None: + return self._parent.store_user(data) + return BaseUser(state=self, data=data) + + @property + def http(self): + if self._parent is not None: + return self._parent.http + + # Some data classes assign state.http and that should be kosher + # however, using it should result in a late-binding error. + return _FriendlyHttpAttributeErrorHelper() + + def __getattr__(self, attr): + if self._parent is not None: + return getattr(self._parent, attr) + + raise AttributeError(f'PartialWebhookState does not support {attr!r}.') + + +class WebhookMessage(Message): + """Represents a message sent from your webhook. + + This allows you to edit or delete a message sent by your + webhook. + + This inherits from :class:`discord.Message` with changes to + :meth:`edit` and :meth:`delete` to work. + + .. versionadded:: 1.6 + """ + + async def edit( + self, + content: Optional[str] = MISSING, + embeds: List[Embed] = MISSING, + embed: Optional[Embed] = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + allowed_mentions: Optional[AllowedMentions] = None, + ): + """|coro| + + Edits a message owned by this webhook. + + This is a lower level interface to :meth:`WebhookMessage.edit` in case + you only have an ID. + + .. versionadded:: 1.6 + + Parameters + ------------ + message_id: :class:`int` + The message ID to edit. + content: Optional[:class:`str`] + The content to edit the message with or ``None`` to clear it. + embeds: List[:class:`Embed`] + A list of embeds to edit the message with. + embed: Optional[:class:`Embed`] + The embed to edit the message with. ``None`` suppresses the embeds. + This should not be mixed with the ``embeds`` parameter. + file: :class:`File` + The file to upload. This cannot be mixed with ``files`` parameter. + files: List[:class:`File`] + A list of files to send with the content. This cannot be mixed with the + ``file`` parameter. + allowed_mentions: :class:`AllowedMentions` + Controls the mentions being processed in this message. + See :meth:`.abc.Messageable.send` for more information. + + Raises + ------- + HTTPException + Editing the message failed. + Forbidden + Edited a message that is not yours. + TypeError + You specified both ``embed`` and ``embeds`` or ``file`` and ``files`` + ValueError + The length of ``embeds`` was invalid + InvalidArgument + There was no token associated with this webhook. + """ + await self._state._webhook.edit_message( + self.id, + content=content, + embeds=embeds, + embed=embed, + file=file, + files=files, + allowed_mentions=allowed_mentions, + ) + + async def delete(self, *, delay: Optional[float] = None) -> None: + """|coro| + + Deletes the message. + + Parameters + ----------- + delay: Optional[:class:`float`] + If provided, the number of seconds to wait before deleting the message. + The waiting is done in the background and deletion failures are ignored. + + Raises + ------ + Forbidden + You do not have proper permissions to delete the message. + NotFound + The message was deleted already. + HTTPException + Deleting the message failed. + """ + + if delay is not None: + + async def inner_call(delay: float = delay): + await asyncio.sleep(delay) + try: + await self._state._webhook.delete_message(self.id) + except HTTPException: + pass + + asyncio.create_task(inner_call()) + else: + await self._state._webhook.delete_message(self.id) + + +class BaseWebhook(Hashable): + __slots__: Tuple[str, ...] = ( + 'id', + 'type', + 'guild_id', + 'channel_id', + 'token', + 'auth_token', + 'user', + 'name', + 'avatar', + 'source_channel', + 'source_guild', + '_state', + ) + + def __init__(self, data: WebhookPayload, token: Optional[str] = None, state=None): + self.auth_token: Optional[str] = token + self._state = state or _WebhookState(self, parent=state) + self._update(data) + + def _update(self, data: WebhookPayload): + self.id = int(data['id']) + self.type = try_enum(WebhookType, int(data['type'])) + self.channel_id = utils._get_as_snowflake(data, 'channel_id') + self.guild_id = utils._get_as_snowflake(data, 'guild_id') + self.name = data.get('name') + self.avatar = data.get('avatar') + self.token = data.get('token') + + user = data.get('user') + self.user: Optional[Union[BaseUser, User]] = None + if user is not None: + if self._state is None: + self.user = BaseUser(state=None, data=user) + else: + self.user = User(state=self._state, data=user) + + source_channel = data.get('source_channel') + if source_channel: + source_channel = PartialWebhookChannel(data=source_channel) + + self.source_channel: Optional[PartialWebhookChannel] = source_channel + + source_guild = data.get('source_guild') + if source_guild: + source_guild = PartialWebhookGuild(data=source_guild, state=self._state) + + self.source_guild: Optional[PartialWebhookGuild] = source_guild + + def is_partial(self) -> bool: + """:class:`bool`: Whether the webhook is a "partial" webhook.""" + return self.channel_id is None + + def is_authenticated(self) -> bool: + """:class:`bool`: Whether the webhook is authenticated with a bot token.""" + return self.auth_token is not None + + @property + def guild(self) -> Optional[Guild]: + """Optional[:class:`Guild`]: The guild this webhook belongs to. + + If this is a partial webhook, then this will always return ``None``. + """ + return self._state and self._state._get_guild(self.guild_id) + + @property + def channel(self) -> Optional[TextChannel]: + """Optional[:class:`TextChannel`]: The text channel this webhook belongs to. + + If this is a partial webhook, then this will always return ``None``. + """ + guild = self.guild + return guild and guild.get_channel(self.channel_id) + + @property + def created_at(self) -> datetime.datetime: + """:class:`datetime.datetime`: Returns the webhook's creation time in UTC.""" + return utils.snowflake_time(self.id) + + @property + def avatar_url(self) -> Asset: + """:class:`Asset`: Returns an :class:`Asset` for the avatar the webhook has. + + If the webhook does not have a traditional avatar, an asset for + the default avatar is returned instead. + + This is equivalent to calling :meth:`avatar_url_as` with the + default parameters. + """ + return self.avatar_url_as() + + def avatar_url_as(self, *, format: Optional[Literal['png', 'jpg', 'jpeg']] = None, size: int = 1024) -> Asset: + """Returns an :class:`Asset` for the avatar the webhook has. + + If the webhook does not have a traditional avatar, an asset for + the default avatar is returned instead. + + The format must be one of 'jpeg', 'jpg', or 'png'. + The size must be a power of 2 between 16 and 1024. + + Parameters + ----------- + format: Optional[:class:`str`] + The format to attempt to convert the avatar to. + If the format is ``None``, then it is equivalent to png. + size: :class:`int` + The size of the image to display. + + Raises + ------ + InvalidArgument + Bad image format passed to ``format`` or invalid ``size``. + + Returns + -------- + :class:`Asset` + The resulting CDN asset. + """ + if self.avatar is None: + # Default is always blurple apparently + return Asset(self._state, '/embed/avatars/0.png') + + if not utils.valid_icon_size(size): + raise InvalidArgument("size must be a power of 2 between 16 and 1024") + + format = format or 'png' + + if format not in ('png', 'jpg', 'jpeg'): + raise InvalidArgument("format must be one of 'png', 'jpg', or 'jpeg'.") + + url = f'/avatars/{self.id}/{self.avatar}.{format}?size={size}' + return Asset(self._state, url) + + +class Webhook(BaseWebhook): + """Represents an asynchronous Discord webhook. + + Webhooks are a form to send messages to channels in Discord without a + bot user or authentication. + + There are two main ways to use Webhooks. The first is through the ones + received by the library such as :meth:`.Guild.webhooks` and + :meth:`.TextChannel.webhooks`. The ones received by the library will + automatically be bound using the library's internal HTTP session. + + The second form involves creating a webhook object manually using the + :meth:`~.Webhook.from_url` or :meth:`~.Webhook.partial` classmethods. + + For example, creating a webhook from a URL and using :doc:`aiohttp <aio:index>`: + + .. code-block:: python3 + + from discord import Webhook, AsyncWebhookAdapter + import aiohttp + + async def foo(): + async with aiohttp.ClientSession() as session: + webhook = Webhook.from_url('url-here', session=session) + await webhook.send('Hello World', username='Foo') + + For a synchronous counterpart, see :class:`SyncWebhook`. + + .. container:: operations + + .. describe:: x == y + + Checks if two webhooks are equal. + + .. describe:: x != y + + Checks if two webhooks are not equal. + + .. describe:: hash(x) + + Returns the webhooks's hash. + + .. versionchanged:: 1.4 + Webhooks are now comparable and hashable. + + Attributes + ------------ + id: :class:`int` + The webhook's ID + type: :class:`WebhookType` + The type of the webhook. + + .. versionadded:: 1.3 + + token: Optional[:class:`str`] + The authentication token of the webhook. If this is ``None`` + then the webhook cannot be used to make requests. + guild_id: Optional[:class:`int`] + The guild ID this webhook is for. + channel_id: Optional[:class:`int`] + The channel ID this webhook is for. + user: Optional[:class:`abc.User`] + The user this webhook was created by. If the webhook was + received without authentication then this will be ``None``. + name: Optional[:class:`str`] + The default name of the webhook. + avatar: Optional[:class:`str`] + The default avatar of the webhook. + source_guild: Optional[:class:`PartialWebhookGuild`] + The guild of the channel that this webhook is following. + Only given if :attr:`type` is :attr:`WebhookType.channel_follower`. + + .. versionadded:: 2.0 + + source_channel: Optional[:class:`PartialWebhookChannel`] + The channel that this webhook is following. + Only given if :attr:`type` is :attr:`WebhookType.channel_follower`. + + .. versionadded:: 2.0 + """ + + __slots__: Tuple[str, ...] = BaseWebhook.__slots__ + ('session',) + + def __init__(self, data: WebhookPayload, session: aiohttp.ClientSession, token: Optional[str] = None, state=None): + super().__init__(data, token, state) + self.session = session + + def __repr__(self): + return f'<Webhook id={self.id!r}>' + + @property + def url(self): + """:class:`str` : Returns the webhook's url.""" + return f'https://discord.com/api/webhooks/{self.id}/{self.token}' + + @classmethod + def partial(cls, id: int, token: str, *, session: aiohttp.ClientSession, bot_token: Optional[str] = None): + """Creates a partial :class:`Webhook`. + + Parameters + ----------- + id: :class:`int` + The ID of the webhook. + token: :class:`str` + The authentication token of the webhook. + session: :class:`aiohttp.ClientSession` + The session to use to send requests with. Note + that the library does not manage the session and + will not close it. + bot_token: Optional[:class:`str`] + The bot authentication token for authenticated requests + involving the webhook. + + Returns + -------- + :class:`Webhook` + A partial :class:`Webhook`. + A partial webhook is just a webhook object with an ID and a token. + """ + data: WebhookPayload = { + 'id': id, + 'type': 1, + 'token': token, + } + + return cls(data, session, token=bot_token) + + @classmethod + def from_url(cls, url: str, *, session: aiohttp.ClientSession, bot_token: Optional[str] = None): + """Creates a partial :class:`Webhook` from a webhook URL. + + Parameters + ------------ + url: :class:`str` + The URL of the webhook. + session: :class:`aiohttp.ClientSession` + The session to use to send requests with. Note + that the library does not manage the session and + will not close it. + bot_token: Optional[:class:`str`] + The bot authentication token for authenticated requests + involving the webhook. + + Raises + ------- + InvalidArgument + The URL is invalid. + + Returns + -------- + :class:`Webhook` + A partial :class:`Webhook`. + A partial webhook is just a webhook object with an ID and a token. + """ + m = re.search(r'discord(?:app)?.com/api/webhooks/(?P<id>[0-9]{17,20})/(?P<token>[A-Za-z0-9\.\-\_]{60,68})', url) + if m is None: + raise InvalidArgument('Invalid webhook URL given.') + + data: Dict[str, Any] = m.groupdict() + data['type'] = 1 + return cls(data, session, token=bot_token) # type: ignore + + @classmethod + def _as_follower(cls, data, *, channel, user): + name = f"{channel.guild} #{channel}" + feed: WebhookPayload = { + 'id': data['webhook_id'], + 'type': 2, + 'name': name, + 'channel_id': channel.id, + 'guild_id': channel.guild.id, + 'user': {'username': user.name, 'discriminator': user.discriminator, 'id': user.id, 'avatar': user.avatar}, + } + + state = channel._state + session = channel._state.http._HTTPClient__session + return cls(feed, session=session, state=state, token=state.http.token) + + @classmethod + def from_state(cls, data, state): + session = state.http._HTTPClient__session + return cls(data, session=session, state=state, token=state.http.token) + + async def fetch(self, *, prefer_auth: bool = True) -> Webhook: + """|coro| + + Fetches the current webhook. + + This could be used to get a full webhook from a partial webhook. + + .. note:: + + When fetching with an unauthenticated webhook, i.e. + :meth:`is_authenticated` returns ``False``, then the + returned webhook does not contain any user information. + + Parameters + ----------- + prefer_auth: :class:`bool` + Whether to use the bot token over the webhook token + if available. Defaults to ``True``. + + Raises + ------- + HTTPException + Could not fetch the webhook + NotFound + Could not find the webhook by this ID + InvalidArgument + This webhook does not have a token associated with it. + + Returns + -------- + :class:`Webhook` + The fetched webhook. + """ + adapter = async_context.get() + + if prefer_auth and self.auth_token: + data = await adapter.fetch_webhook(self.id, self.auth_token, session=self.session) + elif self.token: + data = await adapter.fetch_webhook_with_token(self.id, self.token, session=self.session) + else: + raise InvalidArgument('This webhook does not have a token associated with it') + + return Webhook(data, self.session, token=self.auth_token, state=self._state) + + async def delete(self, *, reason: Optional[str] = None, prefer_auth: bool = True): + """|coro| + + Deletes this Webhook. + + Parameters + ------------ + reason: Optional[:class:`str`] + The reason for deleting this webhook. Shows up on the audit log. + + .. versionadded:: 1.4 + prefer_auth: :class:`bool` + Whether to use the bot token over the webhook token + if available. Defaults to ``True``. + + Raises + ------- + HTTPException + Deleting the webhook failed. + NotFound + This webhook does not exist. + Forbidden + You do not have permissions to delete this webhook. + InvalidArgument + This webhook does not have a token associated with it. + """ + if self.token is None and self.auth_token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + adapter = async_context.get() + + if prefer_auth and self.auth_token: + await adapter.delete_webhook(self.id, token=self.auth_token, session=self.session, reason=reason) + elif self.token: + await adapter.delete_webhook_with_token(self.id, self.token, session=self.session, reason=reason) + + async def edit( + self, + *, + reason: Optional[str] = None, + name: Optional[str] = MISSING, + avatar: Optional[bytes] = MISSING, + channel: Optional[Snowflake] = None, + prefer_auth: bool = True, + ): + """|coro| + + Edits this Webhook. + + Parameters + ------------ + name: Optional[:class:`str`] + The webhook's new default name. + avatar: Optional[:class:`bytes`] + A :term:`py:bytes-like object` representing the webhook's new default avatar. + channel: Optional[:class:`abc.Snowflake`] + The webhook's new channel. This requires an authenticated webhook. + reason: Optional[:class:`str`] + The reason for editing this webhook. Shows up on the audit log. + + .. versionadded:: 1.4 + prefer_auth: :class:`bool` + Whether to use the bot token over the webhook token + if available. Defaults to ``True``. + + Raises + ------- + HTTPException + Editing the webhook failed. + NotFound + This webhook does not exist. + InvalidArgument + This webhook does not have a token associated with it + or it tried editing a channel without authentication. + """ + if self.token is None and self.auth_token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + payload = {} + if name is not MISSING: + payload['name'] = str(name) if name is not None else None + + if avatar is not MISSING: + payload['avatar'] = utils._bytes_to_base64_data(avatar) if avatar is not None else None + + adapter = async_context.get() + + # If a channel is given, always use the authenticated endpoint + if channel is not None: + if self.auth_token is None: + raise InvalidArgument('Editing channel requires authenticated webhook') + + payload['channel_id'] = channel.id + data = await adapter.edit_webhook(self.id, self.auth_token, payload=payload, session=self.session, reason=reason) + self._update(data) + return + + if prefer_auth and self.auth_token: + data = await adapter.edit_webhook(self.id, self.auth_token, payload=payload, session=self.session, reason=reason) + self._update(data) + elif self.token: + data = await adapter.edit_webhook_with_token( + self.id, self.token, payload=payload, session=self.session, reason=reason + ) + self._update(data) + + @overload + async def send( + self, + content: str = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + embed: Embed = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: AllowedMentions = MISSING, + wait: Literal[True], + ) -> WebhookMessage: + ... + + @overload + async def send( + self, + content: str = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + embed: Embed = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: AllowedMentions = MISSING, + wait: Literal[False], + ) -> None: + ... + + async def send( + self, + content: str = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = False, + file: File = MISSING, + files: List[File] = MISSING, + embed: Embed = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: AllowedMentions = MISSING, + wait: bool = False, + ) -> Optional[WebhookMessage]: + """|coro| + + Sends a message using the webhook. + + The content must be a type that can convert to a string through ``str(content)``. + + To upload a single file, the ``file`` parameter should be used with a + single :class:`File` object. + + If the ``embed`` parameter is provided, it must be of type :class:`Embed` and + it must be a rich embed type. You cannot mix the ``embed`` parameter with the + ``embeds`` parameter, which must be a :class:`list` of :class:`Embed` objects to send. + + Parameters + ------------ + content: :class:`str` + The content of the message to send. + wait: :class:`bool` + Whether the server should wait before sending a response. This essentially + means that the return type of this function changes from ``None`` to + a :class:`WebhookMessage` if set to ``True``. + username: :class:`str` + The username to send with this message. If no username is provided + then the default username for the webhook is used. + avatar_url: Union[:class:`str`, :class:`Asset`] + The avatar URL to send with this message. If no avatar URL is provided + then the default avatar for the webhook is used. + tts: :class:`bool` + Indicates if the message should be sent using text-to-speech. + file: :class:`File` + The file to upload. This cannot be mixed with ``files`` parameter. + files: List[:class:`File`] + A list of files to send with the content. This cannot be mixed with the + ``file`` parameter. + embed: :class:`Embed` + The rich embed for the content to send. This cannot be mixed with + ``embeds`` parameter. + embeds: List[:class:`Embed`] + A list of embeds to send with the content. Maximum of 10. This cannot + be mixed with the ``embed`` parameter. + allowed_mentions: :class:`AllowedMentions` + Controls the mentions being processed in this message. + + .. versionadded:: 1.4 + + Raises + -------- + HTTPException + Sending the message failed. + NotFound + This webhook was not found. + Forbidden + The authorization token for the webhook is incorrect. + TypeError + You specified both ``embed`` and ``embeds`` or ``file`` and ``files`` + ValueError + The length of ``embeds`` was invalid + InvalidArgument + There was no token associated with this webhook. + + Returns + --------- + Optional[:class:`WebhookMessage`] + The message that was sent. + """ + + if self.token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + previous_mentions: Optional[AllowedMentions] = getattr(self._state, 'allowed_mentions', None) + if content is None: + content = ... # type: ignore + + params = handle_message_parameters( + content=content, + username=username, + avatar_url=avatar_url, + tts=tts, + file=file, + files=files, + embed=embed, + embeds=embeds, + allowed_mentions=allowed_mentions, + previous_allowed_mentions=previous_mentions, + ) + adapter = async_context.get() + data = await adapter.execute_webhook( + self.id, + self.token, + session=self.session, + payload=params.payload, + multipart=params.multipart, + files=params.files, + wait=wait, + ) + if wait: + state = _WebhookState(self, parent=self._state) + channel = self.channel or Object(id=int(data['channel_id'])) + return WebhookMessage(data=data, state=state, channel=channel) + + async def edit_message( + self, + message_id: int, + *, + content: Optional[str] = MISSING, + embeds: List[Embed] = MISSING, + embed: Optional[Embed] = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + allowed_mentions: Optional[AllowedMentions] = None, + ): + """|coro| + + Edits a message owned by this webhook. + + This is a lower level interface to :meth:`WebhookMessage.edit` in case + you only have an ID. + + .. versionadded:: 1.6 + + Parameters + ------------ + message_id: :class:`int` + The message ID to edit. + content: Optional[:class:`str`] + The content to edit the message with or ``None`` to clear it. + embeds: List[:class:`Embed`] + A list of embeds to edit the message with. + embed: Optional[:class:`Embed`] + The embed to edit the message with. ``None`` suppresses the embeds. + This should not be mixed with the ``embeds`` parameter. + file: :class:`File` + The file to upload. This cannot be mixed with ``files`` parameter. + files: List[:class:`File`] + A list of files to send with the content. This cannot be mixed with the + ``file`` parameter. + allowed_mentions: :class:`AllowedMentions` + Controls the mentions being processed in this message. + See :meth:`.abc.Messageable.send` for more information. + + Raises + ------- + HTTPException + Editing the message failed. + Forbidden + Edited a message that is not yours. + TypeError + You specified both ``embed`` and ``embeds`` or ``file`` and ``files`` + ValueError + The length of ``embeds`` was invalid + InvalidArgument + There was no token associated with this webhook. + """ + + if self.token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + previous_mentions: Optional[AllowedMentions] = getattr(self._state, 'allowed_mentions', None) + params = handle_message_parameters( + content=content, + file=file, + files=files, + embed=embed, + embeds=embeds, + allowed_mentions=allowed_mentions, + previous_allowed_mentions=previous_mentions, + ) + adapter = async_context.get() + await adapter.edit_webhook_message( + self.id, + self.token, + message_id, + session=self.session, + payload=params.payload, + multipart=params.multipart, + files=params.files, + ) + + async def delete_message(self, message_id: int): + """|coro| + + Deletes a message owned by this webhook. + + This is a lower level interface to :meth:`WebhookMessage.delete` in case + you only have an ID. + + .. versionadded:: 1.6 + + Parameters + ------------ + message_id: :class:`int` + The message ID to delete. + + Raises + ------- + HTTPException + Deleting the message failed. + Forbidden + Deleted a message that is not yours. + """ + if self.token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + adapter = async_context.get() + await adapter.delete_webhook_message( + self.id, + self.token, + message_id, + session=self.session, + ) diff --git a/discord/webhook/sync.py b/discord/webhook/sync.py new file mode 100644 index 00000000..b5d0bbce --- /dev/null +++ b/discord/webhook/sync.py @@ -0,0 +1,961 @@ +""" +The MIT License (MIT) + +Copyright (c) 2015-present Rapptz + +Permission is hereby granted, free of charge, to any person obtaining a +copy of this software and associated documentation files (the "Software"), +to deal in the Software without restriction, including without limitation +the rights to use, copy, modify, merge, publish, distribute, sublicense, +and/or sell copies of the Software, and to permit persons to whom the +Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. +""" + +# If you're wondering why this is essentially copy pasted from the async_.py +# file, then it's due to needing two separate types to make the typing shenanigans +# a bit easier to write. It's an unfortunate design. Originally, these types were +# merged and an adapter was used to differentiate between the async and sync versions. +# However, this proved to be difficult to provide typings for, so here we are. + +from __future__ import annotations + +import threading +import logging +import json +import time +import re + +from urllib.parse import quote as urlquote +from typing import Any, Dict, List, Literal, Optional, TYPE_CHECKING, Tuple, Type, TypeVar, Union, overload + +from .. import utils +from ..errors import InvalidArgument, HTTPException, Forbidden, NotFound, DiscordServerError +from ..message import Message +from ..http import Route +from ..object import Object + +from .async_ import BaseWebhook, handle_message_parameters, _WebhookState, MISSING + +__all__ = ( + 'SyncWebhook', + 'SyncWebhookMessage', +) + +log = logging.getLogger(__name__) + +if TYPE_CHECKING: + from ..file import File + from ..embeds import Embed + from ..mentions import AllowedMentions + from ..types.webhook import ( + Webhook as WebhookPayload, + ) + from ..abc import Snowflake + + try: + from requests import Session, Response + except ModuleNotFoundError: + pass + + +class DeferredLock: + def __init__(self, lock: threading.Lock): + self.lock = lock + self.delta: Optional[float] = None + + def __enter__(self): + self.lock.acquire() + return self + + def delay_by(self, delta: float) -> None: + self.delta = delta + + def __exit__(self, type, value, traceback): + if self.delta: + time.sleep(self.delta) + self.lock.release() + + +class WebhookAdapter: + def __init__(self): + self._locks: Dict[Any, threading.Lock] = {} + + def request( + self, + route: Route, + session: Session, + *, + payload: Optional[Dict[str, Any]] = None, + multipart: Optional[List[Dict[str, Any]]] = None, + files: Optional[List[File]] = None, + reason: Optional[str] = None, + auth_token: Optional[str] = None, + params: Optional[Dict[str, Any]] = None, + ) -> Any: + headers: Dict[str, str] = {} + files = files or [] + to_send: Optional[Union[str, Dict[str, Any]]] = None + bucket = (route.webhook_id, route.webhook_token) + + try: + lock = self._locks[bucket] + except KeyError: + self._locks[bucket] = lock = threading.Lock() + + if payload is not None: + headers['Content-Type'] = 'application/json' + to_send = utils.to_json(payload) + + if auth_token is not None: + headers['Authorization'] = f'Bot {auth_token}' + + if reason is not None: + headers['X-Audit-Log-Reason'] = urlquote(reason, safe='/ ') + + response: Optional[Response] = None + data: Optional[Union[Dict[str, Any], str]] = None + method = route.method + url = route.url + webhook_id = route.webhook_id + + with DeferredLock(lock) as lock: + for attempt in range(5): + for file in files: + file.reset(seek=attempt) + + if multipart: + form_data: Dict[str, Any] = {} + for p in multipart: + form_data[p['name']] = (p['filename'], p['value'], p['content_type']) + to_send = form_data + + try: + with session.request(method, url, data=to_send, headers=headers, params=params) as response: + log.debug( + 'Webhook ID %s with %s %s has returned status code %s', + webhook_id, + method, + url, + response.status_code, + ) + response.encoding = 'utf-8' + data = response.text or None + if data and response.headers['Content-Type'] == 'application/json': + data = json.loads(data) + + remaining = response.headers.get('X-Ratelimit-Remaining') + if remaining == '0' and response.status_code != 429: + delta = utils._parse_ratelimit_header(response.headers) + log.debug( + 'Webhook ID %s has been pre-emptively rate limited, waiting %.2f seconds', webhook_id, delta + ) + lock.delay_by(delta) + + if 300 > response.status_code >= 200: + return data + + if response.status_code == 429: + if not response.headers.get('Via'): + raise HTTPException(response, data) + + retry_after = data['retry_after'] / 1000.0 # type: ignore + log.warning('Webhook ID %s is rate limited. Retrying in %.2f seconds', webhook_id, retry_after) + time.sleep(retry_after) + + if response.status_code >= 500: + time.sleep(1 + attempt * 2) + continue + + if response.status_code == 403: + raise Forbidden(response, data) + elif response.status_code == 404: + raise NotFound(response, data) + else: + raise HTTPException(response, data) + + except OSError as e: + if attempt < 4 and e.errno in (54, 10054): + continue + raise + + if response: + if response.status_code >= 500: + raise DiscordServerError(response, data) + raise HTTPException(response, data) + + raise RuntimeError('Unreachable code in HTTP handling.') + + def delete_webhook( + self, + webhook_id: int, + *, + token: Optional[str] = None, + session: Session, + reason: Optional[str] = None, + ): + route = Route('DELETE', '/webhooks/{webhook_id}', webhook_id=webhook_id) + return self.request(route, session, reason=reason, auth_token=token) + + def delete_webhook_with_token( + self, + webhook_id: int, + token: str, + *, + session: Session, + reason: Optional[str] = None, + ): + route = Route('DELETE', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session, reason=reason) + + def edit_webhook( + self, + webhook_id: int, + token: str, + payload: Dict[str, Any], + *, + session: Session, + reason: Optional[str] = None, + ): + route = Route('PATCH', '/webhooks/{webhook_id}', webhook_id=webhook_id) + return self.request(route, session, reason=reason, payload=payload, auth_token=token) + + def edit_webhook_with_token( + self, + webhook_id: int, + token: str, + payload: Dict[str, Any], + *, + session: Session, + reason: Optional[str] = None, + ): + route = Route('PATCH', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session, reason=reason, payload=payload) + + def execute_webhook( + self, + webhook_id: int, + token: str, + *, + session: Session, + payload: Optional[Dict[str, Any]] = None, + multipart: Optional[List[Dict[str, Any]]] = None, + files: Optional[List[File]] = None, + wait: bool = False, + ): + params = {'wait': int(wait)} + route = Route('POST', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session, payload=payload, multipart=multipart, files=files, params=params) + + def edit_webhook_message( + self, + webhook_id: int, + token: str, + message_id: int, + *, + session: Session, + payload: Optional[Dict[str, Any]] = None, + multipart: Optional[List[Dict[str, Any]]] = None, + files: Optional[List[File]] = None, + ): + route = Route( + 'PATCH', + '/webhooks/{webhook_id}/{webhook_token}/messages/{message_id}', + webhook_id=webhook_id, + webhook_token=token, + message_id=message_id, + ) + return self.request(route, session, payload=payload, multipart=multipart, files=files) + + def delete_webhook_message( + self, + webhook_id: int, + token: str, + message_id: int, + *, + session: Session, + ): + route = Route( + 'DELETE', + '/webhooks/{webhook_id}/{webhook_token}/messages/{message_id}', + webhook_id=webhook_id, + webhook_token=token, + message_id=message_id, + ) + return self.request(route, session) + + def fetch_webhook( + self, + webhook_id: int, + token: str, + *, + session: Session, + ): + route = Route('GET', '/webhooks/{webhook_id}', webhook_id=webhook_id) + return self.request(route, session=session, auth_token=token) + + def fetch_webhook_with_token( + self, + webhook_id: int, + token: str, + *, + session: Session, + ): + route = Route('GET', '/webhooks/{webhook_id}/{webhook_token}', webhook_id=webhook_id, webhook_token=token) + return self.request(route, session=session) + +_context = threading.local() +_context.adapter = WebhookAdapter() + + +class SyncWebhookMessage(Message): + """Represents a message sent from your webhook. + + This allows you to edit or delete a message sent by your + webhook. + + This inherits from :class:`discord.Message` with changes to + :meth:`edit` and :meth:`delete` to work. + + .. versionadded:: 2.0 + """ + + def edit( + self, + content: Optional[str] = MISSING, + embeds: List[Embed] = MISSING, + embed: Optional[Embed] = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + allowed_mentions: Optional[AllowedMentions] = None, + ): + """Edits a message owned by this webhook. + + This is a lower level interface to :meth:`WebhookMessage.edit` in case + you only have an ID. + + Parameters + ------------ + message_id: :class:`int` + The message ID to edit. + content: Optional[:class:`str`] + The content to edit the message with or ``None`` to clear it. + embeds: List[:class:`Embed`] + A list of embeds to edit the message with. + embed: Optional[:class:`Embed`] + The embed to edit the message with. ``None`` suppresses the embeds. + This should not be mixed with the ``embeds`` parameter. + file: :class:`File` + The file to upload. This cannot be mixed with ``files`` parameter. + files: List[:class:`File`] + A list of files to send with the content. This cannot be mixed with the + ``file`` parameter. + allowed_mentions: :class:`AllowedMentions` + Controls the mentions being processed in this message. + See :meth:`.abc.Messageable.send` for more information. + + Raises + ------- + HTTPException + Editing the message failed. + Forbidden + Edited a message that is not yours. + TypeError + You specified both ``embed`` and ``embeds`` or ``file`` and ``files`` + ValueError + The length of ``embeds`` was invalid + InvalidArgument + There was no token associated with this webhook. + """ + self._state._webhook.edit_message( + self.id, + content=content, + embeds=embeds, + embed=embed, + file=file, + files=files, + allowed_mentions=allowed_mentions, + ) + + def delete(self, *, delay: Optional[float] = None) -> None: + """Deletes the message. + + Parameters + ----------- + delay: Optional[:class:`float`] + If provided, the number of seconds to wait before deleting the message. + This blocks the thread. + + Raises + ------ + Forbidden + You do not have proper permissions to delete the message. + NotFound + The message was deleted already. + HTTPException + Deleting the message failed. + """ + + if delay is not None: + time.sleep(delay) + self._state._webhook.delete_message(self.id) + + +class SyncWebhook(BaseWebhook): + """Represents a synchronous Discord webhook. + + For an asynchronous counterpart, see :class:`Webhook`. + + .. container:: operations + + .. describe:: x == y + + Checks if two webhooks are equal. + + .. describe:: x != y + + Checks if two webhooks are not equal. + + .. describe:: hash(x) + + Returns the webhooks's hash. + + .. versionchanged:: 1.4 + Webhooks are now comparable and hashable. + + Attributes + ------------ + id: :class:`int` + The webhook's ID + type: :class:`WebhookType` + The type of the webhook. + + .. versionadded:: 1.3 + + token: Optional[:class:`str`] + The authentication token of the webhook. If this is ``None`` + then the webhook cannot be used to make requests. + guild_id: Optional[:class:`int`] + The guild ID this webhook is for. + channel_id: Optional[:class:`int`] + The channel ID this webhook is for. + user: Optional[:class:`abc.User`] + The user this webhook was created by. If the webhook was + received without authentication then this will be ``None``. + name: Optional[:class:`str`] + The default name of the webhook. + avatar: Optional[:class:`str`] + The default avatar of the webhook. + source_guild: Optional[:class:`PartialWebhookGuild`] + The guild of the channel that this webhook is following. + Only given if :attr:`type` is :attr:`WebhookType.channel_follower`. + + .. versionadded:: 2.0 + + source_channel: Optional[:class:`PartialWebhookChannel`] + The channel that this webhook is following. + Only given if :attr:`type` is :attr:`WebhookType.channel_follower`. + + .. versionadded:: 2.0 + """ + + __slots__: Tuple[str, ...] = BaseWebhook.__slots__ + ('session',) + + def __init__(self, data: WebhookPayload, session: Session, token: Optional[str] = None, state=None): + super().__init__(data, token, state) + self.session = session + + def __repr__(self): + return f'<Webhook id={self.id!r}>' + + @property + def url(self): + """:class:`str` : Returns the webhook's url.""" + return f'https://discord.com/api/webhooks/{self.id}/{self.token}' + + @classmethod + def partial(cls, id: int, token: str, *, session: Session = MISSING, bot_token: Optional[str] = None): + """Creates a partial :class:`Webhook`. + + Parameters + ----------- + id: :class:`int` + The ID of the webhook. + token: :class:`str` + The authentication token of the webhook. + session: :class:`requests.Session` + The session to use to send requests with. Note + that the library does not manage the session and + will not close it. If not given, the ``requests`` + auto session creation functions are used instead. + bot_token: Optional[:class:`str`] + The bot authentication token for authenticated requests + involving the webhook. + + Returns + -------- + :class:`Webhook` + A partial :class:`Webhook`. + A partial webhook is just a webhook object with an ID and a token. + """ + data: WebhookPayload = { + 'id': id, + 'type': 1, + 'token': token, + } + import requests + + if session is not MISSING: + if not isinstance(session, requests.Session): + raise TypeError(f'expected requests.Session not {session.__class__!r}') + else: + session = requests # type: ignore + return cls(data, session, token=bot_token) + + @classmethod + def from_url(cls, url: str, *, session: Session = MISSING, bot_token: Optional[str] = None): + """Creates a partial :class:`Webhook` from a webhook URL. + + Parameters + ------------ + url: :class:`str` + The URL of the webhook. + session: :class:`requests.Session` + The session to use to send requests with. Note + that the library does not manage the session and + will not close it. If not given, the ``requests`` + auto session creation functions are used instead. + bot_token: Optional[:class:`str`] + The bot authentication token for authenticated requests + involving the webhook. + + Raises + ------- + InvalidArgument + The URL is invalid. + + Returns + -------- + :class:`Webhook` + A partial :class:`Webhook`. + A partial webhook is just a webhook object with an ID and a token. + """ + m = re.search(r'discord(?:app)?.com/api/webhooks/(?P<id>[0-9]{17,20})/(?P<token>[A-Za-z0-9\.\-\_]{60,68})', url) + if m is None: + raise InvalidArgument('Invalid webhook URL given.') + + data: Dict[str, Any] = m.groupdict() + data['type'] = 1 + import requests + + if session is not MISSING: + if not isinstance(session, requests.Session): + raise TypeError(f'expected requests.Session not {session.__class__!r}') + else: + session = requests # type: ignore + return cls(data, session, token=bot_token) # type: ignore + + def fetch(self, *, prefer_auth: bool = True) -> SyncWebhook: + """Fetches the current webhook. + + This could be used to get a full webhook from a partial webhook. + + .. note:: + + When fetching with an unauthenticated webhook, i.e. + :meth:`is_authenticated` returns ``False``, then the + returned webhook does not contain any user information. + + Parameters + ----------- + prefer_auth: :class:`bool` + Whether to use the bot token over the webhook token + if available. Defaults to ``True``. + + Raises + ------- + HTTPException + Could not fetch the webhook + NotFound + Could not find the webhook by this ID + InvalidArgument + This webhook does not have a token associated with it. + + Returns + -------- + :class:`SyncWebhook` + The fetched webhook. + """ + adapter: WebhookAdapter = _context.adapter + + if prefer_auth and self.auth_token: + data = adapter.fetch_webhook(self.id, self.auth_token, session=self.session) + elif self.token: + data = adapter.fetch_webhook_with_token(self.id, self.token, session=self.session) + else: + raise InvalidArgument('This webhook does not have a token associated with it') + + return SyncWebhook(data, self.session, token=self.auth_token, state=self._state) + + def delete(self, *, reason: Optional[str] = None, prefer_auth: bool = True): + """Deletes this Webhook. + + Parameters + ------------ + reason: Optional[:class:`str`] + The reason for deleting this webhook. Shows up on the audit log. + + .. versionadded:: 1.4 + prefer_auth: :class:`bool` + Whether to use the bot token over the webhook token + if available. Defaults to ``True``. + + Raises + ------- + HTTPException + Deleting the webhook failed. + NotFound + This webhook does not exist. + Forbidden + You do not have permissions to delete this webhook. + InvalidArgument + This webhook does not have a token associated with it. + """ + if self.token is None and self.auth_token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + adapter: WebhookAdapter = _context.adapter + + if prefer_auth and self.auth_token: + adapter.delete_webhook(self.id, token=self.auth_token, session=self.session, reason=reason) + elif self.token: + adapter.delete_webhook_with_token(self.id, self.token, session=self.session, reason=reason) + + def edit( + self, + *, + reason: Optional[str] = None, + name: Optional[str] = MISSING, + avatar: Optional[bytes] = MISSING, + channel: Optional[Snowflake] = None, + prefer_auth: bool = True, + ): + """Edits this Webhook. + + Parameters + ------------ + name: Optional[:class:`str`] + The webhook's new default name. + avatar: Optional[:class:`bytes`] + A :term:`py:bytes-like object` representing the webhook's new default avatar. + channel: Optional[:class:`abc.Snowflake`] + The webhook's new channel. This requires an authenticated webhook. + reason: Optional[:class:`str`] + The reason for editing this webhook. Shows up on the audit log. + + .. versionadded:: 1.4 + prefer_auth: :class:`bool` + Whether to use the bot token over the webhook token + if available. Defaults to ``True``. + + Raises + ------- + HTTPException + Editing the webhook failed. + NotFound + This webhook does not exist. + InvalidArgument + This webhook does not have a token associated with it + or it tried editing a channel without authentication. + """ + if self.token is None and self.auth_token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + payload = {} + if name is not MISSING: + payload['name'] = str(name) if name is not None else None + + if avatar is not MISSING: + payload['avatar'] = utils._bytes_to_base64_data(avatar) if avatar is not None else None + + adapter: WebhookAdapter = _context.adapter + + # If a channel is given, always use the authenticated endpoint + if channel is not None: + if self.auth_token is None: + raise InvalidArgument('Editing channel requires authenticated webhook') + + payload['channel_id'] = channel.id + data = adapter.edit_webhook(self.id, self.auth_token, payload=payload, session=self.session, reason=reason) + self._update(data) + return + + if prefer_auth and self.auth_token: + data = adapter.edit_webhook(self.id, self.auth_token, payload=payload, session=self.session, reason=reason) + self._update(data) + elif self.token: + data = adapter.edit_webhook_with_token(self.id, self.token, payload=payload, session=self.session, reason=reason) + self._update(data) + + @overload + def send( + self, + content: str = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + embed: Embed = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: AllowedMentions = MISSING, + wait: Literal[True], + ) -> SyncWebhookMessage: + ... + + @overload + def send( + self, + content: str = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + embed: Embed = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: AllowedMentions = MISSING, + wait: Literal[False], + ) -> None: + ... + + def send( + self, + content: str = MISSING, + *, + username: str = MISSING, + avatar_url: str = MISSING, + tts: bool = False, + file: File = MISSING, + files: List[File] = MISSING, + embed: Embed = MISSING, + embeds: List[Embed] = MISSING, + allowed_mentions: AllowedMentions = MISSING, + wait: bool = False, + ) -> Optional[SyncWebhookMessage]: + """Sends a message using the webhook. + + The content must be a type that can convert to a string through ``str(content)``. + + To upload a single file, the ``file`` parameter should be used with a + single :class:`File` object. + + If the ``embed`` parameter is provided, it must be of type :class:`Embed` and + it must be a rich embed type. You cannot mix the ``embed`` parameter with the + ``embeds`` parameter, which must be a :class:`list` of :class:`Embed` objects to send. + + Parameters + ------------ + content: :class:`str` + The content of the message to send. + wait: :class:`bool` + Whether the server should wait before sending a response. This essentially + means that the return type of this function changes from ``None`` to + a :class:`WebhookMessage` if set to ``True``. + username: :class:`str` + The username to send with this message. If no username is provided + then the default username for the webhook is used. + avatar_url: Union[:class:`str`, :class:`Asset`] + The avatar URL to send with this message. If no avatar URL is provided + then the default avatar for the webhook is used. + tts: :class:`bool` + Indicates if the message should be sent using text-to-speech. + file: :class:`File` + The file to upload. This cannot be mixed with ``files`` parameter. + files: List[:class:`File`] + A list of files to send with the content. This cannot be mixed with the + ``file`` parameter. + embed: :class:`Embed` + The rich embed for the content to send. This cannot be mixed with + ``embeds`` parameter. + embeds: List[:class:`Embed`] + A list of embeds to send with the content. Maximum of 10. This cannot + be mixed with the ``embed`` parameter. + allowed_mentions: :class:`AllowedMentions` + Controls the mentions being processed in this message. + + .. versionadded:: 1.4 + + Raises + -------- + HTTPException + Sending the message failed. + NotFound + This webhook was not found. + Forbidden + The authorization token for the webhook is incorrect. + TypeError + You specified both ``embed`` and ``embeds`` or ``file`` and ``files`` + ValueError + The length of ``embeds`` was invalid + InvalidArgument + There was no token associated with this webhook. + + Returns + --------- + Optional[:class:`SyncWebhookMessage`] + The message that was sent. + """ + + if self.token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + previous_mentions: Optional[AllowedMentions] = getattr(self._state, 'allowed_mentions', None) + if content is None: + content = ... # type: ignore + + params = handle_message_parameters( + content=content, + username=username, + avatar_url=avatar_url, + tts=tts, + file=file, + files=files, + embed=embed, + embeds=embeds, + allowed_mentions=allowed_mentions, + previous_allowed_mentions=previous_mentions, + ) + adapter: WebhookAdapter = _context.adapter + data = adapter.execute_webhook( + self.id, + self.token, + session=self.session, + payload=params.payload, + multipart=params.multipart, + files=params.files, + wait=wait, + ) + if wait: + state = _WebhookState(self, parent=self._state) + channel = self.channel or Object(id=int(data['channel_id'])) + return SyncWebhookMessage(data=data, state=state, channel=channel) + + def edit_message( + self, + message_id: int, + *, + content: Optional[str] = MISSING, + embeds: List[Embed] = MISSING, + embed: Optional[Embed] = MISSING, + file: File = MISSING, + files: List[File] = MISSING, + allowed_mentions: Optional[AllowedMentions] = None, + ): + """Edits a message owned by this webhook. + + This is a lower level interface to :meth:`WebhookMessage.edit` in case + you only have an ID. + + .. versionadded:: 1.6 + + Parameters + ------------ + message_id: :class:`int` + The message ID to edit. + content: Optional[:class:`str`] + The content to edit the message with or ``None`` to clear it. + embeds: List[:class:`Embed`] + A list of embeds to edit the message with. + embed: Optional[:class:`Embed`] + The embed to edit the message with. ``None`` suppresses the embeds. + This should not be mixed with the ``embeds`` parameter. + file: :class:`File` + The file to upload. This cannot be mixed with ``files`` parameter. + files: List[:class:`File`] + A list of files to send with the content. This cannot be mixed with the + ``file`` parameter. + allowed_mentions: :class:`AllowedMentions` + Controls the mentions being processed in this message. + See :meth:`.abc.Messageable.send` for more information. + + Raises + ------- + HTTPException + Editing the message failed. + Forbidden + Edited a message that is not yours. + TypeError + You specified both ``embed`` and ``embeds`` or ``file`` and ``files`` + ValueError + The length of ``embeds`` was invalid + InvalidArgument + There was no token associated with this webhook. + """ + + if self.token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + previous_mentions: Optional[AllowedMentions] = getattr(self._state, 'allowed_mentions', None) + params = handle_message_parameters( + content=content, + file=file, + files=files, + embed=embed, + embeds=embeds, + allowed_mentions=allowed_mentions, + previous_allowed_mentions=previous_mentions, + ) + adapter: WebhookAdapter = _context.adapter + adapter.edit_webhook_message( + self.id, + self.token, + message_id, + session=self.session, + payload=params.payload, + multipart=params.multipart, + files=params.files, + ) + + def delete_message(self, message_id: int): + """Deletes a message owned by this webhook. + + This is a lower level interface to :meth:`WebhookMessage.delete` in case + you only have an ID. + + .. versionadded:: 1.6 + + Parameters + ------------ + message_id: :class:`int` + The message ID to delete. + + Raises + ------- + HTTPException + Deleting the message failed. + Forbidden + Deleted a message that is not yours. + """ + if self.token is None: + raise InvalidArgument('This webhook does not have a token associated with it') + + adapter: WebhookAdapter = _context.adapter + adapter.delete_webhook_message( + self.id, + self.token, + message_id, + session=self.session, + ) |