iamai.adapter.utils module#

适配器实用工具。

这里定义了一些在编写适配器时常用的基类，适配器开发者可以直接继承自这里的类或者用作参考。

class iamai.adapter.utils.HttpClientAdapter(bot: Bot)[source]#

Bases: PollingAdapter[EventT, ConfigT]

HTTP 客户端适配器示例。

abstract async on_tick() → None[source]#: 当轮询发生。

session: ClientSession#

async shutdown() → None[source]#: 关闭并清理连接。

async startup() → None[source]#: 初始化适配器。

class iamai.adapter.utils.HttpServerAdapter(bot: Bot)[source]#

Bases: Adapter[EventT, ConfigT]

HTTP 服务端适配器示例。

app: Application#

get_url: str#

abstract async handle_response(request: Request) → StreamResponse[source]#: 处理响应。

host: str#

port: int#

post_url: str#

async run() → None[source]#: 运行适配器。

runner: AppRunner#

async shutdown() → None[source]#: 关闭并清理连接。

site: TCPSite#

async startup() → None[source]#: 初始化适配器。

class iamai.adapter.utils.PollingAdapter(bot: Bot)[source]#

Bases: Adapter[EventT, ConfigT]

轮询式适配器示例。

create_task: bool = False#

delay: float = 0.1#

abstract async on_tick() → None[source]#: 当轮询发生。

async run() → None[source]#: 运行适配器。

class iamai.adapter.utils.WebSocketAdapter(bot: Bot)[source]#

Bases: Adapter[EventT, ConfigT]

WebSocket 适配器示例。

同时支持 WebSocket 客户端和服务端。

adapter_type: Literal['ws', 'reverse-ws']#

app: Application | None#

async handle_reverse_ws_response(request: Request) → WebSocketResponse[source]#: 处理 aiohttp WebSocket 服务器的接收。

async handle_websocket() → None[source]#: 处理 WebSocket。

abstract async handle_websocket_msg(msg: WSMessage) → None[source]#: 处理 WebSocket 消息。

host: str#

port: int#

reconnect_interval: int = 3#

async reverse_ws_connection_hook() → None[source]#: 反向 WebSocket 连接建立时的钩子函数。

async run() → None[source]#: 运行适配器。

runner: AppRunner | None#

session: ClientSession | None#

async shutdown() → None[source]#: 关闭并清理连接。

site: TCPSite | None#

async startup() → None[source]#: 初始化适配器。

url: str#

websocket: WebSocketResponse | ClientWebSocketResponse | None = None#

async websocket_connect() → None[source]#: 创建正向 WebSocket 连接。

class iamai.adapter.utils.WebSocketClientAdapter(bot: Bot)[source]#

Bases: Adapter[EventT, ConfigT]

WebSocket 客户端适配器示例。

abstract async handle_response(msg: WSMessage) → None[source]#: 处理响应。

async run() → None[source]#: 运行适配器。

url: str#

class iamai.adapter.utils.WebSocketServerAdapter(bot: Bot)[source]#

Bases: Adapter[EventT, ConfigT]

WebSocket 服务端适配器示例。

app: Application#

async handle_response(request: Request) → WebSocketResponse[source]#: 处理 WebSocket。

abstract async handle_ws_response(msg: WSMessage) → None[source]#: 处理 WebSocket 响应。

host: str#

port: int#

async run() → None[source]#: 运行适配器。

runner: AppRunner#

async shutdown() → None[source]#: 关闭并清理连接。

site: TCPSite#

async startup() → None[source]#: 初始化适配器。

url: str#

websocket: WebSocketResponse#