ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2025-09-07 04:26:45 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

websocket api

Browse Source
This commit is contained in:
yanyongyu
2020-08-13 15:23:04 +08:00
parent 0e73d4ce20
commit e7f9b2c229
10 changed files with 141 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
nonebot/drivers/__init__.py
View File

@ -5,16 +5,21 @@ import abc
from ipaddress import IPv4Address
from nonebot.config import Env, Config
from nonebot.typing import Bot, Dict, Optional, Callable
from nonebot.typing import Bot, Dict, Type, Optional, Callable
class BaseDriver(abc.ABC):
_adapters: Dict[str, Type[Bot]] = {}
@abc.abstractmethod
def __init__(self, env: Env, config: Config):
self.env = env.environment
self.config = config
self._clients: Dict[int, Bot] = {}
self._clients: Dict[str, Bot] = {}
@classmethod
def register_adapter(cls, name: str, adapter: Type[Bot]):
cls._adapters[name] = adapter
@property
@abc.abstractmethod
@ -32,7 +37,7 @@ class BaseDriver(abc.ABC):
raise NotImplementedError
@property
def bots(self) -> Dict[int, Bot]:
def bots(self) -> Dict[str, Bot]:
return self._clients
@abc.abstractmethod

12
nonebot/drivers/fastapi.py
View File

@ -106,14 +106,15 @@ class Driver(BaseDriver):
adapter: str,
response: Response,
data: dict = Body(...),
x_self_id: int = Header(None),
x_self_id: str = Header(None),
access_token: str = OAuth2PasswordBearer(
"/", auto_error=False)):
# TODO: Check authorization
# Create Bot Object
if adapter == "cqhttp":
bot = CQBot("http", self.config, x_self_id)
if adapter in self._adapters:
BotClass = self._adapters[adapter]
bot = BotClass(self, "http", self.config, x_self_id)
else:
response.status_code = status.HTTP_404_NOT_FOUND
return {"status": 404, "message": "adapter not found"}
@ -125,7 +126,7 @@ class Driver(BaseDriver):
async def _handle_ws_reverse(self,
adapter: str,
websocket: FastAPIWebSocket,
x_self_id: int = Header(None),
x_self_id: str = Header(None),
access_token: str = OAuth2PasswordBearer(
"/", auto_error=False)):
websocket = WebSocket(websocket)
@ -134,7 +135,8 @@ class Driver(BaseDriver):
# Create Bot Object
if adapter == "coolq":
bot = CQBot("websocket",
bot = CQBot(self,
"websocket",
self.config,
x_self_id,
websocket=websocket)