Source code for discord.webhook.sync

"""
The MIT License (MIT)

Copyright (c) 2015-2021 Rapptz
Copyright (c) 2021-present Pycord Development

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 json
import logging
import re
import threading
import time
import weakref
from typing import TYPE_CHECKING, Any, Literal, overload
from urllib.parse import quote as urlquote

from .. import utils
from ..channel import PartialMessageable
from ..errors import (
    DiscordServerError,
    Forbidden,
    HTTPException,
    InvalidArgument,
    NotFound,
)
from ..http import Route
from ..message import Message
from ..object import Object
from ..threads import Thread
from .async_ import BaseWebhook, _WebhookState, handle_message_parameters

__all__ = (
    "SyncWebhook",
    "SyncWebhookMessage",
)

_log = logging.getLogger(__name__)

if TYPE_CHECKING:
    from ..abc import Snowflake
    from ..embeds import Embed
    from ..file import File
    from ..mentions import AllowedMentions
    from ..types.webhook import Webhook as WebhookPayload

    try:
        from requests import Response, Session
    except ModuleNotFoundError:
        pass

MISSING = utils.MISSING


class DeferredLock:
    def __init__(self, lock: threading.Lock):
        self.lock = lock
        self.delta: float | None = 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: weakref.WeakValueDictionary = weakref.WeakValueDictionary()

    def request(
        self,
        route: Route,
        session: Session,
        *,
        payload: dict[str, Any] | None = None,
        multipart: list[dict[str, Any]] | None = None,
        files: list[File] | None = None,
        reason: str | None = None,
        auth_token: str | None = None,
        params: dict[str, Any] | None = None,
    ) -> Any:
        headers: dict[str, str] = {}
        files = files or []
        to_send: str | dict[str, Any] | None = 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: Response | None = None
        data: dict[str, Any] | str | None = None
        file_data: dict[str, Any] | None = 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:
                    file_data = {}
                    for p in multipart:
                        name = p["name"]
                        if name == "payload_json":
                            to_send = {"payload_json": p["value"]}
                        else:
                            file_data[name] = (
                                p["filename"],
                                p["value"],
                                p["content_type"],
                            )

                try:
                    with session.request(
                        method,
                        url,
                        data=to_send,
                        files=file_data,
                        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"
                        # Compatibility with aiohttp
                        response.status = response.status_code  # type: ignore

                        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)
                            _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: float = data["retry_after"]  # type: ignore
                            _log.warning(
                                "Webhook ID %s is rate limited. Retrying in %.2f"
                                " seconds",
                                webhook_id,
                                retry_after,
                            )
                            time.sleep(retry_after)
                            continue

                        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):
                        time.sleep(1 + attempt * 2)
                        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: str | None = None,
        session: Session,
        reason: str | None = 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: str | None = 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: str | None = 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: str | None = 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: dict[str, Any] | None = None,
        multipart: list[dict[str, Any]] | None = None,
        files: list[File] | None = None,
        thread_id: int | None = None,
        thread_name: str | None = None,
        wait: bool = False,
    ):
        params = {"wait": int(wait)}
        if thread_id:
            params["thread_id"] = thread_id

        if thread_name:
            payload["thread_name"] = thread_name

        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 get_webhook_message(
        self,
        webhook_id: int,
        token: str,
        message_id: int,
        *,
        session: Session,
        thread_id: int | None = None,
    ):
        params = {}

        if thread_id:
            params["thread_id"] = thread_id

        route = Route(
            "GET",
            "/webhooks/{webhook_id}/{webhook_token}/messages/{message_id}",
            webhook_id=webhook_id,
            webhook_token=token,
            message_id=message_id,
        )
        return self.request(route, session, params=params)

    def edit_webhook_message(
        self,
        webhook_id: int,
        token: str,
        message_id: int,
        *,
        session: Session,
        thread_id: int | None = None,
        payload: dict[str, Any] | None = None,
        multipart: list[dict[str, Any]] | None = None,
        files: list[File] | None = None,
    ):
        params = {}

        if thread_id:
            params["thread_id"] = thread_id

        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,
            params=params,
            payload=payload,
            multipart=multipart,
            files=files,
        )

    def delete_webhook_message(
        self,
        webhook_id: int,
        token: str,
        message_id: int,
        *,
        session: Session,
        thread_id: int | None = None,
    ):
        params = {}

        if thread_id:
            params["thread_id"] = thread_id

        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, params=params)

    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)


class _WebhookContext(threading.local):
    adapter: WebhookAdapter | None = None


_context = _WebhookContext()


def _get_webhook_adapter() -> WebhookAdapter:
    if _context.adapter is None:
        _context.adapter = WebhookAdapter()
    return _context.adapter


[docs]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 """ _state: _WebhookState
[docs] def edit( self, content: str | None = MISSING, embeds: list[Embed] = MISSING, embed: Embed | None = MISSING, file: File = MISSING, files: list[File] = MISSING, allowed_mentions: AllowedMentions | None = None, ) -> SyncWebhookMessage: """Edits the message. 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. 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. Returns ------- :class:`SyncWebhookMessage` The newly edited message. 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. """ thread = MISSING if hasattr(self, "_thread_id"): thread = Object(self._thread_id) elif isinstance(self.channel, Thread): thread = Object(self.channel.id) return self._state._webhook.edit_message( self.id, content=content, embeds=embeds, embed=embed, file=file, files=files, allowed_mentions=allowed_mentions, thread=thread, )
[docs] def delete(self, *, delay: float | None = 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. """ thread_id: int | None = None if hasattr(self, "_thread_id"): thread_id = self._thread_id elif isinstance(self.channel, Thread): thread_id = self.channel.id if delay is not None: time.sleep(delay) self._state._webhook.delete_message(self.id, thread_id=thread_id)
[docs]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 webhook'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. 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, ...] = ("session",) def __init__( self, data: WebhookPayload, session: Session, token: str | None = 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) -> str: """Returns the webhook's url.""" return f"https://discord.com/api/webhooks/{self.id}/{self.token}"
[docs] @classmethod def partial( cls, id: int, token: str, *, session: Session = MISSING, bot_token: str | None = None, ) -> SyncWebhook: """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 MISSING: session = requests # type: ignore elif not isinstance(session, requests.Session): raise TypeError(f"expected requests.Session not {session.__class__!r}") return cls(data, session, token=bot_token)
[docs] @classmethod def from_url( cls, url: str, *, session: Session = MISSING, bot_token: str | None = None ) -> SyncWebhook: """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. Returns ------- :class:`Webhook` A partial :class:`Webhook`. A partial webhook is just a webhook object with an ID and a token. Raises ------ InvalidArgument The URL is invalid. """ m = re.search( r"discord(?:app)?.com/api/webhooks/(?P<id>\d{17,20})/(?P<token>[\w\.\-_]{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 MISSING: session = requests # type: ignore elif not isinstance(session, requests.Session): raise TypeError(f"expected requests.Session not {session.__class__!r}") return cls(data, session, token=bot_token) # type: ignore
[docs] 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``. Returns ------- :class:`SyncWebhook` The fetched webhook. 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. """ adapter: WebhookAdapter = _get_webhook_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)
[docs] def delete(self, *, reason: str | None = None, prefer_auth: bool = True) -> None: """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 = _get_webhook_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 )
[docs] def edit( self, *, reason: str | None = None, name: str | None = MISSING, avatar: bytes | None = MISSING, channel: Snowflake | None = None, prefer_auth: bool = True, ) -> SyncWebhook: """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``. Returns ------- :class:`SyncWebhook` The newly edited webhook. 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 = _get_webhook_adapter() data: WebhookPayload | None = None # 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, ) if prefer_auth and self.auth_token: data = adapter.edit_webhook( self.id, self.auth_token, payload=payload, session=self.session, reason=reason, ) elif self.token: data = adapter.edit_webhook_with_token( self.id, self.token, payload=payload, session=self.session, reason=reason, ) if data is None: raise RuntimeError("Unreachable code hit: data was not assigned") return SyncWebhook( data=data, session=self.session, token=self.auth_token, state=self._state )
def _create_message(self, data): state = _WebhookState(self, parent=self._state) # state may be artificial (unlikely at this point...) channel = self.channel or PartialMessageable(state=self._state, id=int(data["channel_id"])) # type: ignore # state is artificial return SyncWebhookMessage(data=data, state=state, channel=channel) # type: ignore @overload def send( self, content: str = MISSING, *, username: str = MISSING, avatar_url: Any = MISSING, tts: bool = MISSING, file: File = MISSING, files: list[File] = MISSING, embed: Embed = MISSING, embeds: list[Embed] = MISSING, allowed_mentions: AllowedMentions = MISSING, thread: Snowflake = MISSING, thread_name: str | None = None, wait: Literal[True], ) -> SyncWebhookMessage: ... @overload def send( self, content: str = MISSING, *, username: str = MISSING, avatar_url: Any = MISSING, tts: bool = MISSING, file: File = MISSING, files: list[File] = MISSING, embed: Embed = MISSING, embeds: list[Embed] = MISSING, allowed_mentions: AllowedMentions = MISSING, thread: Snowflake = MISSING, thread_name: str | None = None, wait: Literal[False] = ..., ) -> None: ...
[docs] def send( self, content: str = MISSING, *, username: str = MISSING, avatar_url: Any = MISSING, tts: bool = False, file: File = MISSING, files: list[File] = MISSING, embed: Embed = MISSING, embeds: list[Embed] = MISSING, allowed_mentions: AllowedMentions = MISSING, thread: Snowflake = MISSING, thread_name: str | None = None, wait: bool = False, ) -> SyncWebhookMessage | None: """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: :class:`str` The avatar URL to send with this message. If no avatar URL is provided then the default avatar for the webhook is used. If this is not a string then it is explicitly cast using ``str``. 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 thread: :class:`~discord.abc.Snowflake` The thread to send this message to. .. versionadded:: 2.0 thread_name: :class:`str` The name of the thread to create. Only works for forum channels. .. versionadded:: 2.0 Returns ------- Optional[:class:`SyncWebhookMessage`] If ``wait`` is ``True`` then the message that was sent, otherwise ``None``. 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, or you specified both a thread to send to and a thread to create (the ``thread`` and ``thread_name`` parameters). """ if self.token is None: raise InvalidArgument( "This webhook does not have a token associated with it" ) previous_mentions: AllowedMentions | None = getattr( self._state, "allowed_mentions", None ) if content is None: content = MISSING if thread and thread_name: raise InvalidArgument("You cannot specify both a thread and a thread name") 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 = _get_webhook_adapter() thread_id: int | None = None if thread is not MISSING: thread_id = thread.id data = adapter.execute_webhook( self.id, self.token, session=self.session, payload=params.payload, multipart=params.multipart, files=params.files, thread_id=thread_id, thread_name=thread_name, wait=wait, ) if wait: return self._create_message(data)
[docs] def fetch_message( self, id: int, *, thread_id: int | None = None ) -> SyncWebhookMessage: """Retrieves a single :class:`~discord.SyncWebhookMessage` owned by this webhook. .. versionadded:: 2.0 Parameters ---------- id: :class:`int` The message ID to look for. thread_id: Optional[:class:`int`] The ID of the thread that contains the message. Returns ------- :class:`~discord.SyncWebhookMessage` The message asked for. Raises ------ ~discord.NotFound The specified message was not found. ~discord.Forbidden You do not have the permissions required to get a message. ~discord.HTTPException Retrieving the message failed. 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" ) adapter: WebhookAdapter = _get_webhook_adapter() data = adapter.get_webhook_message( self.id, self.token, id, session=self.session, thread_id=thread_id, ) msg = self._create_message(data) if isinstance(msg.channel, PartialMessageable): msg._thread_id = thread_id return msg
[docs] def edit_message( self, message_id: int, *, content: str | None = MISSING, embeds: list[Embed] = MISSING, embed: Embed | None = MISSING, file: File = MISSING, files: list[File] = MISSING, allowed_mentions: AllowedMentions | None = None, thread: Snowflake | None = MISSING, ) -> SyncWebhookMessage: """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. thread: Optional[:class:`~discord.abc.Snowflake`] The thread that contains the message. 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: AllowedMentions | None = 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 = _get_webhook_adapter() thread_id: int | None = None if thread is not MISSING: thread_id = thread.id data = adapter.edit_webhook_message( self.id, self.token, message_id, session=self.session, thread_id=thread_id, payload=params.payload, multipart=params.multipart, files=params.files, ) return self._create_message(data)
[docs] def delete_message(self, message_id: int, *, thread_id: int | None = None) -> None: """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. thread_id: Optional[:class:`int`] The ID of the thread that contains the message. 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 = _get_webhook_adapter() adapter.delete_webhook_message( self.id, self.token, message_id, session=self.session, thread_id=thread_id, )