discord exceptions logger for servers

discord_logger.py

import functools
import traceback
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import asynccontextmanager
from typing import Concatenate, Protocol

import anyio
import discord
from loguru import logger

from core.settings import AppSettings


class DiscordClientWrapper:
	def __init__(self, client: discord.Client, settings: AppSettings) -> None:
		self._client = client
		self._settings = settings

	@staticmethod
	def fail_safe[*T_args](
		fn: Callable[["DiscordClientWrapper", *T_args], Awaitable[None]],
	) -> Callable[["DiscordClientWrapper", *T_args], Awaitable[None]]:
		@functools.wraps(fn)
		async def wrapper(
			self_: "DiscordClientWrapper", *args: *T_args
		) -> None:  # pragma: no cover
			try:
				return await fn(self_, *args)
			except Exception as e:  # noqa: BLE001
				logger.warning(f"failed to interact with discord.\n {e}")

		return wrapper

	@staticmethod
	def format_py(v: str) -> str:
		return f"```py\n{v}\n```"

	@fail_safe
	async def log_exception(self, exc: Exception) -> None:  # pragma: no cover
		chan = self._client.get_channel(404)
		if chan and isinstance(chan, discord.TextChannel):
      role_id = "404"
			await chan.send(
				f"### I caught an exception cc/<@&{role_id}> ",
				allowed_mentions=discord.AllowedMentions(roles=True),
			)
			formatted_tb = traceback.format_tb(exc.__traceback__)
			while formatted_tb:
				buffer = ""
				while len(buffer) < 1850 and formatted_tb:
					buffer += f"{formatted_tb.pop()}\n"
				await chan.send(self.format_py(buffer))


@asynccontextmanager
async def get_discord_client(settings: AppSettings) -> AsyncIterator[DiscordClientWrapper]:
	async with discord.Client() as client:
		if settings.discord_token is not None:
			await client.login(settings.discord_token)
		async with anyio.create_task_group() as tg:
			tg.start_soon(client.connect)
			await client.wait_until_ready()
			yield DiscordClientWrapper(client, settings)


class HasDiscordClientProto(Protocol):
	@property
	def discord_client(self) -> DiscordClientWrapper:  # pragma: no cover
		...


async def log_exceptions_to_discord[T_self: HasDiscordClientProto, **Ps, T_ret](
	fn: Callable[Concatenate[T_self, Ps], Awaitable[T_ret]],
) -> Callable[Concatenate[T_self, Ps], Awaitable[T_ret]]:  # pragma: no cover
	@functools.wraps(fn)
	async def wrapper(self_: T_self, *args: Ps.args, **kwargs: Ps.kwargs) -> T_ret:
		try:
			return await fn(self_, *args, **kwargs)
		except Exception as e:
			await self_.discord_client.log_exception(e)
			raise

	return wrapper