Reference

Bridge

Bridge.get_bridge

@staticmethod
def get_bridge(app: Application) -> 'Bridge':
    """
    从 aiohttp 应用实例中获取 Bridge 实例。

    参数：
        app (Application): aiohttp 应用实例

    返回：
        Bridge: Bridge 实例
    """

Bridge.__init__

def __init__(self, config: str | None = None, app: Application | None = None) -> None:
    """
    config: path to config file/directory
    app: aiohttp application
    """

Bridge.scan

def scan(self, *packages) -> None:
    """
    扫描指定包或模块，自动注册路由、服务和中间件。

    参数：
        *packages: 要扫描的包或模块
    """

Bridge.middlewares

def middlewares(self, *middlewares) -> None:
    """
    注册中间件到应用程序。

    参数：
        *middlewares: Middleware 子类或中间件函数
    """

Bridge.run_app

def run_app(self, ready=True, **kwargs) -> None:
    """
    启动 Web 应用服务器。

    参数：
        ready (bool): 是否调用 ready() 方法，默认为 True
        **kwargs: 传递给 aiohttp.web.run_app 的其他参数
    """

Bridge.beans

def beans(self, *beans) -> None:
    """
    注册 Bean 工厂函数，用于依赖注入。

    参数：
        *beans: 返回 Bean 实例的工厂函数
    """

Bridge.ready

def ready(self) -> None:
    """
    准备应用程序运行，设置 aiojobs 和信号处理器。
    """

Bridge.dump_openapi_components

def dump_openapi_components(self) -> dict:
    """
    生成应用程序的 OpenAPI 组件定义。

    返回：
        dict: 包含 schemas 定义的 OpenAPI 组件字典
    """

load_module_config

def load_module_config(app: Application, module_config_key: str, module_config_cls: Type[T]) -> T:
    """
    加载指定模块的配置。

    参数：
        app (Application): aiohttp的应用对象，包含全局配置信息。
        module_config_key (str): 在全局配置中查找模块配置的键名。
        module_config_cls (Type[T]): 用于校验和解析模块配置的Pydantic模型类，必须是pydantic.BaseModel的子类。

    返回：
        T: 解析后的模块配置实例。

    异常：
        TypeError: 如果module_config_cls不是pydantic.BaseModel的子类，则抛出此异常。
    """

Module

class Module:
    """
    进程级别的单例模块，兼容aiohttp的signals
    """
    def __init__(self) -> None: ...
    async def on_startup(self, app: Application) -> None: ...
    async def on_cleanup(self, app: Application) -> None: ...
    async def on_shutdown(self, app: Application) -> None: ...
    def load_config(self, app: Application) -> Any:
        """
        Example of override:
        >>> def load_config(self, app: Application) -> Annotated[MyAppConfig, 'myapp']:
        ...   return lessweb.load_module_config(app, 'myapp', MyAppConfig)
        """

Middleware

class Middleware:
    """
    请求级别的单例中间件，兼容aiohttp的middlewares
    """
    def __init__(self) -> None: ...
    async def on_request(self, request: Request, handler: HANDLER_TYPE) -> Response:
        """
        Example of override:
        >>> async def on_request(self, request: Request, handler: HANDLER_TYPE) -> Response:
        ...   return await handler(request)
        """

Service

class Service:
    """
    业务逻辑单例服务
    """
    def __init__(self) -> None: ...

autowire_module

def autowire_module(app: Application, cls: Type[T]) -> T:
    """
    进程级别单例模块注入，自动处理依赖关系。

    参数：
        app (Application): aiohttp 应用实例
        cls (Type[T]): 要注入的模块类

    返回：
        T: 模块实例
    """

autowire

def autowire(request: Request, cls: Type[U]) -> U:
    """
    请求级别的依赖注入，自动实例化中间件或服务对象。

    参数：
        request (Request): aiohttp 请求对象
        cls (Type[U]): 要注入的类

    返回：
        U: 类的实例
    """

rest_error

def rest_error(
        error: Type[HTTPError],
        data,
        *,
        headers: Optional[LooseHeaders] = None,
        **kwargs,
) -> HTTPError:
    return error(
        body=TypeCast.dumps(data),
        headers=headers,
        content_type='application/json',
        **kwargs,
    )

rest_response

def rest_response(
        data,
        *,
        status: int = 200,
        reason: Optional[str] = None,
        headers: Optional[LooseHeaders] = None,
) -> Response:
    response = Response(
        body=TypeCast.dumps(data),
        status=status,
        reason=reason,
        headers=headers,
        content_type='application/json',
    )
    return response

get_request_stack

def get_request_stack(request: Request) -> list[bytes]:
    """
    获取请求对象中保存的请求堆栈。

    参数：
        request (Request): aiohttp 的请求对象，通过 request 的字典属性存放中间件等过程中传递的数据堆栈，
                           该堆栈用于在处理请求过程中存放中间转换或验证后的数据（例如经过转换后的请求体）。

    返回：
        list[bytes]: 返回请求堆栈列表。如果请求对象中不存在该堆栈，则会在请求中初始化一个空列表后返回。
    """

push_request_stack

def push_request_stack(request: Request, value: bytes):
    """
    向请求对象的请求堆栈中添加一个值。

    参数：
        request (Request): aiohttp 的请求对象，其内部包含用于存储请求处理过程数据的堆栈。
        value (bytes): 要添加到堆栈中的数据值，必须为 bytes 类型。

    使用示例：
        在一个Middleware中，可能先对原始请求体进行读取和预处理：

            async def some_middleware(request, handler):
                raw_data = await request.read()
                push_request_stack(request, raw_data)
                # 对读取到的数据做进一步转换（例如转换为 JSON 后修改数据），转换后的数据也必须是 bytes 类型
                modified_data = ...
                push_request_stack(request, modified_data)
                return await handler(request)

        在后续处理 endpoint 中，可以使用 positional-only 参数，框架通过调用 get_request_stack(request) 
        获取堆栈列表，并对堆栈进行 pop 操作，将最后一次推入的数据提取出来作为 positional-only 参数传递给目标处理函数。
    """

