mirror of
https://github.com/nonebot/nonebot2.git
synced 2025-07-28 08:41:29 +00:00
💡 add driver docstring
This commit is contained in:
@ -27,12 +27,36 @@ class BaseDriver(abc.ABC):
|
||||
|
||||
@abc.abstractmethod
|
||||
def __init__(self, env: Env, config: Config):
|
||||
"""
|
||||
:参数:
|
||||
* ``env: Env``: 包含环境信息的 Env 对象
|
||||
* ``config: Config``: 包含配置信息的 Config 对象
|
||||
"""
|
||||
self.env = env.environment
|
||||
"""
|
||||
:类型: ``str``
|
||||
:说明: 环境名称
|
||||
"""
|
||||
self.config = config
|
||||
"""
|
||||
:类型: ``Config``
|
||||
:说明: 配置对象
|
||||
"""
|
||||
self._clients: Dict[str, Bot] = {}
|
||||
"""
|
||||
:类型: ``Dict[str, Bot]``
|
||||
:说明: 已连接的 Bot
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def register_adapter(cls, name: str, adapter: Type[Bot]):
|
||||
"""
|
||||
:说明:
|
||||
注册一个协议适配器
|
||||
:参数:
|
||||
* ``name: str``: 适配器名称,用于在连接时进行识别
|
||||
* ``adapter: Type[Bot]``: 适配器 Class
|
||||
"""
|
||||
cls._adapters[name] = adapter
|
||||
logger.opt(
|
||||
colors=True).debug(f'Succeeded to load adapter "<y>{name}</y>"')
|
||||
@ -40,33 +64,43 @@ class BaseDriver(abc.ABC):
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def type(self):
|
||||
"""驱动类型名称"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def server_app(self):
|
||||
"""驱动 APP 对象"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def asgi(self):
|
||||
"""驱动 ASGI 对象"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def logger(self):
|
||||
"""驱动专属 logger 日志记录器"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def bots(self) -> Dict[str, Bot]:
|
||||
"""
|
||||
:类型: ``Dict[str, Bot]``
|
||||
:说明: 获取当前所有已连接的 Bot
|
||||
"""
|
||||
return self._clients
|
||||
|
||||
@abc.abstractmethod
|
||||
def on_startup(self, func: Callable) -> Callable:
|
||||
"""注册一个在驱动启动时运行的函数"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def on_shutdown(self, func: Callable) -> Callable:
|
||||
"""注册一个在驱动停止时运行的函数"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
@ -75,44 +109,69 @@ class BaseDriver(abc.ABC):
|
||||
port: Optional[int] = None,
|
||||
*args,
|
||||
**kwargs):
|
||||
"""
|
||||
:说明:
|
||||
启动驱动框架
|
||||
:参数:
|
||||
* ``host: Optional[str]``: 驱动绑定 IP
|
||||
* ``post: Optional[int]``: 驱动绑定端口
|
||||
* ``*args``
|
||||
* ``**kwargs``
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
async def _handle_http(self):
|
||||
"""用于处理 HTTP 类型请求的函数"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
async def _handle_ws_reverse(self):
|
||||
"""用于处理 WebSocket 类型请求的函数"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class BaseWebSocket(object):
|
||||
"""WebSocket 连接封装,统一接口方便外部调用。"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def __init__(self, websocket):
|
||||
"""
|
||||
:参数:
|
||||
* ``websocket: Any``: WebSocket 连接对象
|
||||
"""
|
||||
self._websocket = websocket
|
||||
|
||||
@property
|
||||
def websocket(self):
|
||||
"""WebSocket 连接对象"""
|
||||
return self._websocket
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def closed(self):
|
||||
"""
|
||||
:类型: ``bool``
|
||||
:说明: 连接是否已经关闭
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
async def accept(self):
|
||||
"""接受 WebSocket 连接请求"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
async def close(self, code: int):
|
||||
"""关闭 WebSocket 连接请求"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
async def receive(self) -> dict:
|
||||
"""接收一条 WebSocket 信息"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
async def send(self, data: dict):
|
||||
"""发送一条 WebSocket 信息"""
|
||||
raise NotImplementedError
|
||||
|
@ -1,5 +1,14 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
FastAPI 驱动适配
|
||||
================
|
||||
|
||||
后端使用方法请参考: `FastAPI 文档`_
|
||||
|
||||
.. _FastAPI 文档:
|
||||
https://fastapi.tiangolo.com/
|
||||
"""
|
||||
|
||||
import hmac
|
||||
import json
|
||||
@ -31,6 +40,7 @@ def get_auth_bearer(access_token: Optional[str] = Header(
|
||||
|
||||
|
||||
class Driver(BaseDriver):
|
||||
"""FastAPI 驱动框架"""
|
||||
|
||||
def __init__(self, env: Env, config: Config):
|
||||
super().__init__(env, config)
|
||||
@ -50,29 +60,35 @@ class Driver(BaseDriver):
|
||||
@property
|
||||
@overrides(BaseDriver)
|
||||
def type(self) -> str:
|
||||
"""驱动名称: ``fastapi``"""
|
||||
return "fastapi"
|
||||
|
||||
@property
|
||||
@overrides(BaseDriver)
|
||||
def server_app(self) -> FastAPI:
|
||||
"""``FastAPI APP`` 对象"""
|
||||
return self._server_app
|
||||
|
||||
@property
|
||||
@overrides(BaseDriver)
|
||||
def asgi(self):
|
||||
"""``FastAPI APP`` 对象"""
|
||||
return self._server_app
|
||||
|
||||
@property
|
||||
@overrides(BaseDriver)
|
||||
def logger(self) -> logging.Logger:
|
||||
"""fastapi 使用的 logger"""
|
||||
return logging.getLogger("fastapi")
|
||||
|
||||
@overrides(BaseDriver)
|
||||
def on_startup(self, func: Callable) -> Callable:
|
||||
"""参考文档: `Events <https://fastapi.tiangolo.com/advanced/events/#startup-event>`_"""
|
||||
return self.server_app.on_event("startup")(func)
|
||||
|
||||
@overrides(BaseDriver)
|
||||
def on_shutdown(self, func: Callable) -> Callable:
|
||||
"""参考文档: `Events <https://fastapi.tiangolo.com/advanced/events/#startup-event>`_"""
|
||||
return self.server_app.on_event("shutdown")(func)
|
||||
|
||||
@overrides(BaseDriver)
|
||||
@ -82,6 +98,7 @@ class Driver(BaseDriver):
|
||||
*,
|
||||
app: Optional[str] = None,
|
||||
**kwargs):
|
||||
"""使用 ``uvicorn`` 启动 FastAPI"""
|
||||
LOGGING_CONFIG = {
|
||||
"version": 1,
|
||||
"disable_existing_loggers": False,
|
||||
|
Reference in New Issue
Block a user