From 9fabf1fbacda20017afc0b54d4473749b2867039 Mon Sep 17 00:00:00 2001 From: Slipstream Date: Sun, 15 Jun 2025 18:49:37 -0600 Subject: [PATCH] Add Webhook.from_token and fetch support (#110) --- disagreement/http.py | 23 +++++++-- disagreement/models.py | 113 +++++++++++++++++++++++++---------------- tests/test_webhooks.py | 10 ++++ 3 files changed, 99 insertions(+), 47 deletions(-) diff --git a/disagreement/http.py b/disagreement/http.py index 543b9ee..ab50062 100644 --- a/disagreement/http.py +++ b/disagreement/http.py @@ -748,10 +748,25 @@ class HTTPClient: return Webhook(data) - async def delete_webhook(self, webhook_id: "Snowflake") -> None: - """Deletes a webhook.""" - - await self.request("DELETE", f"/webhooks/{webhook_id}") + async def delete_webhook(self, webhook_id: "Snowflake") -> None: + """Deletes a webhook.""" + + await self.request("DELETE", f"/webhooks/{webhook_id}") + + async def get_webhook( + self, webhook_id: "Snowflake", token: Optional[str] = None + ) -> "Webhook": + """Fetches a webhook by ID, optionally using its token.""" + + endpoint = f"/webhooks/{webhook_id}" + use_auth = True + if token is not None: + endpoint += f"/{token}" + use_auth = False + data = await self.request("GET", endpoint, use_auth_header=use_auth) + from .models import Webhook + + return Webhook(data) async def execute_webhook( self, diff --git a/disagreement/models.py b/disagreement/models.py index 833e06c..6c046e8 100644 --- a/disagreement/models.py +++ b/disagreement/models.py @@ -8,17 +8,17 @@ import json import os import re from dataclasses import dataclass -from typing import ( - Any, - AsyncIterator, - Dict, - List, - Optional, - TYPE_CHECKING, - Union, - cast, - IO, -) +from typing import ( + Any, + AsyncIterator, + Dict, + List, + Optional, + TYPE_CHECKING, + Union, + cast, + IO, +) from .cache import ChannelCache, MemberCache from .caching import MemberCacheFlags @@ -49,10 +49,10 @@ from .permissions import Permissions if TYPE_CHECKING: from .client import Client # For type hinting to avoid circular imports from .enums import OverwriteType # For PermissionOverwrite model - from .ui.view import View - from .interactions import Snowflake - from .typing import Typing - from .shard_manager import Shard + from .ui.view import View + from .interactions import Snowflake + from .typing import Typing + from .shard_manager import Shard # Forward reference Message if it were used in type hints before its definition # from .models import Message # Not needed as Message is defined before its use in TextChannel.send etc. @@ -1129,12 +1129,12 @@ class Guild: max_video_channel_users (Optional[int]): The maximum number of users in a video channel. welcome_screen (Optional[Dict]): The welcome screen of a Community guild. (Consider a WelcomeScreen model) nsfw_level (GuildNSFWLevel): Guild NSFW level. - stickers (Optional[List[Dict]]): Custom stickers in the guild. (Consider a Sticker model) - premium_progress_bar_enabled (bool): Whether the guild has the premium progress bar enabled. - text_channels (List[TextChannel]): List of text-based channels in this guild. - voice_channels (List[VoiceChannel]): List of voice-based channels in this guild. - category_channels (List[CategoryChannel]): List of category channels in this guild. - """ + stickers (Optional[List[Dict]]): Custom stickers in the guild. (Consider a Sticker model) + premium_progress_bar_enabled (bool): Whether the guild has the premium progress bar enabled. + text_channels (List[TextChannel]): List of text-based channels in this guild. + voice_channels (List[VoiceChannel]): List of voice-based channels in this guild. + category_channels (List[CategoryChannel]): List of category channels in this guild. + """ def __init__( self, @@ -1222,14 +1222,14 @@ class Guild: ) # Internal caches, populated by events or specific fetches - self._channels: ChannelCache = ChannelCache() - self._members: MemberCache = MemberCache( - getattr(client_instance, "member_cache_flags", MemberCacheFlags()) - ) - self._threads: Dict[str, "Thread"] = {} - self.text_channels: List["TextChannel"] = [] - self.voice_channels: List["VoiceChannel"] = [] - self.category_channels: List["CategoryChannel"] = [] + self._channels: ChannelCache = ChannelCache() + self._members: MemberCache = MemberCache( + getattr(client_instance, "member_cache_flags", MemberCacheFlags()) + ) + self._threads: Dict[str, "Thread"] = {} + self.text_channels: List["TextChannel"] = [] + self.voice_channels: List["VoiceChannel"] = [] + self.category_channels: List["CategoryChannel"] = [] @property def shard_id(self) -> Optional[int]: @@ -1253,10 +1253,10 @@ class Guild: def get_channel(self, channel_id: str) -> Optional["Channel"]: return self._channels.get(channel_id) - def get_member(self, user_id: str) -> Optional[Member]: - return self._members.get(user_id) - - def get_member_named(self, name: str) -> Optional[Member]: + def get_member(self, user_id: str) -> Optional[Member]: + return self._members.get(user_id) + + def get_member_named(self, name: str) -> Optional[Member]: """Retrieve a cached member by username or nickname. The lookup is case-insensitive and searches both the username and @@ -1282,16 +1282,16 @@ class Guild: return None def get_role(self, role_id: str) -> Optional[Role]: - return next((role for role in self.roles if role.id == role_id), None) - - @property - def me(self) -> Optional[Member]: - """The member object for the connected bot in this guild, if present.""" - - client_user = getattr(self._client, "user", None) - if not client_user: - return None - return self.get_member(client_user.id) + return next((role for role in self.roles if role.id == role_id), None) + + @property + def me(self) -> Optional[Member]: + """The member object for the connected bot in this guild, if present.""" + + client_user = getattr(self._client, "user", None) + if not client_user: + return None + return self.get_member(client_user.id) def __repr__(self) -> str: return f"" @@ -1965,6 +1965,33 @@ class Webhook: return cls({"id": webhook_id, "token": token, "url": url}) + @classmethod + def from_token( + cls, + webhook_id: str, + token: str, + session: Optional[aiohttp.ClientSession] = None, + ) -> "Webhook": + """Create a minimal :class:`Webhook` from an ID and token. + + Parameters + ---------- + webhook_id: + The ID of the webhook. + token: + The webhook token. + session: + Unused for now. Present for API compatibility. + + Returns + ------- + Webhook + A webhook instance containing only the ``id``, ``token`` and ``url``. + """ + + url = f"https://discord.com/api/webhooks/{webhook_id}/{token}" + return cls({"id": webhook_id, "token": token, "url": url}) + async def send( self, content: Optional[str] = None, diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py index fa82803..ec41110 100644 --- a/tests/test_webhooks.py +++ b/tests/test_webhooks.py @@ -146,6 +146,16 @@ def test_webhook_from_url_parses_id_and_token(): assert webhook.url == url +def test_webhook_from_token_builds_url_and_fields(): + from disagreement.models import Webhook + + webhook = Webhook.from_token("123", "token") + + assert webhook.id == "123" + assert webhook.token == "token" + assert webhook.url == "https://discord.com/api/webhooks/123/token" + + @pytest.mark.asyncio async def test_execute_webhook_calls_request(): http = HTTPClient(token="t")