EventEmitter

class EventEmitter(Module):
    """
    事件发射器类，用于注册事件处理函数并触发事件。

    该类继承自 Module，在应用启动时会为所有标记为事件订阅（使用 OnEvent 装饰器）的处理函数构建 HTTP POST 路由，
    从而实现事件的统一管理。触发事件时，该类会模拟一个 HTTP POST 请求，将事件名称和负载数据传递给相应的处理函数，
    支持同步返回响应，也能将事件处理后台异步执行。
    """

EventEmitter.emit

async def emit(self, event: str, payload: Any) -> StreamResponse:
    """
    触发指定的事件。
    通过构造模拟的 HTTP POST 请求，将 payload 序列化为 JSON 数据传入对应的事件处理路由中，
    根据处理函数是否标记为后台异步执行，返回相应的响应（异步时返回 204 状态码）。
    """

TypeCast

TypeCast.init_orjson_option

@classmethod
def init_orjson_option(cls, option_text: str) -> None:
    """
    初始化 orjson 的序列化选项。

    参数：
        option_text (str): 逗号分隔的选项字符串，如 'INDENT_2,SORT_KEYS'
    """

TypeCast.dumps

@classmethod
def dumps(cls, data: Any) -> bytes:
    """
    将数据序列化为 JSON 字节串。

    参数：
        data (Any): 要序列化的数据

    返回：
        bytes: JSON 字节串
    """

TypeCast.validate_json

@classmethod
def validate_json(cls, data: Union[str, bytes], tp: Type[T]) -> T:
    """
    将 JSON 数据验证并转换为指定类型。

    支持的类型：
        - list[pydantic.BaseModel]
        - pydantic.BaseModel

    验证过程：
        - 如果 T 是 list[pydantic.BaseModel]，则逐个验证列表项
        - 如果 T 是 pydantic.BaseModel，则使用 pydantic 的 model_validate_json 方法

    参数：
        data (Union[str, bytes]): JSON 数据
        tp (Type[T]): 目标类型

    返回：
        T: 验证后的对象

    异常：
        ValueError: 数据无法解析为指定类型（4xx 错误）
        TypeError: 不支持的类型（5xx 错误）
    """

TypeCast.validate_query

@classmethod
def validate_query(cls, data: str, tp: Type[T]) -> T:
    """
    将查询字符串转换为指定类型。

    参数：
        data (str): 查询字符串
        tp (Type[T]): 目标类型

    返回：
        T: 转换后的值

    异常：
        ValueError: 数据无法转换为指定类型（4xx 错误）
        TypeError: 不支持的类型（5xx 错误）
    """

annotation.Endpoint

class Endpoint:
    """
    HTTP 端点注解基类。

    属性：
        method (HTTP_METHOD_TYPE): HTTP 方法
        path (str): 路由路径
    """

annotation.Get

class Get(Endpoint):
    """
    GET 请求端点注解。

    参数：
        path (str): 路由路径
    """

annotation.Post

class Post(Endpoint):
    """
    POST 请求端点注解。

    参数：
        path (str): 路由路径
    """

annotation.Put

class Put(Endpoint):
    """
    PUT 请求端点注解。

    参数：
        path (str): 路由路径
    """

annotation.Delete

class Delete(Endpoint):
    """
    DELETE 请求端点注解。

    参数：
        path (str): 路由路径
    """

annotation.Patch

class Patch(Endpoint):
    """
    PATCH 请求端点注解。

    参数：
        path (str): 路由路径
    """

annotation.DefaultFactory

class DefaultFactory:
    """
    DefaultFactory 用于为函数参数提供默认值生成器，通过传入一个无参可调用对象，
    当需要默认值时调用该工厂函数生成默认值。该类常用于注解中，用于描述参数的默认生成逻辑，
    类似于 dataclasses 中的 default_factory。

    示例用法：
    >>> async def foo(*, ids: Annotated(list[str], DefaultFactory(list)) -> Annotated[dict, Get('/')]:
    ...   pass

    注意：
    - factory_func 必须是一个不接受任何参数的可调用对象；
    - 在参数默认值处理中，当未提供具体值时，框架会调用该工厂函数返回默认值实例。
    """
    def __init__(self, factory_func) -> None: ...

annotation.OnEvent

class OnEvent:
    """
    OnEvent 是一个用于标记事件订阅的注解类。它允许开发者在定义endpoint函数时,
    标记函数为事件订阅，从而实现事件的统一管理。

    示例用法：
    async def foo_handle_event(request: Request) -> Annotated[dict, OnEvent('my_event_name')]:
        ...
    """
    def __init__(self, event: str, background: bool = False): ...

annotation.TextResponse

class TextResponse:
    """
    TextResponse 是一个用于指定 HTTP 响应内容类型的注解类。它允许开发者在定义endpoint函数时，
    明确指定返回的响应数据的 MIME 类型，默认为 'text/plain'。

    示例用法：
    async def text_endpoint() -> Annotated[str, TextResponse()]:
        return 'This is a text response.'
    """

annotation.Html

class Html(TextResponse):
    def __init__(self):
        super().__init__('text/html')

annotation.PlainText

class PlainText(TextResponse):
    """
    纯文本响应注解，内容类型为 'text/plain'。
    """
    def __init__(self):
        super().__init__('text/plain')