2698 lines
88 KiB
Python
2698 lines
88 KiB
Python
# SPDX-License-Identifier: MIT
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
import sys
|
|
import weakref
|
|
from errno import ECONNRESET
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
Any,
|
|
ClassVar,
|
|
Coroutine,
|
|
Dict,
|
|
Iterable,
|
|
List,
|
|
Literal,
|
|
Optional,
|
|
Sequence,
|
|
Tuple,
|
|
Type,
|
|
TypeVar,
|
|
Union,
|
|
)
|
|
from urllib.parse import quote as _uriquote
|
|
|
|
import aiohttp
|
|
import yarl
|
|
|
|
from . import __version__, utils
|
|
from .errors import (
|
|
DiscordServerError,
|
|
Forbidden,
|
|
GatewayNotFound,
|
|
HTTPException,
|
|
LoginFailure,
|
|
NotFound,
|
|
)
|
|
from .gateway import DiscordClientWebSocketResponse
|
|
from .utils import MISSING
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
if TYPE_CHECKING:
|
|
from types import TracebackType
|
|
|
|
from typing_extensions import Self
|
|
|
|
from .enums import InteractionResponseType
|
|
from .file import File
|
|
from .message import Attachment
|
|
from .types import (
|
|
appinfo,
|
|
application_role_connection,
|
|
audit_log,
|
|
automod,
|
|
channel,
|
|
components,
|
|
embed,
|
|
emoji,
|
|
gateway,
|
|
guild,
|
|
guild_scheduled_event,
|
|
integration,
|
|
interactions,
|
|
invite,
|
|
member,
|
|
message,
|
|
onboarding,
|
|
role,
|
|
sticker,
|
|
template,
|
|
threads,
|
|
user,
|
|
voice,
|
|
webhook,
|
|
welcome_screen,
|
|
widget,
|
|
)
|
|
from .types.snowflake import Snowflake, SnowflakeList
|
|
|
|
T = TypeVar("T")
|
|
Response = Coroutine[Any, Any, T]
|
|
|
|
_API_VERSION = 10
|
|
|
|
|
|
def _workaround_set_api_version(version: Literal[9, 10]) -> None:
|
|
"""Stopgap measure for verified bots without message content intent while intent is not enforced on api v9.
|
|
|
|
.. note::
|
|
This must be ran **before** connecting to the gateway.
|
|
"""
|
|
if version not in (9, 10):
|
|
raise TypeError("version must be either 9 or 10")
|
|
|
|
global _API_VERSION # noqa: PLW0603
|
|
_API_VERSION = version
|
|
Route.BASE = f"https://discord.com/api/v{_API_VERSION}"
|
|
|
|
|
|
async def json_or_text(response: aiohttp.ClientResponse) -> Union[Dict[str, Any], str]:
|
|
text = await response.text(encoding="utf-8")
|
|
try:
|
|
if response.headers["content-type"] == "application/json":
|
|
return utils._from_json(text)
|
|
except KeyError:
|
|
# Thanks Cloudflare
|
|
pass
|
|
|
|
return text
|
|
|
|
|
|
def set_attachments(payload: Dict[str, Any], files: Sequence[File]) -> None:
|
|
"""Updates the payload's attachments list based on the provided files
|
|
|
|
note: this method modifies the provided ``payload`` and ``payload["attachments"]`` collections
|
|
"""
|
|
attachments = payload.get("attachments", [])
|
|
for index, file in enumerate(files):
|
|
attachments.append(
|
|
{
|
|
"id": index,
|
|
"description": file.description,
|
|
}
|
|
)
|
|
|
|
# if existing attachments weren't in the payload before and we
|
|
# didn't add any new ones, don't add the list to the payload.
|
|
if attachments:
|
|
payload["attachments"] = attachments
|
|
|
|
|
|
def to_multipart(payload: Dict[str, Any], files: Sequence[File]) -> List[Dict[str, Any]]:
|
|
"""Converts the payload and list of files to a multipart payload,
|
|
as specified by https://discord.com/developers/docs/reference#uploading-files
|
|
"""
|
|
multipart: List[Dict[str, Any]] = []
|
|
for index, file in enumerate(files):
|
|
multipart.append(
|
|
{
|
|
"name": f"files[{index}]",
|
|
"value": file.fp,
|
|
"filename": file.filename,
|
|
"content_type": "application/octet-stream",
|
|
}
|
|
)
|
|
|
|
multipart.append({"name": "payload_json", "value": utils._to_json(payload)})
|
|
return multipart
|
|
|
|
|
|
def to_multipart_with_attachments(
|
|
payload: Dict[str, Any], files: Sequence[File]
|
|
) -> List[Dict[str, Any]]:
|
|
"""Updates the payload's attachments and converts it to a multipart payload
|
|
|
|
Shorthand for ``set_attachments`` + ``to_multipart``
|
|
"""
|
|
set_attachments(payload, files)
|
|
return to_multipart(payload, files)
|
|
|
|
|
|
class Route:
|
|
BASE: ClassVar[str] = "https://discord.com/api/v10"
|
|
|
|
def __init__(self, method: str, path: str, **parameters: Any) -> None:
|
|
self.path: str = path
|
|
self.method: str = method
|
|
url = self.BASE + self.path
|
|
if parameters:
|
|
url = url.format_map(
|
|
{k: _uriquote(v) if isinstance(v, str) else v for k, v in parameters.items()}
|
|
)
|
|
self.url: str = url
|
|
|
|
# major parameters:
|
|
self.channel_id: Optional[Snowflake] = parameters.get("channel_id")
|
|
self.guild_id: Optional[Snowflake] = parameters.get("guild_id")
|
|
self.webhook_id: Optional[Snowflake] = parameters.get("webhook_id")
|
|
self.webhook_token: Optional[str] = parameters.get("webhook_token")
|
|
|
|
@property
|
|
def bucket(self) -> str:
|
|
# the bucket is just method + path w/ major parameters
|
|
return f"{self.channel_id}:{self.guild_id}:{self.path}"
|
|
|
|
|
|
class MaybeUnlock:
|
|
def __init__(self, lock: asyncio.Lock) -> None:
|
|
self.lock: asyncio.Lock = lock
|
|
self._unlock: bool = True
|
|
|
|
def __enter__(self) -> Self:
|
|
return self
|
|
|
|
def defer(self) -> None:
|
|
self._unlock = False
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: Optional[Type[BaseException]],
|
|
exc: Optional[BaseException],
|
|
traceback: Optional[TracebackType],
|
|
) -> None:
|
|
if self._unlock:
|
|
self.lock.release()
|
|
|
|
|
|
# For some reason, the Discord voice websocket expects this header to be
|
|
# completely lowercase while aiohttp respects spec and does it as case-insensitive
|
|
aiohttp.hdrs.WEBSOCKET = "websocket" # type: ignore
|
|
|
|
|
|
class HTTPClient:
|
|
"""Represents an HTTP client sending HTTP requests to the Discord API."""
|
|
|
|
def __init__(
|
|
self,
|
|
connector: Optional[aiohttp.BaseConnector] = None,
|
|
*,
|
|
loop: asyncio.AbstractEventLoop,
|
|
proxy: Optional[str] = None,
|
|
proxy_auth: Optional[aiohttp.BasicAuth] = None,
|
|
unsync_clock: bool = True,
|
|
) -> None:
|
|
self.loop: asyncio.AbstractEventLoop = loop
|
|
self.connector = connector
|
|
self.__session: aiohttp.ClientSession = MISSING # filled in static_login
|
|
self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = weakref.WeakValueDictionary()
|
|
self._global_over: asyncio.Event = asyncio.Event()
|
|
self._global_over.set()
|
|
self.token: Optional[str] = None
|
|
self.bot_token: bool = False
|
|
self.proxy: Optional[str] = proxy
|
|
self.proxy_auth: Optional[aiohttp.BasicAuth] = proxy_auth
|
|
self.use_clock: bool = not unsync_clock
|
|
|
|
user_agent = "DiscordBot (https://github.com/DisnakeDev/disnake {0}) Python/{1[0]}.{1[1]} aiohttp/{2}"
|
|
self.user_agent: str = user_agent.format(__version__, sys.version_info, aiohttp.__version__)
|
|
|
|
def recreate(self) -> None:
|
|
if self.__session.closed:
|
|
self.__session = aiohttp.ClientSession(
|
|
connector=self.connector, ws_response_class=DiscordClientWebSocketResponse
|
|
)
|
|
|
|
async def ws_connect(self, url: str, *, compress: int = 0) -> aiohttp.ClientWebSocketResponse:
|
|
kwargs = {
|
|
"proxy_auth": self.proxy_auth,
|
|
"proxy": self.proxy,
|
|
"max_msg_size": 0,
|
|
"timeout": 30.0,
|
|
"autoclose": False,
|
|
"headers": {
|
|
"User-Agent": self.user_agent,
|
|
},
|
|
"compress": compress,
|
|
}
|
|
|
|
return await self.__session.ws_connect(url, **kwargs)
|
|
|
|
async def request(
|
|
self,
|
|
route: Route,
|
|
*,
|
|
files: Optional[Sequence[File]] = None,
|
|
form: Optional[Iterable[Dict[str, Any]]] = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
bucket = route.bucket
|
|
method = route.method
|
|
url = route.url
|
|
|
|
lock = self._locks.get(bucket)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
if bucket is not None:
|
|
self._locks[bucket] = lock
|
|
|
|
# header creation
|
|
headers: Dict[str, str] = {
|
|
"User-Agent": self.user_agent,
|
|
}
|
|
|
|
if self.token is not None:
|
|
headers["Authorization"] = "Bot " + self.token
|
|
# some checking if it's a JSON request
|
|
if "json" in kwargs:
|
|
headers["Content-Type"] = "application/json"
|
|
kwargs["data"] = utils._to_json(kwargs.pop("json"))
|
|
|
|
try:
|
|
reason = kwargs.pop("reason")
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
if reason:
|
|
headers["X-Audit-Log-Reason"] = _uriquote(reason, safe="/ ")
|
|
|
|
kwargs["headers"] = headers
|
|
|
|
# Proxy support
|
|
if self.proxy is not None:
|
|
kwargs["proxy"] = self.proxy
|
|
if self.proxy_auth is not None:
|
|
kwargs["proxy_auth"] = self.proxy_auth
|
|
|
|
if not self._global_over.is_set():
|
|
# wait until the global lock is complete
|
|
await self._global_over.wait()
|
|
|
|
response: Optional[aiohttp.ClientResponse] = None
|
|
data: Optional[Union[Dict[str, Any], str]] = None
|
|
await lock.acquire()
|
|
with MaybeUnlock(lock) as maybe_lock:
|
|
for tries in range(5):
|
|
if files:
|
|
for f in files:
|
|
f.reset(seek=tries)
|
|
|
|
if form:
|
|
# NOTE: for `quote_fields`, see https://github.com/aio-libs/aiohttp/issues/4012
|
|
form_data = aiohttp.FormData(quote_fields=False)
|
|
for p in form:
|
|
# manually escape chars, just in case
|
|
name = re.sub(
|
|
r"[^\x21\x23-\x5b\x5d-\x7e]", lambda m: f"\\{m.group(0)}", p["name"]
|
|
)
|
|
form_data.add_field(
|
|
name=name, **{k: v for k, v in p.items() if k != "name"}
|
|
)
|
|
kwargs["data"] = form_data
|
|
|
|
try:
|
|
async with self.__session.request(method, url, **kwargs) as response:
|
|
_log.debug(
|
|
"%s %s with %s has returned %s",
|
|
method,
|
|
url,
|
|
kwargs.get("data"),
|
|
response.status,
|
|
)
|
|
|
|
# even errors have text involved in them so this is safe to call
|
|
data = await json_or_text(response)
|
|
|
|
# check if we have rate limit header information
|
|
remaining = response.headers.get("X-Ratelimit-Remaining")
|
|
if remaining == "0" and response.status != 429:
|
|
# we've depleted our current bucket
|
|
delta = utils._parse_ratelimit_header(
|
|
response, use_clock=self.use_clock
|
|
)
|
|
_log.debug(
|
|
"A rate limit bucket has been exhausted (bucket: %s, retry: %s).",
|
|
bucket,
|
|
delta,
|
|
)
|
|
maybe_lock.defer()
|
|
self.loop.call_later(delta, lock.release)
|
|
|
|
# the request was successful so just return the text/json
|
|
if 300 > response.status >= 200:
|
|
_log.debug("%s %s has received %s", method, url, data)
|
|
return data
|
|
|
|
# we are being rate limited
|
|
if response.status == 429:
|
|
if not response.headers.get("Via") or isinstance(data, str):
|
|
# Banned by Cloudflare more than likely.
|
|
raise HTTPException(response, data)
|
|
|
|
fmt = 'We are being rate limited. Retrying in %.2f seconds. Handled under the bucket "%s"'
|
|
|
|
# sleep a bit
|
|
retry_after: float = data["retry_after"]
|
|
_log.warning(fmt, retry_after, bucket)
|
|
|
|
# check if it's a global rate limit
|
|
is_global = data.get("global", False)
|
|
if is_global:
|
|
_log.warning(
|
|
"Global rate limit has been hit. Retrying in %.2f seconds.",
|
|
retry_after,
|
|
)
|
|
self._global_over.clear()
|
|
|
|
await asyncio.sleep(retry_after)
|
|
_log.debug("Done sleeping for the rate limit. Retrying...")
|
|
|
|
# release the global lock now that the
|
|
# global rate limit has passed
|
|
if is_global:
|
|
self._global_over.set()
|
|
_log.debug("Global rate limit is now over.")
|
|
|
|
continue
|
|
|
|
# we've received a 500, 502, or 504, unconditional retry
|
|
if response.status in {500, 502, 504}:
|
|
await asyncio.sleep(1 + tries * 2)
|
|
continue
|
|
|
|
# the usual error cases
|
|
if response.status == 403:
|
|
raise Forbidden(response, data)
|
|
elif response.status == 404:
|
|
raise NotFound(response, data)
|
|
elif response.status >= 500:
|
|
raise DiscordServerError(response, data)
|
|
else:
|
|
raise HTTPException(response, data)
|
|
|
|
# This is handling exceptions from the request
|
|
except OSError as e:
|
|
# Connection reset by peer
|
|
if tries < 4 and e.errno == ECONNRESET:
|
|
await asyncio.sleep(1 + tries * 2)
|
|
continue
|
|
raise
|
|
|
|
if response is not None:
|
|
# We've run out of retries, raise.
|
|
if response.status >= 500:
|
|
raise DiscordServerError(response, data)
|
|
|
|
raise HTTPException(response, data)
|
|
|
|
raise RuntimeError("Unreachable code in HTTP handling")
|
|
|
|
async def get_from_cdn(self, url: str) -> bytes:
|
|
async with self.__session.get(url) as resp:
|
|
if resp.status == 200:
|
|
return await resp.read()
|
|
elif resp.status == 404:
|
|
raise NotFound(resp, "asset not found")
|
|
elif resp.status == 403:
|
|
raise Forbidden(resp, "cannot retrieve asset")
|
|
else:
|
|
raise HTTPException(resp, "failed to get asset")
|
|
|
|
# state management
|
|
|
|
async def close(self) -> None:
|
|
if self.__session:
|
|
await self.__session.close()
|
|
|
|
# login management
|
|
|
|
async def static_login(self, token: str) -> user.User:
|
|
# Necessary to get aiohttp to stop complaining about session creation
|
|
self.__session = aiohttp.ClientSession(
|
|
connector=self.connector, ws_response_class=DiscordClientWebSocketResponse
|
|
)
|
|
old_token = self.token
|
|
self.token = token
|
|
|
|
try:
|
|
data: user.User = await self.request(Route("GET", "/users/@me"))
|
|
except HTTPException as exc:
|
|
self.token = old_token
|
|
if exc.status == 401:
|
|
raise LoginFailure("Improper token has been passed.") from exc
|
|
raise
|
|
|
|
return data
|
|
|
|
def create_party(
|
|
self,
|
|
channel_id: Snowflake,
|
|
max_age: int,
|
|
max_uses: int,
|
|
target_application_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[invite.Invite]:
|
|
payload = {
|
|
"max_age": max_age,
|
|
"max_uses": max_uses,
|
|
"target_application_id": target_application_id,
|
|
"target_type": 2,
|
|
"temporary": False,
|
|
"validate": None,
|
|
}
|
|
return self.request(
|
|
Route("POST", "/channels/{channel_id}/invites", channel_id=channel_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
# Group functionality
|
|
|
|
def start_group(
|
|
self, user_id: Snowflake, recipients: List[int]
|
|
) -> Response[channel.GroupDMChannel]:
|
|
payload = {
|
|
"recipients": recipients,
|
|
}
|
|
|
|
return self.request(
|
|
Route("POST", "/users/{user_id}/channels", user_id=user_id), json=payload
|
|
)
|
|
|
|
def leave_group(self, channel_id: Snowflake) -> Response[None]:
|
|
return self.request(Route("DELETE", "/channels/{channel_id}", channel_id=channel_id))
|
|
|
|
# Message management
|
|
|
|
def start_private_message(self, user_id: Snowflake) -> Response[channel.DMChannel]:
|
|
payload = {
|
|
"recipient_id": user_id,
|
|
}
|
|
|
|
return self.request(Route("POST", "/users/@me/channels"), json=payload)
|
|
|
|
def send_message(
|
|
self,
|
|
channel_id: Snowflake,
|
|
content: Optional[str],
|
|
*,
|
|
tts: bool = False,
|
|
embed: Optional[embed.Embed] = None,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
nonce: Optional[Union[str, int]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
message_reference: Optional[message.MessageReference] = None,
|
|
stickers: Optional[Sequence[Snowflake]] = None,
|
|
components: Optional[Sequence[components.Component]] = None,
|
|
flags: Optional[int] = None,
|
|
) -> Response[message.Message]:
|
|
r = Route("POST", "/channels/{channel_id}/messages", channel_id=channel_id)
|
|
payload: Dict[str, Any] = {}
|
|
|
|
if content:
|
|
payload["content"] = content
|
|
|
|
if tts:
|
|
payload["tts"] = True
|
|
|
|
if embed:
|
|
payload["embeds"] = [embed]
|
|
|
|
if embeds:
|
|
payload["embeds"] = embeds
|
|
|
|
if nonce:
|
|
payload["nonce"] = nonce
|
|
|
|
if allowed_mentions:
|
|
payload["allowed_mentions"] = allowed_mentions
|
|
|
|
if message_reference:
|
|
payload["message_reference"] = message_reference
|
|
|
|
if components:
|
|
payload["components"] = components
|
|
|
|
if stickers:
|
|
payload["sticker_ids"] = stickers
|
|
|
|
if flags is not None:
|
|
payload["flags"] = flags
|
|
|
|
return self.request(r, json=payload)
|
|
|
|
def send_typing(self, channel_id: Snowflake) -> Response[None]:
|
|
return self.request(Route("POST", "/channels/{channel_id}/typing", channel_id=channel_id))
|
|
|
|
def send_multipart_helper(
|
|
self,
|
|
route: Route,
|
|
*,
|
|
files: Sequence[File],
|
|
content: Optional[str] = None,
|
|
tts: bool = False,
|
|
embed: Optional[embed.Embed] = None,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
nonce: Optional[Union[str, int]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
message_reference: Optional[message.MessageReference] = None,
|
|
stickers: Optional[Sequence[Snowflake]] = None,
|
|
components: Optional[Sequence[components.Component]] = None,
|
|
flags: Optional[int] = None,
|
|
) -> Response[message.Message]:
|
|
payload: Dict[str, Any] = {"tts": tts}
|
|
if content:
|
|
payload["content"] = content
|
|
if embed:
|
|
payload["embeds"] = [embed]
|
|
if embeds:
|
|
payload["embeds"] = embeds
|
|
if nonce:
|
|
payload["nonce"] = nonce
|
|
if allowed_mentions:
|
|
payload["allowed_mentions"] = allowed_mentions
|
|
if message_reference:
|
|
payload["message_reference"] = message_reference
|
|
if components:
|
|
payload["components"] = components
|
|
if stickers:
|
|
payload["sticker_ids"] = stickers
|
|
if flags is not None:
|
|
payload["flags"] = flags
|
|
|
|
multipart = to_multipart_with_attachments(payload, files)
|
|
|
|
return self.request(route, form=multipart, files=files)
|
|
|
|
def send_files(
|
|
self,
|
|
channel_id: Snowflake,
|
|
*,
|
|
files: Sequence[File],
|
|
content: Optional[str] = None,
|
|
tts: bool = False,
|
|
embed: Optional[embed.Embed] = None,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
nonce: Optional[Union[str, int]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
message_reference: Optional[message.MessageReference] = None,
|
|
stickers: Optional[Sequence[Snowflake]] = None,
|
|
components: Optional[Sequence[components.Component]] = None,
|
|
flags: Optional[int] = None,
|
|
) -> Response[message.Message]:
|
|
r = Route("POST", "/channels/{channel_id}/messages", channel_id=channel_id)
|
|
return self.send_multipart_helper(
|
|
r,
|
|
files=files,
|
|
content=content,
|
|
tts=tts,
|
|
embed=embed,
|
|
embeds=embeds,
|
|
nonce=nonce,
|
|
allowed_mentions=allowed_mentions,
|
|
message_reference=message_reference,
|
|
stickers=stickers,
|
|
components=components,
|
|
flags=flags,
|
|
)
|
|
|
|
def delete_message(
|
|
self, channel_id: Snowflake, message_id: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/messages/{message_id}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def delete_messages(
|
|
self, channel_id: Snowflake, message_ids: SnowflakeList, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route("POST", "/channels/{channel_id}/messages/bulk-delete", channel_id=channel_id)
|
|
payload = {
|
|
"messages": message_ids,
|
|
}
|
|
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def edit_message(
|
|
self,
|
|
channel_id: Snowflake,
|
|
message_id: Snowflake,
|
|
*,
|
|
files: Optional[List[File]] = None,
|
|
**fields: Any,
|
|
) -> Response[message.Message]:
|
|
r = Route(
|
|
"PATCH",
|
|
"/channels/{channel_id}/messages/{message_id}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
if files:
|
|
multipart = to_multipart_with_attachments(fields, files)
|
|
return self.request(r, form=multipart, files=files)
|
|
return self.request(r, json=fields)
|
|
|
|
def add_reaction(
|
|
self, channel_id: Snowflake, message_id: Snowflake, emoji: str
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"PUT",
|
|
"/channels/{channel_id}/messages/{message_id}/reactions/{emoji}/@me",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
emoji=emoji,
|
|
)
|
|
return self.request(r)
|
|
|
|
def remove_reaction(
|
|
self, channel_id: Snowflake, message_id: Snowflake, emoji: str, member_id: Snowflake
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/messages/{message_id}/reactions/{emoji}/{member_id}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
member_id=member_id,
|
|
emoji=emoji,
|
|
)
|
|
return self.request(r)
|
|
|
|
def remove_own_reaction(
|
|
self, channel_id: Snowflake, message_id: Snowflake, emoji: str
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/messages/{message_id}/reactions/{emoji}/@me",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
emoji=emoji,
|
|
)
|
|
return self.request(r)
|
|
|
|
def get_reaction_users(
|
|
self,
|
|
channel_id: Snowflake,
|
|
message_id: Snowflake,
|
|
emoji: str,
|
|
limit: int,
|
|
after: Optional[Snowflake] = None,
|
|
) -> Response[List[user.User]]:
|
|
r = Route(
|
|
"GET",
|
|
"/channels/{channel_id}/messages/{message_id}/reactions/{emoji}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
emoji=emoji,
|
|
)
|
|
|
|
params: Dict[str, Any] = {
|
|
"limit": limit,
|
|
}
|
|
if after:
|
|
params["after"] = after
|
|
return self.request(r, params=params)
|
|
|
|
def clear_reactions(self, channel_id: Snowflake, message_id: Snowflake) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/messages/{message_id}/reactions",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
|
|
return self.request(r)
|
|
|
|
def clear_single_reaction(
|
|
self, channel_id: Snowflake, message_id: Snowflake, emoji: str
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/messages/{message_id}/reactions/{emoji}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
emoji=emoji,
|
|
)
|
|
return self.request(r)
|
|
|
|
def get_message(
|
|
self, channel_id: Snowflake, message_id: Snowflake
|
|
) -> Response[message.Message]:
|
|
r = Route(
|
|
"GET",
|
|
"/channels/{channel_id}/messages/{message_id}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def get_channel(self, channel_id: Snowflake) -> Response[channel.Channel]:
|
|
r = Route("GET", "/channels/{channel_id}", channel_id=channel_id)
|
|
return self.request(r)
|
|
|
|
def logs_from(
|
|
self,
|
|
channel_id: Snowflake,
|
|
limit: int,
|
|
before: Optional[Snowflake] = None,
|
|
after: Optional[Snowflake] = None,
|
|
around: Optional[Snowflake] = None,
|
|
) -> Response[List[message.Message]]:
|
|
params: Dict[str, Any] = {
|
|
"limit": limit,
|
|
}
|
|
|
|
if before is not None:
|
|
params["before"] = before
|
|
if after is not None:
|
|
params["after"] = after
|
|
if around is not None:
|
|
params["around"] = around
|
|
|
|
return self.request(
|
|
Route("GET", "/channels/{channel_id}/messages", channel_id=channel_id), params=params
|
|
)
|
|
|
|
def publish_message(
|
|
self, channel_id: Snowflake, message_id: Snowflake
|
|
) -> Response[message.Message]:
|
|
return self.request(
|
|
Route(
|
|
"POST",
|
|
"/channels/{channel_id}/messages/{message_id}/crosspost",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
)
|
|
|
|
def pin_message(
|
|
self, channel_id: Snowflake, message_id: Snowflake, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"PUT",
|
|
"/channels/{channel_id}/pins/{message_id}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def unpin_message(
|
|
self, channel_id: Snowflake, message_id: Snowflake, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/pins/{message_id}",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def pins_from(self, channel_id: Snowflake) -> Response[List[message.Message]]:
|
|
return self.request(Route("GET", "/channels/{channel_id}/pins", channel_id=channel_id))
|
|
|
|
# Member management
|
|
|
|
def search_guild_members(
|
|
self, guild_id: Snowflake, query: str, limit: int = 1
|
|
) -> Response[List[member.MemberWithUser]]:
|
|
r = Route("GET", "/guilds/{guild_id}/members/search", guild_id=guild_id)
|
|
|
|
return self.request(r, params={"query": query, "limit": limit})
|
|
|
|
def kick(
|
|
self, user_id: Snowflake, guild_id: Snowflake, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE", "/guilds/{guild_id}/members/{user_id}", guild_id=guild_id, user_id=user_id
|
|
)
|
|
|
|
return self.request(r, reason=reason)
|
|
|
|
def ban(
|
|
self,
|
|
user_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
*,
|
|
delete_message_seconds: int = 86400,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
r = Route("PUT", "/guilds/{guild_id}/bans/{user_id}", guild_id=guild_id, user_id=user_id)
|
|
payload = {
|
|
"delete_message_seconds": delete_message_seconds,
|
|
}
|
|
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def unban(
|
|
self, user_id: Snowflake, guild_id: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route("DELETE", "/guilds/{guild_id}/bans/{user_id}", guild_id=guild_id, user_id=user_id)
|
|
return self.request(r, reason=reason)
|
|
|
|
def get_guild_voice_regions(self, guild_id: Snowflake) -> Response[List[voice.VoiceRegion]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/regions", guild_id=guild_id))
|
|
|
|
def guild_voice_state(
|
|
self,
|
|
user_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
*,
|
|
mute: Optional[bool] = None,
|
|
deafen: Optional[bool] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[member.Member]:
|
|
r = Route(
|
|
"PATCH", "/guilds/{guild_id}/members/{user_id}", guild_id=guild_id, user_id=user_id
|
|
)
|
|
payload = {}
|
|
if mute is not None:
|
|
payload["mute"] = mute
|
|
|
|
if deafen is not None:
|
|
payload["deaf"] = deafen
|
|
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def edit_profile(self, payload: Dict[str, Any]) -> Response[user.User]:
|
|
return self.request(Route("PATCH", "/users/@me"), json=payload)
|
|
|
|
def change_nickname(
|
|
self,
|
|
guild_id: Snowflake,
|
|
user_id: Snowflake,
|
|
nickname: str,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[member.Member]:
|
|
r = Route(
|
|
"PATCH", "/guilds/{guild_id}/members/{user_id}", guild_id=guild_id, user_id=user_id
|
|
)
|
|
payload = {
|
|
"nick": nickname,
|
|
}
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def edit_my_voice_state(self, guild_id: Snowflake, payload: Dict[str, Any]) -> Response[None]:
|
|
r = Route("PATCH", "/guilds/{guild_id}/voice-states/@me", guild_id=guild_id)
|
|
return self.request(r, json=payload)
|
|
|
|
def edit_voice_state(
|
|
self, guild_id: Snowflake, user_id: Snowflake, payload: Dict[str, Any]
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"PATCH", "/guilds/{guild_id}/voice-states/{user_id}", guild_id=guild_id, user_id=user_id
|
|
)
|
|
return self.request(r, json=payload)
|
|
|
|
def edit_my_member(
|
|
self,
|
|
guild_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**fields: Any,
|
|
) -> Response[member.MemberWithUser]:
|
|
r = Route("PATCH", "/guilds/{guild_id}/members/@me", guild_id=guild_id)
|
|
return self.request(r, json=fields, reason=reason)
|
|
|
|
def edit_member(
|
|
self,
|
|
guild_id: Snowflake,
|
|
user_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**fields: Any,
|
|
) -> Response[member.MemberWithUser]:
|
|
r = Route(
|
|
"PATCH", "/guilds/{guild_id}/members/{user_id}", guild_id=guild_id, user_id=user_id
|
|
)
|
|
return self.request(r, json=fields, reason=reason)
|
|
|
|
# Channel management
|
|
|
|
def edit_channel(
|
|
self,
|
|
channel_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**options: Any,
|
|
) -> Response[channel.Channel]:
|
|
r = Route("PATCH", "/channels/{channel_id}", channel_id=channel_id)
|
|
valid_keys = (
|
|
"name",
|
|
"parent_id",
|
|
"topic",
|
|
"bitrate",
|
|
"nsfw",
|
|
"user_limit",
|
|
"position",
|
|
"permission_overwrites",
|
|
"rate_limit_per_user",
|
|
"default_thread_rate_limit_per_user",
|
|
"type",
|
|
"rtc_region",
|
|
"video_quality_mode",
|
|
"archived",
|
|
"auto_archive_duration",
|
|
"locked",
|
|
"invitable",
|
|
"default_auto_archive_duration",
|
|
"flags",
|
|
"available_tags",
|
|
"applied_tags",
|
|
"default_reaction_emoji",
|
|
"default_sort_order",
|
|
"default_forum_layout",
|
|
)
|
|
payload = {k: v for k, v in options.items() if k in valid_keys}
|
|
return self.request(r, reason=reason, json=payload)
|
|
|
|
def bulk_channel_update(
|
|
self,
|
|
guild_id: Snowflake,
|
|
data: List[guild.ChannelPositionUpdate],
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
r = Route("PATCH", "/guilds/{guild_id}/channels", guild_id=guild_id)
|
|
return self.request(r, json=data, reason=reason)
|
|
|
|
def create_channel(
|
|
self,
|
|
guild_id: Snowflake,
|
|
channel_type: channel.ChannelType,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**options: Any,
|
|
) -> Response[channel.GuildChannel]:
|
|
payload = {
|
|
"type": channel_type,
|
|
}
|
|
|
|
valid_keys = (
|
|
"name",
|
|
"parent_id",
|
|
"topic",
|
|
"bitrate",
|
|
"nsfw",
|
|
"user_limit",
|
|
"position",
|
|
"permission_overwrites",
|
|
"rate_limit_per_user",
|
|
"default_thread_rate_limit_per_user",
|
|
"rtc_region",
|
|
"video_quality_mode",
|
|
"auto_archive_duration",
|
|
"default_auto_archive_duration",
|
|
"flags",
|
|
"available_tags",
|
|
"default_reaction_emoji",
|
|
"default_sort_order",
|
|
)
|
|
payload.update({k: v for k, v in options.items() if k in valid_keys and v is not None})
|
|
|
|
return self.request(
|
|
Route("POST", "/guilds/{guild_id}/channels", guild_id=guild_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def delete_channel(
|
|
self,
|
|
channel_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
return self.request(
|
|
Route("DELETE", "/channels/{channel_id}", channel_id=channel_id), reason=reason
|
|
)
|
|
|
|
# Thread management
|
|
|
|
def start_thread_with_message(
|
|
self,
|
|
channel_id: Snowflake,
|
|
message_id: Snowflake,
|
|
*,
|
|
name: str,
|
|
auto_archive_duration: threads.ThreadArchiveDurationLiteral,
|
|
rate_limit_per_user: Optional[int] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[threads.Thread]:
|
|
payload = {
|
|
"name": name,
|
|
"auto_archive_duration": auto_archive_duration,
|
|
}
|
|
|
|
if rate_limit_per_user is not None:
|
|
payload["rate_limit_per_user"] = rate_limit_per_user
|
|
|
|
route = Route(
|
|
"POST",
|
|
"/channels/{channel_id}/messages/{message_id}/threads",
|
|
channel_id=channel_id,
|
|
message_id=message_id,
|
|
)
|
|
return self.request(route, json=payload, reason=reason)
|
|
|
|
def start_thread_without_message(
|
|
self,
|
|
channel_id: Snowflake,
|
|
*,
|
|
name: str,
|
|
auto_archive_duration: threads.ThreadArchiveDurationLiteral,
|
|
type: threads.ThreadType,
|
|
invitable: bool = True,
|
|
rate_limit_per_user: Optional[int] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[threads.Thread]:
|
|
payload = {
|
|
"name": name,
|
|
"auto_archive_duration": auto_archive_duration,
|
|
"type": type,
|
|
"invitable": invitable,
|
|
}
|
|
|
|
if rate_limit_per_user is not None:
|
|
payload["rate_limit_per_user"] = rate_limit_per_user
|
|
|
|
route = Route("POST", "/channels/{channel_id}/threads", channel_id=channel_id)
|
|
return self.request(route, json=payload, reason=reason)
|
|
|
|
def join_thread(self, channel_id: Snowflake) -> Response[None]:
|
|
return self.request(
|
|
Route("PUT", "/channels/{channel_id}/thread-members/@me", channel_id=channel_id)
|
|
)
|
|
|
|
def add_user_to_thread(self, channel_id: Snowflake, user_id: Snowflake) -> Response[None]:
|
|
return self.request(
|
|
Route(
|
|
"PUT",
|
|
"/channels/{channel_id}/thread-members/{user_id}",
|
|
channel_id=channel_id,
|
|
user_id=user_id,
|
|
)
|
|
)
|
|
|
|
def leave_thread(self, channel_id: Snowflake) -> Response[None]:
|
|
return self.request(
|
|
Route("DELETE", "/channels/{channel_id}/thread-members/@me", channel_id=channel_id)
|
|
)
|
|
|
|
def remove_user_from_thread(self, channel_id: Snowflake, user_id: Snowflake) -> Response[None]:
|
|
route = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/thread-members/{user_id}",
|
|
channel_id=channel_id,
|
|
user_id=user_id,
|
|
)
|
|
return self.request(route)
|
|
|
|
def get_public_archived_threads(
|
|
self, channel_id: Snowflake, before: Optional[Snowflake] = None, limit: int = 50
|
|
) -> Response[threads.ThreadPaginationPayload]:
|
|
route = Route(
|
|
"GET", "/channels/{channel_id}/threads/archived/public", channel_id=channel_id
|
|
)
|
|
|
|
params = {}
|
|
if before:
|
|
params["before"] = before
|
|
params["limit"] = limit
|
|
return self.request(route, params=params)
|
|
|
|
def get_private_archived_threads(
|
|
self, channel_id: Snowflake, before: Optional[Snowflake] = None, limit: int = 50
|
|
) -> Response[threads.ThreadPaginationPayload]:
|
|
route = Route(
|
|
"GET", "/channels/{channel_id}/threads/archived/private", channel_id=channel_id
|
|
)
|
|
|
|
params = {}
|
|
if before:
|
|
params["before"] = before
|
|
params["limit"] = limit
|
|
return self.request(route, params=params)
|
|
|
|
def get_joined_private_archived_threads(
|
|
self, channel_id: Snowflake, before: Optional[Snowflake] = None, limit: int = 50
|
|
) -> Response[threads.ThreadPaginationPayload]:
|
|
route = Route(
|
|
"GET",
|
|
"/channels/{channel_id}/users/@me/threads/archived/private",
|
|
channel_id=channel_id,
|
|
)
|
|
params = {}
|
|
if before:
|
|
params["before"] = before
|
|
params["limit"] = limit
|
|
return self.request(route, params=params)
|
|
|
|
def get_active_threads(self, guild_id: Snowflake) -> Response[threads.ThreadPaginationPayload]:
|
|
route = Route("GET", "/guilds/{guild_id}/threads/active", guild_id=guild_id)
|
|
return self.request(route)
|
|
|
|
def get_thread_member(
|
|
self, channel_id: Snowflake, user_id: Snowflake
|
|
) -> Response[threads.ThreadMember]:
|
|
route = Route(
|
|
"GET",
|
|
"/channels/{channel_id}/thread-members/{user_id}",
|
|
channel_id=channel_id,
|
|
user_id=user_id,
|
|
)
|
|
return self.request(route)
|
|
|
|
def get_thread_members(self, channel_id: Snowflake) -> Response[List[threads.ThreadMember]]:
|
|
route = Route("GET", "/channels/{channel_id}/thread-members", channel_id=channel_id)
|
|
return self.request(route)
|
|
|
|
def start_thread_in_forum_channel(
|
|
self,
|
|
channel_id: Snowflake,
|
|
files: Optional[Sequence[File]] = None,
|
|
reason: Optional[str] = None,
|
|
**fields: Any,
|
|
) -> Response[threads.ForumThread]:
|
|
valid_thread_keys = (
|
|
"name",
|
|
"auto_archive_duration",
|
|
"rate_limit_per_user",
|
|
"applied_tags",
|
|
"type",
|
|
)
|
|
valid_message_keys = (
|
|
"content",
|
|
"embeds",
|
|
"allowed_mentions",
|
|
"components",
|
|
"sticker_ids",
|
|
"flags",
|
|
)
|
|
payload = {k: v for k, v in fields.items() if k in valid_thread_keys}
|
|
payload["message"] = {k: v for k, v in fields.items() if k in valid_message_keys}
|
|
|
|
route = Route("POST", "/channels/{channel_id}/threads", channel_id=channel_id)
|
|
query_params = {"use_nested_fields": 1}
|
|
|
|
if files:
|
|
# This cannot directly call `to_multipart_with_attachments`,
|
|
# since attachment data needs to be added to the nested `message` object
|
|
set_attachments(payload["message"], files)
|
|
multipart = to_multipart(payload, files)
|
|
|
|
return self.request(
|
|
route, form=multipart, params=query_params, files=files, reason=reason
|
|
)
|
|
|
|
return self.request(route, json=payload, params=query_params, reason=reason)
|
|
|
|
# Webhook management
|
|
|
|
def create_webhook(
|
|
self,
|
|
channel_id: Snowflake,
|
|
*,
|
|
name: str,
|
|
avatar: Optional[str] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[webhook.Webhook]:
|
|
payload: Dict[str, Any] = {
|
|
"name": name,
|
|
}
|
|
if avatar is not None:
|
|
payload["avatar"] = avatar
|
|
|
|
r = Route("POST", "/channels/{channel_id}/webhooks", channel_id=channel_id)
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def channel_webhooks(self, channel_id: Snowflake) -> Response[List[webhook.Webhook]]:
|
|
return self.request(Route("GET", "/channels/{channel_id}/webhooks", channel_id=channel_id))
|
|
|
|
def guild_webhooks(self, guild_id: Snowflake) -> Response[List[webhook.Webhook]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/webhooks", guild_id=guild_id))
|
|
|
|
def get_webhook(self, webhook_id: Snowflake) -> Response[webhook.Webhook]:
|
|
return self.request(Route("GET", "/webhooks/{webhook_id}", webhook_id=webhook_id))
|
|
|
|
def follow_webhook(
|
|
self,
|
|
channel_id: Snowflake,
|
|
webhook_channel_id: Snowflake,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
payload = {
|
|
"webhook_channel_id": str(webhook_channel_id),
|
|
}
|
|
return self.request(
|
|
Route("POST", "/channels/{channel_id}/followers", channel_id=channel_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
# Guild management
|
|
|
|
def get_guilds(
|
|
self,
|
|
limit: int,
|
|
before: Optional[Snowflake] = None,
|
|
after: Optional[Snowflake] = None,
|
|
) -> Response[List[guild.Guild]]:
|
|
params: Dict[str, Any] = {
|
|
"limit": limit,
|
|
}
|
|
|
|
if before:
|
|
params["before"] = before
|
|
if after:
|
|
params["after"] = after
|
|
|
|
return self.request(Route("GET", "/users/@me/guilds"), params=params)
|
|
|
|
def leave_guild(self, guild_id: Snowflake) -> Response[None]:
|
|
return self.request(Route("DELETE", "/users/@me/guilds/{guild_id}", guild_id=guild_id))
|
|
|
|
def get_guild(self, guild_id: Snowflake, *, with_counts: bool = True) -> Response[guild.Guild]:
|
|
params = {"with_counts": int(with_counts)}
|
|
return self.request(Route("GET", "/guilds/{guild_id}", guild_id=guild_id), params=params)
|
|
|
|
def delete_guild(self, guild_id: Snowflake) -> Response[None]:
|
|
return self.request(Route("DELETE", "/guilds/{guild_id}", guild_id=guild_id))
|
|
|
|
def create_guild(
|
|
self,
|
|
name: str,
|
|
icon: Optional[str] = None,
|
|
*,
|
|
verification_level: Optional[guild.VerificationLevel] = None,
|
|
default_message_notifications: Optional[guild.DefaultMessageNotificationLevel] = None,
|
|
explicit_content_filter: Optional[guild.ExplicitContentFilterLevel] = None,
|
|
roles: Optional[List[guild.CreateGuildPlaceholderRole]] = None,
|
|
channels: Optional[List[guild.CreateGuildPlaceholderChannel]] = None,
|
|
afk_channel: Optional[Snowflake] = None,
|
|
afk_timeout: Optional[int] = None,
|
|
system_channel: Optional[Snowflake] = None,
|
|
system_channel_flags: Optional[int] = None,
|
|
) -> Response[guild.Guild]:
|
|
payload: guild.CreateGuild = {
|
|
"name": name,
|
|
}
|
|
if icon:
|
|
payload["icon"] = icon
|
|
if verification_level is not None:
|
|
payload["verification_level"] = verification_level
|
|
if default_message_notifications is not None:
|
|
payload["default_message_notifications"] = default_message_notifications
|
|
if explicit_content_filter is not None:
|
|
payload["explicit_content_filter"] = explicit_content_filter
|
|
if roles is not None:
|
|
payload["roles"] = roles
|
|
if channels is not None:
|
|
payload["channels"] = channels
|
|
if afk_channel is not None:
|
|
payload["afk_channel_id"] = afk_channel
|
|
if afk_timeout is not None:
|
|
payload["afk_timeout"] = afk_timeout
|
|
if system_channel is not None:
|
|
payload["system_channel_id"] = system_channel
|
|
if system_channel_flags is not None:
|
|
payload["system_channel_flags"] = system_channel_flags
|
|
|
|
return self.request(Route("POST", "/guilds"), json=payload)
|
|
|
|
def edit_guild(
|
|
self, guild_id: Snowflake, *, reason: Optional[str] = None, **fields: Any
|
|
) -> Response[guild.Guild]:
|
|
valid_keys = (
|
|
"name",
|
|
"icon",
|
|
"afk_timeout",
|
|
"owner_id",
|
|
"afk_channel_id",
|
|
"splash",
|
|
"discovery_splash",
|
|
"features",
|
|
"verification_level",
|
|
"system_channel_id",
|
|
"default_message_notifications",
|
|
"description",
|
|
"explicit_content_filter",
|
|
"banner",
|
|
"system_channel_flags",
|
|
"rules_channel_id",
|
|
"public_updates_channel_id",
|
|
"preferred_locale",
|
|
"premium_progress_bar_enabled",
|
|
"safety_alerts_channel_id",
|
|
)
|
|
|
|
payload = {k: v for k, v in fields.items() if k in valid_keys}
|
|
|
|
return self.request(
|
|
Route("PATCH", "/guilds/{guild_id}", guild_id=guild_id), json=payload, reason=reason
|
|
)
|
|
|
|
def get_template(self, code: str) -> Response[template.Template]:
|
|
return self.request(Route("GET", "/guilds/templates/{code}", code=code))
|
|
|
|
def guild_templates(self, guild_id: Snowflake) -> Response[List[template.Template]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/templates", guild_id=guild_id))
|
|
|
|
def create_template(
|
|
self, guild_id: Snowflake, payload: template.CreateTemplate
|
|
) -> Response[template.Template]:
|
|
return self.request(
|
|
Route("POST", "/guilds/{guild_id}/templates", guild_id=guild_id), json=payload
|
|
)
|
|
|
|
def sync_template(self, guild_id: Snowflake, code: str) -> Response[template.Template]:
|
|
return self.request(
|
|
Route("PUT", "/guilds/{guild_id}/templates/{code}", guild_id=guild_id, code=code)
|
|
)
|
|
|
|
def edit_template(
|
|
self, guild_id: Snowflake, code: str, payload: Dict[str, Any]
|
|
) -> Response[template.Template]:
|
|
valid_keys = (
|
|
"name",
|
|
"description",
|
|
)
|
|
payload = {k: v for k, v in payload.items() if k in valid_keys}
|
|
return self.request(
|
|
Route("PATCH", "/guilds/{guild_id}/templates/{code}", guild_id=guild_id, code=code),
|
|
json=payload,
|
|
)
|
|
|
|
def delete_template(self, guild_id: Snowflake, code: str) -> Response[None]:
|
|
return self.request(
|
|
Route("DELETE", "/guilds/{guild_id}/templates/{code}", guild_id=guild_id, code=code)
|
|
)
|
|
|
|
def create_from_template(
|
|
self, code: str, name: str, icon: Optional[str]
|
|
) -> Response[guild.Guild]:
|
|
payload = {
|
|
"name": name,
|
|
}
|
|
if icon:
|
|
payload["icon"] = icon
|
|
return self.request(Route("POST", "/guilds/templates/{code}", code=code), json=payload)
|
|
|
|
def get_guild_preview(self, guild_id: Snowflake) -> Response[guild.GuildPreview]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/preview", guild_id=guild_id))
|
|
|
|
def get_bans(
|
|
self,
|
|
guild_id: Snowflake,
|
|
limit: Optional[int] = None,
|
|
before: Optional[Snowflake] = None,
|
|
after: Optional[Snowflake] = None,
|
|
) -> Response[List[guild.Ban]]:
|
|
params: Dict[str, Any] = {}
|
|
|
|
if limit is not None:
|
|
params["limit"] = limit
|
|
if before is not None:
|
|
params["before"] = before
|
|
if after is not None:
|
|
params["after"] = after
|
|
|
|
return self.request(
|
|
Route("GET", "/guilds/{guild_id}/bans", guild_id=guild_id), params=params
|
|
)
|
|
|
|
def get_ban(self, user_id: Snowflake, guild_id: Snowflake) -> Response[guild.Ban]:
|
|
return self.request(
|
|
Route("GET", "/guilds/{guild_id}/bans/{user_id}", guild_id=guild_id, user_id=user_id)
|
|
)
|
|
|
|
def get_vanity_code(self, guild_id: Snowflake) -> Response[invite.VanityInvite]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/vanity-url", guild_id=guild_id))
|
|
|
|
def change_vanity_code(
|
|
self, guild_id: Snowflake, code: str, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
payload: Dict[str, Any] = {"code": code}
|
|
return self.request(
|
|
Route("PATCH", "/guilds/{guild_id}/vanity-url", guild_id=guild_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def edit_mfa_level(
|
|
self, guild_id: Snowflake, mfa_level: guild.MFALevel, *, reason: Optional[str] = None
|
|
) -> Response[guild.MFALevelUpdate]:
|
|
payload: guild.MFALevelUpdate = {"level": mfa_level}
|
|
return self.request(
|
|
Route("POST", "/guilds/{guild_id}/mfa", guild_id=guild_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def get_all_guild_channels(self, guild_id: Snowflake) -> Response[List[guild.GuildChannel]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/channels", guild_id=guild_id))
|
|
|
|
def get_members(
|
|
self, guild_id: Snowflake, limit: int, after: Optional[Snowflake]
|
|
) -> Response[List[member.MemberWithUser]]:
|
|
params: Dict[str, Any] = {
|
|
"limit": limit,
|
|
}
|
|
if after:
|
|
params["after"] = after
|
|
|
|
r = Route("GET", "/guilds/{guild_id}/members", guild_id=guild_id)
|
|
return self.request(r, params=params)
|
|
|
|
def get_member(
|
|
self, guild_id: Snowflake, member_id: Snowflake
|
|
) -> Response[member.MemberWithUser]:
|
|
return self.request(
|
|
Route(
|
|
"GET",
|
|
"/guilds/{guild_id}/members/{member_id}",
|
|
guild_id=guild_id,
|
|
member_id=member_id,
|
|
)
|
|
)
|
|
|
|
def prune_members(
|
|
self,
|
|
guild_id: Snowflake,
|
|
days: int,
|
|
compute_prune_count: bool,
|
|
roles: List[str],
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[guild.GuildPrune]:
|
|
payload: Dict[str, Any] = {
|
|
"days": days,
|
|
"compute_prune_count": "true" if compute_prune_count else "false",
|
|
}
|
|
if roles:
|
|
payload["include_roles"] = ", ".join(roles)
|
|
|
|
return self.request(
|
|
Route("POST", "/guilds/{guild_id}/prune", guild_id=guild_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def estimate_pruned_members(
|
|
self,
|
|
guild_id: Snowflake,
|
|
days: int,
|
|
roles: List[str],
|
|
) -> Response[guild.GuildPrune]:
|
|
params: Dict[str, Any] = {
|
|
"days": days,
|
|
}
|
|
if roles:
|
|
params["include_roles"] = ", ".join(roles)
|
|
|
|
return self.request(
|
|
Route("GET", "/guilds/{guild_id}/prune", guild_id=guild_id), params=params
|
|
)
|
|
|
|
def get_sticker(self, sticker_id: Snowflake) -> Response[sticker.Sticker]:
|
|
return self.request(Route("GET", "/stickers/{sticker_id}", sticker_id=sticker_id))
|
|
|
|
def list_premium_sticker_packs(self) -> Response[sticker.ListPremiumStickerPacks]:
|
|
return self.request(Route("GET", "/sticker-packs"))
|
|
|
|
def get_all_guild_stickers(self, guild_id: Snowflake) -> Response[List[sticker.GuildSticker]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/stickers", guild_id=guild_id))
|
|
|
|
def get_guild_sticker(
|
|
self, guild_id: Snowflake, sticker_id: Snowflake
|
|
) -> Response[sticker.GuildSticker]:
|
|
return self.request(
|
|
Route(
|
|
"GET",
|
|
"/guilds/{guild_id}/stickers/{sticker_id}",
|
|
guild_id=guild_id,
|
|
sticker_id=sticker_id,
|
|
)
|
|
)
|
|
|
|
def create_guild_sticker(
|
|
self,
|
|
guild_id: Snowflake,
|
|
payload: sticker.CreateGuildSticker,
|
|
file: File,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[sticker.GuildSticker]:
|
|
initial_bytes = file.fp.read(16)
|
|
|
|
try:
|
|
mime_type = utils._get_mime_type_for_image(initial_bytes)
|
|
except ValueError:
|
|
if initial_bytes.startswith(b"{"):
|
|
mime_type = "application/json"
|
|
else:
|
|
mime_type = "application/octet-stream"
|
|
finally:
|
|
file.reset()
|
|
|
|
form: List[Dict[str, Any]] = [
|
|
{
|
|
"name": "file",
|
|
"value": file.fp,
|
|
"filename": file.filename,
|
|
"content_type": mime_type,
|
|
}
|
|
]
|
|
|
|
for k, v in payload.items():
|
|
form.append(
|
|
{
|
|
"name": k,
|
|
"value": v,
|
|
}
|
|
)
|
|
|
|
return self.request(
|
|
Route("POST", "/guilds/{guild_id}/stickers", guild_id=guild_id),
|
|
form=form,
|
|
files=[file],
|
|
reason=reason,
|
|
)
|
|
|
|
def modify_guild_sticker(
|
|
self,
|
|
guild_id: Snowflake,
|
|
sticker_id: Snowflake,
|
|
payload: sticker.EditGuildSticker,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[sticker.GuildSticker]:
|
|
return self.request(
|
|
Route(
|
|
"PATCH",
|
|
"/guilds/{guild_id}/stickers/{sticker_id}",
|
|
guild_id=guild_id,
|
|
sticker_id=sticker_id,
|
|
),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def delete_guild_sticker(
|
|
self, guild_id: Snowflake, sticker_id: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
return self.request(
|
|
Route(
|
|
"DELETE",
|
|
"/guilds/{guild_id}/stickers/{sticker_id}",
|
|
guild_id=guild_id,
|
|
sticker_id=sticker_id,
|
|
),
|
|
reason=reason,
|
|
)
|
|
|
|
def get_all_custom_emojis(self, guild_id: Snowflake) -> Response[List[emoji.Emoji]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/emojis", guild_id=guild_id))
|
|
|
|
def get_custom_emoji(self, guild_id: Snowflake, emoji_id: Snowflake) -> Response[emoji.Emoji]:
|
|
return self.request(
|
|
Route(
|
|
"GET", "/guilds/{guild_id}/emojis/{emoji_id}", guild_id=guild_id, emoji_id=emoji_id
|
|
)
|
|
)
|
|
|
|
def create_custom_emoji(
|
|
self,
|
|
guild_id: Snowflake,
|
|
name: str,
|
|
image: str,
|
|
*,
|
|
roles: Optional[SnowflakeList] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[emoji.Emoji]:
|
|
payload: Dict[str, Any] = {
|
|
"name": name,
|
|
"image": image,
|
|
"roles": roles or [],
|
|
}
|
|
|
|
r = Route("POST", "/guilds/{guild_id}/emojis", guild_id=guild_id)
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def delete_custom_emoji(
|
|
self,
|
|
guild_id: Snowflake,
|
|
emoji_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE", "/guilds/{guild_id}/emojis/{emoji_id}", guild_id=guild_id, emoji_id=emoji_id
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def edit_custom_emoji(
|
|
self,
|
|
guild_id: Snowflake,
|
|
emoji_id: Snowflake,
|
|
*,
|
|
payload: Dict[str, Any],
|
|
reason: Optional[str] = None,
|
|
) -> Response[emoji.Emoji]:
|
|
r = Route(
|
|
"PATCH", "/guilds/{guild_id}/emojis/{emoji_id}", guild_id=guild_id, emoji_id=emoji_id
|
|
)
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def get_all_integrations(self, guild_id: Snowflake) -> Response[List[integration.Integration]]:
|
|
r = Route("GET", "/guilds/{guild_id}/integrations", guild_id=guild_id)
|
|
|
|
return self.request(r)
|
|
|
|
def create_integration(
|
|
self, guild_id: Snowflake, type: integration.IntegrationType, id: int
|
|
) -> Response[None]:
|
|
payload = {
|
|
"type": type,
|
|
"id": id,
|
|
}
|
|
|
|
r = Route("POST", "/guilds/{guild_id}/integrations", guild_id=guild_id)
|
|
return self.request(r, json=payload)
|
|
|
|
def edit_integration(
|
|
self, guild_id: Snowflake, integration_id: Snowflake, **payload: Any
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"PATCH",
|
|
"/guilds/{guild_id}/integrations/{integration_id}",
|
|
guild_id=guild_id,
|
|
integration_id=integration_id,
|
|
)
|
|
|
|
return self.request(r, json=payload)
|
|
|
|
def sync_integration(self, guild_id: Snowflake, integration_id: Snowflake) -> Response[None]:
|
|
r = Route(
|
|
"POST",
|
|
"/guilds/{guild_id}/integrations/{integration_id}/sync",
|
|
guild_id=guild_id,
|
|
integration_id=integration_id,
|
|
)
|
|
|
|
return self.request(r)
|
|
|
|
def delete_integration(
|
|
self, guild_id: Snowflake, integration_id: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/guilds/{guild_id}/integrations/{integration_id}",
|
|
guild_id=guild_id,
|
|
integration_id=integration_id,
|
|
)
|
|
|
|
return self.request(r, reason=reason)
|
|
|
|
def get_audit_logs(
|
|
self,
|
|
guild_id: Snowflake,
|
|
limit: int = 100,
|
|
# only one of these two may be specified, otherwise `after` gets ignored
|
|
before: Optional[Snowflake] = None,
|
|
after: Optional[Snowflake] = None,
|
|
user_id: Optional[Snowflake] = None,
|
|
action_type: Optional[audit_log.AuditLogEvent] = None,
|
|
) -> Response[audit_log.AuditLog]:
|
|
params: Dict[str, Any] = {"limit": limit}
|
|
if before is not None:
|
|
params["before"] = before
|
|
if after is not None:
|
|
params["after"] = after
|
|
if user_id:
|
|
params["user_id"] = user_id
|
|
if action_type:
|
|
params["action_type"] = action_type
|
|
|
|
r = Route("GET", "/guilds/{guild_id}/audit-logs", guild_id=guild_id)
|
|
return self.request(r, params=params)
|
|
|
|
def get_widget(self, guild_id: Snowflake) -> Response[widget.Widget]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/widget.json", guild_id=guild_id))
|
|
|
|
def get_widget_settings(self, guild_id: Snowflake) -> Response[widget.WidgetSettings]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/widget", guild_id=guild_id))
|
|
|
|
def edit_widget(
|
|
self, guild_id: Snowflake, payload: Dict[str, Any], *, reason: Optional[str] = None
|
|
) -> Response[widget.WidgetSettings]:
|
|
return self.request(
|
|
Route("PATCH", "/guilds/{guild_id}/widget", guild_id=guild_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def widget_image_url(self, guild_id: Snowflake, *, style: str) -> str:
|
|
return str(
|
|
yarl.URL(Route.BASE)
|
|
.with_path(f"/api/guilds/{guild_id}/widget.png")
|
|
.with_query(style=style)
|
|
)
|
|
|
|
# Invite management
|
|
|
|
def create_invite(
|
|
self,
|
|
channel_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
max_age: int = 0,
|
|
max_uses: int = 0,
|
|
temporary: bool = False,
|
|
unique: bool = True,
|
|
target_type: Optional[invite.InviteTargetType] = None,
|
|
target_user_id: Optional[Snowflake] = None,
|
|
target_application_id: Optional[Snowflake] = None,
|
|
) -> Response[invite.Invite]:
|
|
r = Route("POST", "/channels/{channel_id}/invites", channel_id=channel_id)
|
|
payload: Dict[str, Any] = {
|
|
"max_age": max_age,
|
|
"max_uses": max_uses,
|
|
"temporary": temporary,
|
|
"unique": unique,
|
|
}
|
|
|
|
if target_type:
|
|
payload["target_type"] = target_type
|
|
|
|
if target_user_id:
|
|
payload["target_user_id"] = target_user_id
|
|
|
|
if target_application_id:
|
|
payload["target_application_id"] = str(target_application_id)
|
|
|
|
return self.request(r, reason=reason, json=payload)
|
|
|
|
def get_invite(
|
|
self,
|
|
invite_id: str,
|
|
*,
|
|
with_counts: bool = True,
|
|
with_expiration: bool = True,
|
|
guild_scheduled_event_id: Optional[int] = None,
|
|
) -> Response[invite.Invite]:
|
|
params = {
|
|
"with_counts": int(with_counts),
|
|
"with_expiration": int(with_expiration),
|
|
}
|
|
if guild_scheduled_event_id:
|
|
params["guild_scheduled_event_id"] = guild_scheduled_event_id
|
|
|
|
return self.request(
|
|
Route("GET", "/invites/{invite_id}", invite_id=invite_id), params=params
|
|
)
|
|
|
|
def invites_from(self, guild_id: Snowflake) -> Response[List[invite.Invite]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/invites", guild_id=guild_id))
|
|
|
|
def invites_from_channel(self, channel_id: Snowflake) -> Response[List[invite.Invite]]:
|
|
return self.request(Route("GET", "/channels/{channel_id}/invites", channel_id=channel_id))
|
|
|
|
def delete_invite(self, invite_id: str, *, reason: Optional[str] = None) -> Response[None]:
|
|
return self.request(
|
|
Route("DELETE", "/invites/{invite_id}", invite_id=invite_id), reason=reason
|
|
)
|
|
|
|
# Role management
|
|
|
|
def get_roles(self, guild_id: Snowflake) -> Response[List[role.Role]]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/roles", guild_id=guild_id))
|
|
|
|
def edit_role(
|
|
self,
|
|
guild_id: Snowflake,
|
|
role_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**fields: Any,
|
|
) -> Response[role.Role]:
|
|
r = Route("PATCH", "/guilds/{guild_id}/roles/{role_id}", guild_id=guild_id, role_id=role_id)
|
|
valid_keys = (
|
|
"name",
|
|
"permissions",
|
|
"color",
|
|
"hoist",
|
|
"mentionable",
|
|
"icon",
|
|
"unicode_emoji",
|
|
)
|
|
payload = {k: v for k, v in fields.items() if k in valid_keys}
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def delete_role(
|
|
self, guild_id: Snowflake, role_id: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE", "/guilds/{guild_id}/roles/{role_id}", guild_id=guild_id, role_id=role_id
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def replace_roles(
|
|
self,
|
|
user_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
role_ids: List[int],
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[member.MemberWithUser]:
|
|
return self.edit_member(guild_id=guild_id, user_id=user_id, roles=role_ids, reason=reason)
|
|
|
|
def create_role(
|
|
self, guild_id: Snowflake, *, reason: Optional[str] = None, **fields: Any
|
|
) -> Response[role.Role]:
|
|
r = Route("POST", "/guilds/{guild_id}/roles", guild_id=guild_id)
|
|
return self.request(r, json=fields, reason=reason)
|
|
|
|
def move_role_position(
|
|
self,
|
|
guild_id: Snowflake,
|
|
positions: List[guild.RolePositionUpdate],
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[List[role.Role]]:
|
|
r = Route("PATCH", "/guilds/{guild_id}/roles", guild_id=guild_id)
|
|
return self.request(r, json=positions, reason=reason)
|
|
|
|
def add_role(
|
|
self,
|
|
guild_id: Snowflake,
|
|
user_id: Snowflake,
|
|
role_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"PUT",
|
|
"/guilds/{guild_id}/members/{user_id}/roles/{role_id}",
|
|
guild_id=guild_id,
|
|
user_id=user_id,
|
|
role_id=role_id,
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def remove_role(
|
|
self,
|
|
guild_id: Snowflake,
|
|
user_id: Snowflake,
|
|
role_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/guilds/{guild_id}/members/{user_id}/roles/{role_id}",
|
|
guild_id=guild_id,
|
|
user_id=user_id,
|
|
role_id=role_id,
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
def edit_channel_permissions(
|
|
self,
|
|
channel_id: Snowflake,
|
|
target: Snowflake,
|
|
allow: int,
|
|
deny: int,
|
|
type: channel.OverwriteType,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
payload = {"id": target, "allow": allow, "deny": deny, "type": type}
|
|
r = Route(
|
|
"PUT",
|
|
"/channels/{channel_id}/permissions/{target}",
|
|
channel_id=channel_id,
|
|
target=target,
|
|
)
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def delete_channel_permissions(
|
|
self, channel_id: Snowflake, target: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/channels/{channel_id}/permissions/{target}",
|
|
channel_id=channel_id,
|
|
target=target,
|
|
)
|
|
return self.request(r, reason=reason)
|
|
|
|
# Voice management
|
|
|
|
def move_member(
|
|
self,
|
|
user_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
channel_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[member.MemberWithUser]:
|
|
return self.edit_member(
|
|
guild_id=guild_id, user_id=user_id, channel_id=channel_id, reason=reason
|
|
)
|
|
|
|
# Stage instance management
|
|
|
|
def get_stage_instance(self, channel_id: Snowflake) -> Response[channel.StageInstance]:
|
|
return self.request(Route("GET", "/stage-instances/{channel_id}", channel_id=channel_id))
|
|
|
|
def create_stage_instance(
|
|
self, *, reason: Optional[str] = None, **payload: Any
|
|
) -> Response[channel.StageInstance]:
|
|
valid_keys = (
|
|
"channel_id",
|
|
"topic",
|
|
"privacy_level",
|
|
"send_start_notification",
|
|
)
|
|
payload = {k: v for k, v in payload.items() if k in valid_keys}
|
|
|
|
return self.request(Route("POST", "/stage-instances"), json=payload, reason=reason)
|
|
|
|
def edit_stage_instance(
|
|
self, channel_id: Snowflake, *, reason: Optional[str] = None, **payload: Any
|
|
) -> Response[None]:
|
|
valid_keys = (
|
|
"topic",
|
|
"privacy_level",
|
|
)
|
|
payload = {k: v for k, v in payload.items() if k in valid_keys}
|
|
|
|
return self.request(
|
|
Route("PATCH", "/stage-instances/{channel_id}", channel_id=channel_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def delete_stage_instance(
|
|
self, channel_id: Snowflake, *, reason: Optional[str] = None
|
|
) -> Response[None]:
|
|
return self.request(
|
|
Route("DELETE", "/stage-instances/{channel_id}", channel_id=channel_id), reason=reason
|
|
)
|
|
|
|
# Scheduled event management
|
|
|
|
def get_guild_scheduled_events(
|
|
self, guild_id: Snowflake, with_user_count: bool = False
|
|
) -> Response[List[guild_scheduled_event.GuildScheduledEvent]]:
|
|
params = {"with_user_count": int(with_user_count)}
|
|
r = Route("GET", "/guilds/{guild_id}/scheduled-events", guild_id=guild_id)
|
|
return self.request(r, params=params)
|
|
|
|
def create_guild_scheduled_event(
|
|
self,
|
|
guild_id: Snowflake,
|
|
*,
|
|
name: str,
|
|
privacy_level: int,
|
|
scheduled_start_time: str,
|
|
entity_type: int,
|
|
channel_id: Optional[Snowflake] = None,
|
|
entity_metadata: Optional[Dict[str, Any]] = None,
|
|
scheduled_end_time: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
image: Optional[str] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[guild_scheduled_event.GuildScheduledEvent]:
|
|
r = Route("POST", "/guilds/{guild_id}/scheduled-events", guild_id=guild_id)
|
|
payload: Dict[str, Any] = {
|
|
"name": name,
|
|
"privacy_level": privacy_level,
|
|
"scheduled_start_time": scheduled_start_time,
|
|
"entity_type": entity_type,
|
|
}
|
|
|
|
if channel_id is not None:
|
|
payload["channel_id"] = channel_id
|
|
|
|
if entity_metadata is not None:
|
|
payload["entity_metadata"] = entity_metadata
|
|
|
|
if scheduled_end_time is not None:
|
|
payload["scheduled_end_time"] = scheduled_end_time
|
|
|
|
if description is not None:
|
|
payload["description"] = description
|
|
|
|
if image is not None:
|
|
payload["image"] = image
|
|
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
def get_guild_scheduled_event(
|
|
self, guild_id: Snowflake, event_id: Snowflake, with_user_count: bool = False
|
|
) -> Response[guild_scheduled_event.GuildScheduledEvent]:
|
|
params = {"with_user_count": int(with_user_count)}
|
|
route = Route(
|
|
method="GET",
|
|
path="/guilds/{guild_id}/scheduled-events/{event_id}",
|
|
guild_id=guild_id,
|
|
event_id=event_id,
|
|
)
|
|
return self.request(route, params=params)
|
|
|
|
def edit_guild_scheduled_event(
|
|
self,
|
|
guild_id: Snowflake,
|
|
event_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**fields: Any,
|
|
) -> Response[guild_scheduled_event.GuildScheduledEvent]:
|
|
route = Route(
|
|
method="PATCH",
|
|
path="/guilds/{guild_id}/scheduled-events/{event_id}",
|
|
guild_id=guild_id,
|
|
event_id=event_id,
|
|
)
|
|
|
|
return self.request(route, json=fields, reason=reason)
|
|
|
|
def delete_guild_scheduled_event(
|
|
self, guild_id: Snowflake, event_id: Snowflake
|
|
) -> Response[None]:
|
|
route = Route(
|
|
method="DELETE",
|
|
path="/guilds/{guild_id}/scheduled-events/{event_id}",
|
|
guild_id=guild_id,
|
|
event_id=event_id,
|
|
)
|
|
return self.request(route)
|
|
|
|
def get_guild_scheduled_event_users(
|
|
self,
|
|
guild_id: Snowflake,
|
|
event_id: Snowflake,
|
|
limit: Optional[int] = None,
|
|
with_member: Optional[bool] = None,
|
|
before: Optional[Snowflake] = None,
|
|
after: Optional[Snowflake] = None,
|
|
) -> Response[List[guild_scheduled_event.GuildScheduledEventUser]]:
|
|
params: Dict[str, Any] = {}
|
|
|
|
if limit is not None:
|
|
params["limit"] = limit
|
|
|
|
if with_member is not None:
|
|
params["with_member"] = int(with_member)
|
|
|
|
if before is not None:
|
|
params["before"] = before
|
|
|
|
if after is not None:
|
|
params["after"] = after
|
|
|
|
route = Route(
|
|
method="GET",
|
|
path="/guilds/{guild_id}/scheduled-events/{event_id}/users",
|
|
guild_id=guild_id,
|
|
event_id=event_id,
|
|
)
|
|
return self.request(route, params=params)
|
|
|
|
# Welcome screens
|
|
|
|
def get_guild_welcome_screen(
|
|
self, guild_id: Snowflake
|
|
) -> Response[welcome_screen.WelcomeScreen]:
|
|
r = Route("GET", "/guilds/{guild_id}/welcome-screen", guild_id=guild_id)
|
|
return self.request(r)
|
|
|
|
def edit_guild_welcome_screen(
|
|
self,
|
|
guild_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**kwargs,
|
|
) -> Response[welcome_screen.WelcomeScreen]:
|
|
valid_keys = (
|
|
"enabled",
|
|
"welcome_channels",
|
|
"description",
|
|
)
|
|
payload = {k: v for k, v in kwargs.items() if k in valid_keys}
|
|
|
|
r = Route("PATCH", "/guilds/{guild_id}/welcome-screen", guild_id=guild_id)
|
|
return self.request(r, json=payload, reason=reason)
|
|
|
|
# Auto moderation
|
|
|
|
def get_auto_moderation_rules(self, guild_id: Snowflake) -> Response[List[automod.AutoModRule]]:
|
|
return self.request(
|
|
Route("GET", "/guilds/{guild_id}/auto-moderation/rules", guild_id=guild_id)
|
|
)
|
|
|
|
def get_auto_moderation_rule(
|
|
self, guild_id: Snowflake, rule_id: Snowflake
|
|
) -> Response[automod.AutoModRule]:
|
|
return self.request(
|
|
Route(
|
|
"GET",
|
|
"/guilds/{guild_id}/auto-moderation/rules/{rule_id}",
|
|
guild_id=guild_id,
|
|
rule_id=rule_id,
|
|
)
|
|
)
|
|
|
|
def create_auto_moderation_rule(
|
|
self,
|
|
guild_id: Snowflake,
|
|
*,
|
|
name: str,
|
|
event_type: automod.AutoModEventType,
|
|
trigger_type: automod.AutoModTriggerType,
|
|
actions: List[automod.AutoModAction],
|
|
trigger_metadata: Optional[automod.AutoModTriggerMetadata] = None,
|
|
enabled: Optional[bool] = None,
|
|
exempt_roles: Optional[SnowflakeList] = None,
|
|
exempt_channels: Optional[SnowflakeList] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Response[automod.AutoModRule]:
|
|
payload: automod.CreateAutoModRule = {
|
|
"name": name,
|
|
"event_type": event_type,
|
|
"trigger_type": trigger_type,
|
|
"actions": actions,
|
|
}
|
|
|
|
if trigger_metadata is not None:
|
|
payload["trigger_metadata"] = trigger_metadata
|
|
if enabled is not None:
|
|
payload["enabled"] = enabled
|
|
if exempt_roles is not None:
|
|
payload["exempt_roles"] = exempt_roles
|
|
if exempt_channels is not None:
|
|
payload["exempt_channels"] = exempt_channels
|
|
|
|
return self.request(
|
|
Route("POST", "/guilds/{guild_id}/auto-moderation/rules", guild_id=guild_id),
|
|
json=payload,
|
|
reason=reason,
|
|
)
|
|
|
|
def edit_auto_moderation_rule(
|
|
self,
|
|
guild_id: Snowflake,
|
|
rule_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
**fields: Any,
|
|
) -> Response[automod.AutoModRule]:
|
|
return self.request(
|
|
Route(
|
|
"PATCH",
|
|
"/guilds/{guild_id}/auto-moderation/rules/{rule_id}",
|
|
guild_id=guild_id,
|
|
rule_id=rule_id,
|
|
),
|
|
json=fields,
|
|
reason=reason,
|
|
)
|
|
|
|
def delete_auto_moderation_rule(
|
|
self,
|
|
guild_id: Snowflake,
|
|
rule_id: Snowflake,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
) -> Response[None]:
|
|
return self.request(
|
|
Route(
|
|
"DELETE",
|
|
"/guilds/{guild_id}/auto-moderation/rules/{rule_id}",
|
|
guild_id=guild_id,
|
|
rule_id=rule_id,
|
|
),
|
|
reason=reason,
|
|
)
|
|
|
|
# Guild Onboarding
|
|
|
|
def get_guild_onboarding(self, guild_id: Snowflake) -> Response[onboarding.Onboarding]:
|
|
return self.request(Route("GET", "/guilds/{guild_id}/onboarding", guild_id=guild_id))
|
|
|
|
# Application commands (global)
|
|
|
|
def get_global_commands(
|
|
self,
|
|
application_id: Snowflake,
|
|
*,
|
|
with_localizations: bool = True,
|
|
) -> Response[List[interactions.ApplicationCommand]]:
|
|
params: Dict[str, Any] = {}
|
|
# the API currently interprets any non-empty value as truthy
|
|
if with_localizations:
|
|
params["with_localizations"] = int(with_localizations)
|
|
|
|
return self.request(
|
|
Route("GET", "/applications/{application_id}/commands", application_id=application_id),
|
|
params=params,
|
|
)
|
|
|
|
def get_global_command(
|
|
self, application_id: Snowflake, command_id: Snowflake
|
|
) -> Response[interactions.ApplicationCommand]:
|
|
r = Route(
|
|
"GET",
|
|
"/applications/{application_id}/commands/{command_id}",
|
|
application_id=application_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def upsert_global_command(
|
|
self, application_id: Snowflake, payload: interactions.EditApplicationCommand
|
|
) -> Response[interactions.ApplicationCommand]:
|
|
r = Route("POST", "/applications/{application_id}/commands", application_id=application_id)
|
|
return self.request(r, json=payload)
|
|
|
|
def edit_global_command(
|
|
self,
|
|
application_id: Snowflake,
|
|
command_id: Snowflake,
|
|
payload: interactions.EditApplicationCommand,
|
|
) -> Response[interactions.ApplicationCommand]:
|
|
r = Route(
|
|
"PATCH",
|
|
"/applications/{application_id}/commands/{command_id}",
|
|
application_id=application_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r, json=payload)
|
|
|
|
def delete_global_command(
|
|
self, application_id: Snowflake, command_id: Snowflake
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/applications/{application_id}/commands/{command_id}",
|
|
application_id=application_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def bulk_upsert_global_commands(
|
|
self, application_id: Snowflake, payload: List[interactions.EditApplicationCommand]
|
|
) -> Response[List[interactions.ApplicationCommand]]:
|
|
r = Route("PUT", "/applications/{application_id}/commands", application_id=application_id)
|
|
return self.request(r, json=payload)
|
|
|
|
# Application commands (guild)
|
|
|
|
def get_guild_commands(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
*,
|
|
with_localizations: bool = True,
|
|
) -> Response[List[interactions.ApplicationCommand]]:
|
|
params: Dict[str, Any] = {}
|
|
# the API currently interprets any non-empty value as truthy
|
|
if with_localizations:
|
|
params["with_localizations"] = int(with_localizations)
|
|
|
|
r = Route(
|
|
"GET",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
)
|
|
return self.request(r, params=params)
|
|
|
|
def get_guild_command(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
command_id: Snowflake,
|
|
) -> Response[interactions.ApplicationCommand]:
|
|
r = Route(
|
|
"GET",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands/{command_id}",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def upsert_guild_command(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
payload: interactions.EditApplicationCommand,
|
|
) -> Response[interactions.ApplicationCommand]:
|
|
r = Route(
|
|
"POST",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
)
|
|
return self.request(r, json=payload)
|
|
|
|
def edit_guild_command(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
command_id: Snowflake,
|
|
payload: interactions.EditApplicationCommand,
|
|
) -> Response[interactions.ApplicationCommand]:
|
|
r = Route(
|
|
"PATCH",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands/{command_id}",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r, json=payload)
|
|
|
|
def delete_guild_command(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
command_id: Snowflake,
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands/{command_id}",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def bulk_upsert_guild_commands(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
payload: List[interactions.EditApplicationCommand],
|
|
) -> Response[List[interactions.ApplicationCommand]]:
|
|
r = Route(
|
|
"PUT",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
)
|
|
return self.request(r, json=payload)
|
|
|
|
# Interaction responses
|
|
|
|
def _edit_webhook_helper(
|
|
self,
|
|
route: Route,
|
|
file: Optional[File] = None,
|
|
content: Optional[str] = None,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
attachments: Optional[List[Attachment]] = None,
|
|
) -> Response[message.Message]:
|
|
# TODO: this does not work how it should (e.g. `embeds=[]` is ignored).
|
|
# This method (or rather its calling methods) is completely unused, and hence likely untested
|
|
|
|
payload: Dict[str, Any] = {}
|
|
if content:
|
|
payload["content"] = content
|
|
if embeds:
|
|
payload["embeds"] = embeds
|
|
if allowed_mentions:
|
|
payload["allowed_mentions"] = allowed_mentions
|
|
if attachments:
|
|
payload["attachments"] = attachments
|
|
|
|
if file:
|
|
multipart = to_multipart_with_attachments(payload, [file])
|
|
return self.request(route, form=multipart, files=[file])
|
|
return self.request(route, json=payload)
|
|
|
|
def create_interaction_response(
|
|
self,
|
|
interaction_id: Snowflake,
|
|
token: str,
|
|
*,
|
|
type: InteractionResponseType,
|
|
data: Optional[interactions.InteractionCallbackData] = None,
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"POST",
|
|
"/interactions/{interaction_id}/{interaction_token}/callback",
|
|
interaction_id=interaction_id,
|
|
interaction_token=token,
|
|
)
|
|
payload: Dict[str, Any] = {
|
|
"type": type,
|
|
}
|
|
|
|
if data is not None:
|
|
payload["data"] = data
|
|
|
|
return self.request(r, json=payload)
|
|
|
|
def get_original_interaction_response(
|
|
self,
|
|
application_id: Snowflake,
|
|
token: str,
|
|
) -> Response[message.Message]:
|
|
r = Route(
|
|
"GET",
|
|
"/webhooks/{application_id}/{interaction_token}/messages/@original",
|
|
application_id=application_id,
|
|
interaction_token=token,
|
|
)
|
|
return self.request(r)
|
|
|
|
def edit_original_interaction_response(
|
|
self,
|
|
application_id: Snowflake,
|
|
token: str,
|
|
file: Optional[File] = None,
|
|
content: Optional[str] = None,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
attachments: Optional[List[Attachment]] = None,
|
|
) -> Response[message.Message]:
|
|
r = Route(
|
|
"PATCH",
|
|
"/webhooks/{application_id}/{interaction_token}/messages/@original",
|
|
application_id=application_id,
|
|
interaction_token=token,
|
|
)
|
|
return self._edit_webhook_helper(
|
|
r,
|
|
file=file,
|
|
content=content,
|
|
embeds=embeds,
|
|
allowed_mentions=allowed_mentions,
|
|
attachments=attachments,
|
|
)
|
|
|
|
def delete_original_interaction_response(
|
|
self, application_id: Snowflake, token: str
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/webhooks/{application_id}/{interaction_token}/messages/@original",
|
|
application_id=application_id,
|
|
interaction_token=token,
|
|
)
|
|
return self.request(r)
|
|
|
|
def create_followup_message(
|
|
self,
|
|
application_id: Snowflake,
|
|
token: str,
|
|
files: Optional[List[File]] = None,
|
|
content: Optional[str] = None,
|
|
tts: bool = False,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
) -> Response[message.Message]:
|
|
r = Route(
|
|
"POST",
|
|
"/webhooks/{application_id}/{interaction_token}",
|
|
application_id=application_id,
|
|
interaction_token=token,
|
|
)
|
|
if files is None:
|
|
files = []
|
|
return self.send_multipart_helper(
|
|
r,
|
|
content=content,
|
|
files=files,
|
|
tts=tts,
|
|
embeds=embeds,
|
|
allowed_mentions=allowed_mentions,
|
|
)
|
|
|
|
def edit_followup_message(
|
|
self,
|
|
application_id: Snowflake,
|
|
token: str,
|
|
message_id: Snowflake,
|
|
file: Optional[File] = None,
|
|
content: Optional[str] = None,
|
|
embeds: Optional[List[embed.Embed]] = None,
|
|
allowed_mentions: Optional[message.AllowedMentions] = None,
|
|
attachments: Optional[List[Attachment]] = None,
|
|
) -> Response[message.Message]:
|
|
r = Route(
|
|
"PATCH",
|
|
"/webhooks/{application_id}/{interaction_token}/messages/{message_id}",
|
|
application_id=application_id,
|
|
interaction_token=token,
|
|
message_id=message_id,
|
|
)
|
|
return self._edit_webhook_helper(
|
|
r,
|
|
file=file,
|
|
content=content,
|
|
embeds=embeds,
|
|
allowed_mentions=allowed_mentions,
|
|
attachments=attachments,
|
|
)
|
|
|
|
def delete_followup_message(
|
|
self, application_id: Snowflake, token: str, message_id: Snowflake
|
|
) -> Response[None]:
|
|
r = Route(
|
|
"DELETE",
|
|
"/webhooks/{application_id}/{interaction_token}/messages/{message_id}",
|
|
application_id=application_id,
|
|
interaction_token=token,
|
|
message_id=message_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def get_guild_application_command_permissions(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
) -> Response[List[interactions.GuildApplicationCommandPermissions]]:
|
|
r = Route(
|
|
"GET",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands/permissions",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
def get_application_command_permissions(
|
|
self,
|
|
application_id: Snowflake,
|
|
guild_id: Snowflake,
|
|
command_id: Snowflake,
|
|
) -> Response[interactions.GuildApplicationCommandPermissions]:
|
|
r = Route(
|
|
"GET",
|
|
"/applications/{application_id}/guilds/{guild_id}/commands/{command_id}/permissions",
|
|
application_id=application_id,
|
|
guild_id=guild_id,
|
|
command_id=command_id,
|
|
)
|
|
return self.request(r)
|
|
|
|
# Misc
|
|
|
|
def get_voice_regions(self) -> Response[List[voice.VoiceRegion]]:
|
|
return self.request(Route("GET", "/voice/regions"))
|
|
|
|
def application_info(self) -> Response[appinfo.AppInfo]:
|
|
return self.request(Route("GET", "/oauth2/applications/@me"))
|
|
|
|
def get_application_role_connection_metadata_records(
|
|
self, application_id: Snowflake
|
|
) -> Response[List[application_role_connection.ApplicationRoleConnectionMetadata]]:
|
|
return self.request(
|
|
Route(
|
|
"GET",
|
|
"/applications/{application_id}/role-connections/metadata",
|
|
application_id=application_id,
|
|
)
|
|
)
|
|
|
|
def edit_application_role_connection_metadata_records(
|
|
self,
|
|
application_id: Snowflake,
|
|
records: Sequence[application_role_connection.ApplicationRoleConnectionMetadata],
|
|
) -> Response[List[application_role_connection.ApplicationRoleConnectionMetadata]]:
|
|
return self.request(
|
|
Route(
|
|
"PUT",
|
|
"/applications/{application_id}/role-connections/metadata",
|
|
application_id=application_id,
|
|
),
|
|
json=records,
|
|
)
|
|
|
|
async def get_gateway(self, *, encoding: str = "json", zlib: bool = True) -> str:
|
|
try:
|
|
data: gateway.Gateway = await self.request(Route("GET", "/gateway"))
|
|
except HTTPException as exc:
|
|
raise GatewayNotFound from exc
|
|
|
|
return self._format_gateway_url(data["url"], encoding=encoding, zlib=zlib)
|
|
|
|
async def get_bot_gateway(
|
|
self, *, encoding: str = "json", zlib: bool = True
|
|
) -> Tuple[int, str, gateway.SessionStartLimit]:
|
|
try:
|
|
data: gateway.GatewayBot = await self.request(Route("GET", "/gateway/bot"))
|
|
except HTTPException as exc:
|
|
raise GatewayNotFound from exc
|
|
|
|
return (
|
|
data["shards"],
|
|
self._format_gateway_url(data["url"], encoding=encoding, zlib=zlib),
|
|
data["session_start_limit"],
|
|
)
|
|
|
|
@staticmethod
|
|
def _format_gateway_url(url: str, *, encoding: str, zlib: bool) -> str:
|
|
_url = yarl.URL(url)
|
|
params = _url.query.copy()
|
|
params["v"] = str(_API_VERSION)
|
|
params["encoding"] = encoding
|
|
if zlib:
|
|
params["compress"] = "zlib-stream"
|
|
else:
|
|
params.popall("compress", None)
|
|
return str(_url.with_query(params))
|
|
|
|
def get_user(self, user_id: Snowflake) -> Response[user.User]:
|
|
return self.request(Route("GET", "/users/{user_id}", user_id=user_id))
|