Add Webhook.from_token and fetch support (#110)
This commit is contained in:
parent
223c86cb78
commit
9fabf1fbac
@ -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,
|
||||
|
@ -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"<Guild id='{self.id}' name='{self.name}'>"
|
||||
@ -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,
|
||||
|
@ -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")
|
||||
|
Loading…
x
Reference in New Issue
Block a user