# Basic wrapper for webhook connections
from __future__ import annotations
import typing
from aiohttp import ClientSession, ClientResponse, FormData
from asyncio import (
AbstractEventLoop,
get_event_loop
)
from acord.errors import *
from acord.core.abc import Route
T = typing.TypeVar("T")
Response = typing.Coroutine[typing.Any, typing.Any, T]
[docs]class WebhookConnection:
"""Base webhook connection,
wraps the session to be compatible with other ACord objects.
Parameters
----------
loop: :obj:`py:asyncio.AbstractEventLoop`
Loop to create connection off of
session: :class:`~aiohttp.ClientSession`
A pre-existing session to use
client: :class:`Client`
Client to attach webhook to.
.. DANGER::
If this param is provided,
:param:`WebhookConnection.session` is ignored.
We will instead use the session generated by the client
**kwds:
Additional kwargs to pass through session,
if it has not been provided.
.. note::
You may not provide the loop param through kwds,
use :param:`WebhookConnection.loop`.
"""
def __init__(self, *,
loop: AbstractEventLoop = get_event_loop(),
session: ClientSession = None,
client: typing.Any = None,
**kwds
) -> None:
self.loop = loop
self.client = client
if client:
self.session = client.http._session
else:
if session is None:
self.session = session or ClientSession(loop=self.loop, **kwds)
else:
self.session = session
[docs] async def request(
self,
route: Route,
**kwds,
) -> ClientResponse:
resp = await self.session.request(method=route.method, url=route.url, **kwds)
if 200 <= resp.status < 300:
return resp
respData = await self.decodeResponse(resp)
if 500 <= resp.status < 600:
raise DiscordError(str(respData))
if resp.status == 403:
raise Forbidden(str(respData), payload=respData, status_code=403)
if resp.status == 404:
raise NotFound(str(respData), payload=respData, status_code=404)
raise BadRequest(str(respData), payload=respData, status_code=resp.status)