mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-02 20:14:36 +00:00
87 lines
2.3 KiB
Python
87 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import abc
|
|
import typing
|
|
import quart
|
|
from quart.typing import RouteCallable
|
|
|
|
from ....core import app
|
|
|
|
|
|
preregistered_groups: list[type[RouterGroup]] = []
|
|
"""RouterGroup 的预注册列表"""
|
|
|
|
def group_class(name: str, path: str) -> None:
|
|
"""注册一个 RouterGroup"""
|
|
|
|
def decorator(cls: typing.Type[RouterGroup]) -> typing.Type[RouterGroup]:
|
|
cls.name = name
|
|
cls.path = path
|
|
preregistered_groups.append(cls)
|
|
return cls
|
|
|
|
return decorator
|
|
|
|
|
|
class RouterGroup(abc.ABC):
|
|
|
|
name: str
|
|
|
|
path: str
|
|
|
|
ap: app.Application
|
|
|
|
quart_app: quart.Quart
|
|
|
|
def __init__(self, ap: app.Application, quart_app: quart.Quart) -> None:
|
|
self.ap = ap
|
|
self.quart_app = quart_app
|
|
|
|
@abc.abstractmethod
|
|
async def initialize(self) -> None:
|
|
pass
|
|
|
|
def route(self, rule: str, **options: typing.Any) -> typing.Callable[[RouteCallable], RouteCallable]: # decorator
|
|
"""注册一个路由"""
|
|
def decorator(f: RouteCallable) -> RouteCallable:
|
|
nonlocal rule
|
|
rule = self.path + rule
|
|
|
|
async def handler_error(*args, **kwargs):
|
|
try:
|
|
return await f(*args, **kwargs)
|
|
except Exception as e: # 自动 500
|
|
return self.http_status(500, -2, str(e))
|
|
|
|
new_f = handler_error
|
|
new_f.__name__ = (self.name + rule).replace('/', '__')
|
|
new_f.__doc__ = f.__doc__
|
|
|
|
self.quart_app.route(rule, **options)(new_f)
|
|
return f
|
|
|
|
return decorator
|
|
|
|
def _cors(self, response: quart.Response) -> quart.Response:
|
|
return response
|
|
|
|
def success(self, data: typing.Any = None) -> quart.Response:
|
|
"""返回一个 200 响应"""
|
|
return self._cors(quart.jsonify({
|
|
'code': 0,
|
|
'msg': 'ok',
|
|
'data': data,
|
|
}))
|
|
|
|
def fail(self, code: int, msg: str) -> quart.Response:
|
|
"""返回一个异常响应"""
|
|
|
|
return self._cors(quart.jsonify({
|
|
'code': code,
|
|
'msg': msg,
|
|
}))
|
|
|
|
def http_status(self, status: int, code: int, msg: str) -> quart.Response:
|
|
"""返回一个指定状态码的响应"""
|
|
return self.fail(code, msg), status
|