add webhooks

This commit is contained in:
Victor Kotlin 2021-09-12 12:50:38 +03:00
parent 44810409de
commit b30729eca2
7 changed files with 211 additions and 3 deletions

View file

@ -1 +1,2 @@
from .client import BoticordClient
from .webhook import BoticordWebhook

View file

@ -32,7 +32,8 @@ class BoticordClient:
"Bots",
"Servers",
"Users",
"bot"
"bot",
"events"
)
bot: Union[commands.Bot, commands.AutoShardedBot]
@ -40,11 +41,29 @@ class BoticordClient:
def __init__(self, bot, token=None, **kwargs):
loop = kwargs.get('loop') or asyncio.get_event_loop()
session = kwargs.get('session') or aiohttp.ClientSession(loop=loop)
self.events = {}
self.bot = bot
self.Bots = Bots(bot, token=token, loop=loop, session=session)
self.Servers = Servers(bot, token=token, loop=loop, session=session)
self.Users = Users(token=token, loop=loop, session=session)
def event(self, event_name: str):
"""
A decorator that registers an event to listen to.
You can find all the events on Event Reference page.
Parameters
----------
event_name :class:`str`
boticord event name
"""
def inner(func):
if not asyncio.iscoroutinefunction(func):
raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
self.events[event_name] = func
return func
return inner
def start_loop(self, sleep_time: int = None) -> None:
"""
@ -53,8 +72,7 @@ class BoticordClient:
Parameters
----------
sleep_time: :class:`int`
loop sleep time - can be unfilled or None
stats posting interval - can be not specified or None (default interval - 15 minutes)
"""
self.bot.loop.create_task(self.__loop(sleep_time=sleep_time))

103
boticordpy/webhook.py Normal file
View file

@ -0,0 +1,103 @@
from aiohttp import web
import aiohttp
from discord.ext.commands import Bot, AutoShardedBot
import sys
if sys.version_info >= (3, 8):
from typing import TypedDict
else:
from typing_extensions import TypedDict
from aiohttp.web_urldispatcher import _WebHandler
from typing import Dict, Union
from . import BoticordClient
class _Webhook(TypedDict):
route: str
hook_key: str
func: "_WebHandler"
class BoticordWebhook:
"""
This class is used as a manager for the Boticord webhook.
Parameters
----------
bot : :class:`commands.Bot` | :class:`commands.AutoShardedBot`
The discord.py Bot instance
"""
__app: web.Application
_webhooks: Dict[
str,
_Webhook,
]
_webserver: web.TCPSite
def __init__(self, bot: Union[Bot, AutoShardedBot], boticord_client: BoticordClient):
self.bot = bot
self.boticord_client = boticord_client
self._webhooks = {}
self.__app = web.Application()
def bot_webhook(self, route: str = "/bot", hook_key: str = "") -> "BoticordWebhook":
"""This method may be used to configure the route of boticord bot's webhook.
Parameters
----------
route: str
Bot's webhook route. Must start with ``/``. Defaults - ``/bot``.
hook_key: str
Webhook authorization key.
Returns
----------
:class:`BoticordWebhook`
"""
self._webhooks["bot"] = _Webhook(
route=route or "/bot",
hook_key=hook_key or "",
func=self._bot_webhook_interaction_handler,
)
return self
async def _bot_webhook_interaction_handler(self, request: aiohttp.web.Request) -> web.Response:
auth = request.headers.get("X-Hook-Key")
if auth == self._webhooks["bot"]["hook_key"]:
data = (await request.json())
try:
await self.boticord_client.events[data["type"]](data)
except:
pass
return web.Response(status=200)
return web.Response(status=401)
async def _run(self, port: int):
for webhook in self._webhooks.values():
self.__app.router.add_post(webhook["route"], webhook["func"])
runner = web.AppRunner(self.__app)
await runner.setup()
self._webserver = web.TCPSite(runner, "0.0.0.0", port)
await self._webserver.start()
def run(self, port: int):
"""Runs the webhook.
Parameters
----------
port: int
The port to run the webhook on.
"""
self.bot.loop.create_task(self._run(port))
async def close(self) -> None:
"""Stops the webhook."""
await self._webserver.stop()

60
docs/events.rst Normal file
View file

@ -0,0 +1,60 @@
.. currentmodule:: boticordpy
.. event_reference:
Event Reference
---------------
Example of event creation:
::
from discord.ext import commands
from boticordpy import BoticordWebhook, BoticordClient
bot = commands.Bot(command_prefix="!")
boticord = BoticordClient(bot, "boticord-api-token")
boticord_webhook = BoticordWebhook(bot, boticord).bot_webhook("/bot", "X-Hook-Key")
boticord_webhook.run(5000)
@boticord.event("edit_bot_comment")
async def on_boticord_comment_edit(data):
print(data)
You can name the function whatever you want, but the decorator must always specify an existing event as an argument.
.. warning::
All the events must be a **coroutine**. If they aren't, then you might get unexpected
errors. In order to turn a function into a coroutine they must be ``async def``
functions.
.. function:: new_bot_bump
Called when the user bumps the bot.
Return Example: ``{'type': 'new_bot_bump', 'data': {'user': '809377165534822410', 'at': 1631436624444}}``
.. function:: new_bot_comment
Called when the user creates new comment.
Return Example: ``{'type': 'new_bot_comment', 'data': {'user': '704373738086465607', 'comment': {'old': None, 'new': 'boticord po jizni top'}, 'at': 1631439995678}}
``
.. function:: edit_bot_comment
Called when the user edits his comment.
Return Example: ``{'type': 'edit_bot_comment', 'data': {'user': '585766846268047370', 'comment': {'old': 'Boticord eto horosho', 'new': 'Boticord horoshiy monitoring'}, 'at': 1631438224813}}``
.. function:: delete_bot_comment
Called when the user deletes his comment.
Return Example:
{'type': 'delete_bot_comment', 'data': {'user': '704373738086465607', 'comment': 'допустим что я картофель', 'vote': 1, 'reason': 'self', 'at': 1631439759384}}

View file

@ -17,6 +17,8 @@ This is a documentation for simple python module to work with the boticord api.
quickstart
main
modules
webhook
event_reference
exceptions
Links

8
docs/webhook.rst Normal file
View file

@ -0,0 +1,8 @@
.. currentmodule:: boticordpy
.. webhook:
BoticordWebhook
---------------
.. autoclass:: boticordpy.BoticordWebhook
:members:

16
examples/webhook.py Normal file
View file

@ -0,0 +1,16 @@
from discord.ext import commands
from boticordpy import BoticordWebhook, BoticordClient
bot = commands.Bot(command_prefix="!")
boticord = BoticordClient(bot, "boticord-api-token")
boticord_webhook = BoticordWebhook(bot, boticord).bot_webhook("/bot", "X-Hook-Key")
boticord_webhook.run(5000)
@boticord.event("edit_bot_comment")
async def on_boticord_comment_edit(data):
print(data)
bot.run("bot-token")