""" 后端驱动适配基类 ================= 各驱动请继承以下基类 """ import abc import asyncio from dataclasses import dataclass from contextlib import asynccontextmanager from typing import ( TYPE_CHECKING, Any, Set, Dict, Type, Callable, Awaitable, AsyncGenerator, ) from ._model import URL as URL from nonebot.log import logger from nonebot.utils import escape_tag from ._model import Request as Request from nonebot.config import Env, Config from ._model import Response as Response from ._model import WebSocket as WebSocket from ._model import HTTPVersion as HTTPVersion from nonebot.typing import T_BotConnectionHook, T_BotDisconnectionHook if TYPE_CHECKING: from nonebot.adapters import Bot, Adapter class Driver(abc.ABC): """ Driver 基类。 """ _adapters: Dict[str, "Adapter"] = {} """ 已注册的适配器列表 """ _bot_connection_hook: Set[T_BotConnectionHook] = set() """ Bot 连接建立时执行的函数 """ _bot_disconnection_hook: Set[T_BotDisconnectionHook] = set() """ Bot 连接断开时执行的函数 """ def __init__(self, env: Env, config: Config): """ :参数: env: 包含环境信息的 Env 对象 config: 包含配置信息的 Config 对象 """ self.env: str = env.environment """ 环境名称 """ self.config: Config = config """ 配置对象 """ self._clients: Dict[str, "Bot"] = {} """ 已连接的 Bot """ @property def bots(self) -> Dict[str, "Bot"]: """ 获取当前所有已连接的 Bot """ return self._clients def register_adapter(self, adapter: Type["Adapter"], **kwargs) -> None: """ 注册一个协议适配器 :参数: name: 适配器名称,用于在连接时进行识别 adapter: 适配器 Class **kwargs: 其他传递给适配器的参数 """ name = adapter.get_name() if name in self._adapters: logger.opt(colors=True).debug( f'Adapter "{escape_tag(name)}" already exists' ) return self._adapters[name] = adapter(self, **kwargs) logger.opt(colors=True).debug( f'Succeeded to load adapter "{escape_tag(name)}"' ) @property @abc.abstractmethod def type(self) -> str: """驱动类型名称""" raise NotImplementedError @property @abc.abstractmethod def logger(self): """驱动专属 logger 日志记录器""" raise NotImplementedError @abc.abstractmethod def run(self, *args, **kwargs): """ 启动驱动框架 :参数: *args **kwargs """ logger.opt(colors=True).debug( f"Loaded adapters: {escape_tag(', '.join(self._adapters))}" ) @abc.abstractmethod def on_startup(self, func: Callable) -> Callable: """注册一个在驱动启动时运行的函数""" raise NotImplementedError @abc.abstractmethod def on_shutdown(self, func: Callable) -> Callable: """注册一个在驱动停止时运行的函数""" raise NotImplementedError def on_bot_connect(self, func: T_BotConnectionHook) -> T_BotConnectionHook: """ 装饰一个函数使他在 bot 通过 WebSocket 连接成功时执行。 :函数参数: bot: 当前连接上的 Bot 对象 """ self._bot_connection_hook.add(func) return func def on_bot_disconnect(self, func: T_BotDisconnectionHook) -> T_BotDisconnectionHook: """ 装饰一个函数使他在 bot 通过 WebSocket 连接断开时执行。 :函数参数: bot: 当前连接上的 Bot 对象 """ self._bot_disconnection_hook.add(func) return func def _bot_connect(self, bot: "Bot") -> None: """在 WebSocket 连接成功后,调用该函数来注册 bot 对象""" if bot.self_id in self._clients: raise RuntimeError(f"Duplicate bot connection with id {bot.self_id}") self._clients[bot.self_id] = bot async def _run_hook(bot: "Bot") -> None: coros = list(map(lambda x: x(bot), self._bot_connection_hook)) if coros: try: await asyncio.gather(*coros) except Exception as e: logger.opt(colors=True, exception=e).error( "Error when running WebSocketConnection hook. " "Running cancelled!" ) asyncio.create_task(_run_hook(bot)) def _bot_disconnect(self, bot: "Bot") -> None: """在 WebSocket 连接断开后,调用该函数来注销 bot 对象""" if bot.self_id in self._clients: del self._clients[bot.self_id] async def _run_hook(bot: "Bot") -> None: coros = list(map(lambda x: x(bot), self._bot_disconnection_hook)) if coros: try: await asyncio.gather(*coros) except Exception as e: logger.opt(colors=True, exception=e).error( "Error when running WebSocketDisConnection hook. " "Running cancelled!" ) asyncio.create_task(_run_hook(bot)) class ForwardMixin(abc.ABC): @property @abc.abstractmethod def type(self) -> str: raise NotImplementedError @abc.abstractmethod async def request(self, setup: Request) -> Response: raise NotImplementedError @abc.abstractmethod @asynccontextmanager async def websocket(self, setup: Request) -> AsyncGenerator[WebSocket, None]: raise NotImplementedError yield # used for static type checking's generator detection class ForwardDriver(Driver, ForwardMixin): """ Forward Driver 基类。将客户端框架封装,以满足适配器使用。 """ class ReverseDriver(Driver): """ Reverse Driver 基类。将后端框架封装,以满足适配器使用。 """ @property @abc.abstractmethod def server_app(self) -> Any: """驱动 APP 对象""" raise NotImplementedError @property @abc.abstractmethod def asgi(self) -> Any: """驱动 ASGI 对象""" raise NotImplementedError @abc.abstractmethod def setup_http_server(self, setup: "HTTPServerSetup") -> None: raise NotImplementedError @abc.abstractmethod def setup_websocket_server(self, setup: "WebSocketServerSetup") -> None: raise NotImplementedError def combine_driver(driver: Type[Driver], *mixins: Type[ForwardMixin]) -> Type[Driver]: # check first assert issubclass(driver, Driver), "`driver` must be subclass of Driver" assert all( map(lambda m: issubclass(m, ForwardMixin), mixins) ), "`mixins` must be subclass of ForwardMixin" class CombinedDriver(*mixins, driver, ForwardDriver): # type: ignore @property def type(self) -> str: return ( driver.type.__get__(self) + "+" + "+".join(map(lambda x: x.type.__get__(self), mixins)) ) return CombinedDriver @dataclass class HTTPServerSetup: path: URL # path should not be absolute, check it by URL.is_absolute() == False method: str name: str handle_func: Callable[[Request], Awaitable[Response]] @dataclass class WebSocketServerSetup: path: URL # path should not be absolute, check it by URL.is_absolute() == False name: str handle_func: Callable[[WebSocket], Awaitable[Any]]