跳转至

connection

本模块实现了 OneBot Connect 的 HTTP 和 HTTP Webhook 部分 以及连接基类的定义。

ClientConnection

Bases: Connection

客户端连接（HTTP Webhook/反向 WebSocket）基类。

Attributes:

Name Type Description
access_token str | None

访问令牌
url str

OneBot 应用的连接目标地址
ua str

连接时使用的 User-Agent
Source code in src/pylibob/connection.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
class ClientConnection(Connection):
    """客户端连接（HTTP Webhook/反向 WebSocket）基类。

    Attributes:
        access_token (str | None): 访问令牌
        url (str): OneBot 应用的连接目标地址
        ua (str): 连接时使用的 User-Agent
    """

    def __init__(
        self,
        url: str,
        *,
        access_token: str | None = None,
    ) -> None:
        """初始化客户端连接。

        Args:
            access_token (str | None): 访问令牌
            url (str): OneBot 应用的连接目标地址
        """
        super().__init__(access_token=access_token)
        self.url = url

    @property
    def ua(self) -> str:
        """连接时使用的 User-Agent。"""
        return (
            f"OneBot/{self.impl.onebot_version} "
            f"pylibob/{__version__} "
            f"{self.impl.name}/{self.impl.version}"
        )

ua: str property

连接时使用的 User-Agent。

__init__(url, *, access_token=None)

初始化客户端连接。

Parameters:

Name Type Description Default
access_token str | None

访问令牌

 None
url str

OneBot 应用的连接目标地址

 required
Source code in src/pylibob/connection.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def __init__(
    self,
    url: str,
    *,
    access_token: str | None = None,
) -> None:
    """初始化客户端连接。

    Args:
        access_token (str | None): 访问令牌
        url (str): OneBot 应用的连接目标地址
    """
    super().__init__(access_token=access_token)
    self.url = url

Connection

连接基类。

init_connection 方法为初始化连接的方法，用于在 impl 确定后继续初始化。 顺序如下: 

graph LR
    A[__init__] --> B[确定 impl] --> C[init_connection]

run_action 用于以原始数据运行动作响应器。

emit_event 用于向应用端推送事件，需各连接自行实现。

Attributes:

Name Type Description
access_token str | None

访问令牌
impl OneBotImpl

OneBot 实现实例
Source code in src/pylibob/connection.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class Connection:
    """连接基类。

    `init_connection` 方法为初始化连接的方法，用于在 `impl` 确定后继续初始化。
    顺序如下:
    ``` mermaid
    graph LR
        A[__init__] --> B[确定 impl] --> C[init_connection]
    ```

    `run_action` 用于以原始数据运行动作响应器。

    `emit_event` 用于向应用端推送事件，需各连接自行实现。

    Attributes:
        access_token (str | None): 访问令牌
        impl (OneBotImpl): OneBot 实现实例
    """

    def __init__(
        self,
        *,
        access_token: str | None = None,
    ) -> None:
        """初始化连接。

        Args:
            access_token (str | None): 访问令牌
        """
        self.access_token = access_token
        self._impl: "OneBotImpl" | None = None
        super().__init__()

    @property
    def impl(self) -> "OneBotImpl":
        """OneBot 实现实例。`"""
        if self._impl is None:
            raise ValueError("OneBotImpl is not initialed")
        return self._impl

    async def run_action(
        self,
        data: dict[str, Any],
    ) -> ActionResponse:
        """以原始数据运行动作响应器。

        - 未传入 `action` 或 `params` 时返回 10001 Bad Request。

        Args:
            data (dict[str, Any]): 原始数据
        """
        if not (action := data.get("action")):
            return FailedActionResponse(
                retcode=BAD_REQUEST,
                message="`action` is not exist.",
            )
        if (params := data.get("params")) is None:
            return FailedActionResponse(
                retcode=BAD_REQUEST,
                message="`params` is not exist.",
            )
        echo = data.get("echo")
        bot_self: BotSelf | None = data.get("self")
        return await self.impl.handle_action(action, params, bot_self, echo)

    def init_connection(self) -> None:
        """初始化连接。"""

    async def emit_event(self, event: Event) -> None:
        """向应用端推送事件。

        需要各连接自行实现。

        Args:
            event (Event): 事件
        """
        raise NotImplementedError

impl: 'OneBotImpl' property

OneBot 实现实例。`

__init__(*, access_token=None)

初始化连接。

Parameters:

Name Type Description Default
access_token str | None

访问令牌

 None
Source code in src/pylibob/connection.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    access_token: str | None = None,
) -> None:
    """初始化连接。

    Args:
        access_token (str | None): 访问令牌
    """
    self.access_token = access_token
    self._impl: "OneBotImpl" | None = None
    super().__init__()

emit_event(event) async

向应用端推送事件。

需要各连接自行实现。

Parameters:

Name Type Description Default
event Event

事件

 required
Source code in src/pylibob/connection.py 
109
110
111
112
113
114
115
116
117
async def emit_event(self, event: Event) -> None:
    """向应用端推送事件。

    需要各连接自行实现。

    Args:
        event (Event): 事件
    """
    raise NotImplementedError

init_connection()

初始化连接。

Source code in src/pylibob/connection.py 
106
107
def init_connection(self) -> None:
    """初始化连接。"""

run_action(data) async

以原始数据运行动作响应器。

  • 未传入 action 或 params 时返回 10001 Bad Request。

Parameters:

Name Type Description Default
data dict[str, Any]

原始数据

 required
Source code in src/pylibob/connection.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def run_action(
    self,
    data: dict[str, Any],
) -> ActionResponse:
    """以原始数据运行动作响应器。

    - 未传入 `action` 或 `params` 时返回 10001 Bad Request。

    Args:
        data (dict[str, Any]): 原始数据
    """
    if not (action := data.get("action")):
        return FailedActionResponse(
            retcode=BAD_REQUEST,
            message="`action` is not exist.",
        )
    if (params := data.get("params")) is None:
        return FailedActionResponse(
            retcode=BAD_REQUEST,
            message="`params` is not exist.",
        )
    echo = data.get("echo")
    bot_self: BotSelf | None = data.get("self")
    return await self.impl.handle_action(action, params, bot_self, echo)

HTTP

Bases: ServerConnection

HTTP 连接。

当启用 get_latest_events 元动作时，连接会创建一个大小为 event_buffer_size 的事件队列 event_queue。

Attributes:

Name Type Description
access_token str | None

访问令牌
host str

HTTP 服务器监听 IP
port int

HTTP 服务器监听端口
event_queue Queue[Event] | None

事件队列
Source code in src/pylibob/connection.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class HTTP(ServerConnection):
    """[HTTP 连接](https://12.onebot.dev/connect/communication/http/)。

    当启用 `get_latest_events` 元动作时，连接会创建一个大小为 `event_buffer_size` 的事件队列 `event_queue`。

    Attributes:
        access_token (str | None): 访问令牌
        host (str): HTTP 服务器监听 IP
        port (int): HTTP 服务器监听端口
        event_queue (Queue[Event] | None): 事件队列
    """  # noqa: E501

    def __init__(
        self,
        *,
        access_token: str | None = None,
        host: str = "0.0.0.0",
        port: int = 8080,
        event_enabled: bool = True,
        event_buffer_size: int = 20,
    ) -> None:
        """初始化 HTTP 连接。

        Args:
            access_token (str | None): 访问令牌
            host (str): HTTP 服务器监听 IP
            port (int): HTTP 服务器监听端口
            event_enabled (bool): 是否启用 `get_latest_events` 元动作
            event_buffer_size (int): 事件缓冲区大小
        """
        self.logger = logging.getLogger("pylibob.connection.http")

        super().__init__(access_token=access_token, host=host, port=port)
        self.event_queue: Queue[Event] | None = (
            Queue(maxsize=event_buffer_size) if event_enabled else None
        )

    async def receive_http_request(self, request: Request) -> Response:
        # 鉴权
        if not authorize(self.access_token, request):
            # 如果鉴权失败，必须返回 HTTP 状态码 401 Unauthorized
            self.logger.warning(f"{request.url} 鉴权失败")
            return Response(status_code=HTTP_401_UNAUTHORIZED)
        # 检查 Content-Type
        if not (
            content_type := detect_content_type(
                request.headers.get("Content-Type") or "",
            )
        ):
            self.logger.warning(
                f"{request.url} Content-Type 不为 application/json "
                "或 application/msgpack",
            )
            return Response(status_code=HTTP_415_UNSUPPORTED_MEDIA_TYPE)

        body = await request.body()
        if content_type == ContentType.JSON:
            try:
                data = json.loads(body)
            except json.JSONDecodeError:
                return JSONResponse(
                    msgspec.to_builtins(
                        FailedActionResponse(
                            retcode=BAD_REQUEST,
                            message="Invalid JSON",
                        ),
                    ),
                )
        else:
            try:
                data = msgpack.unpackb(body)
            except msgpack.UnpackException:
                return Response(
                    msgspec.msgpack.encode(
                        FailedActionResponse(
                            retcode=BAD_REQUEST,
                            message="Invalid MessagePack",
                        ),
                    ),
                )
        self.logger.info(
            f"[RECEIVE({content_type.name}) <= {request.url}] {data}",
        )
        resp = await self.run_action(data)
        convert_func = (
            msgspec.json.encode
            if content_type == ContentType.JSON
            else msgspec.msgpack.encode
        )
        return Response(
            convert_func(resp),
            headers={"Content-Type": content_type.value},
        )

    async def action_get_latest_events(self, limit: int = 0, timeout: int = 0):
        """[获取最新事件列表](https://12.onebot.dev/interface/meta/actions/#get_latest_events)"""
        # TODO: long polling
        assert self.event_queue
        times = 1
        events = []
        while not self.event_queue.empty() and (limit == 0 or times <= limit):
            events.append((await self.event_queue.get()).dict())
            times += 1
        return events

    def init_connection(self) -> None:
        asgi_app.add_route(
            "/",
            self.receive_http_request,
            ["POST"],
            f"{self.impl.name}-{self.impl.version}-http",
            False,
        )
        if self.event_queue:
            self.logger.info("启用元动作 get_latest_events")
            self.impl.register_action_handler(
                "get_latest_events",
                self.action_get_latest_events,
            )

    async def emit_event(self, event: Event) -> None:
        if self.event_queue:
            if self.event_queue.full():
                self.event_queue.get_nowait()  # pop the oldest event
            self.event_queue.put_nowait(event)

__init__(*, access_token=None, host='0.0.0.0', port=8080, event_enabled=True, event_buffer_size=20)

初始化 HTTP 连接。

Parameters:

Name Type Description Default
access_token str | None

访问令牌

 None
host str

HTTP 服务器监听 IP

 '0.0.0.0'
port int

HTTP 服务器监听端口

 8080
event_enabled bool

是否启用 get_latest_events 元动作

 True
event_buffer_size int

事件缓冲区大小

 20
Source code in src/pylibob/connection.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def __init__(
    self,
    *,
    access_token: str | None = None,
    host: str = "0.0.0.0",
    port: int = 8080,
    event_enabled: bool = True,
    event_buffer_size: int = 20,
) -> None:
    """初始化 HTTP 连接。

    Args:
        access_token (str | None): 访问令牌
        host (str): HTTP 服务器监听 IP
        port (int): HTTP 服务器监听端口
        event_enabled (bool): 是否启用 `get_latest_events` 元动作
        event_buffer_size (int): 事件缓冲区大小
    """
    self.logger = logging.getLogger("pylibob.connection.http")

    super().__init__(access_token=access_token, host=host, port=port)
    self.event_queue: Queue[Event] | None = (
        Queue(maxsize=event_buffer_size) if event_enabled else None
    )

action_get_latest_events(limit=0, timeout=0) async

获取最新事件列表

Source code in src/pylibob/connection.py 
244
245
246
247
248
249
250
251
252
253
async def action_get_latest_events(self, limit: int = 0, timeout: int = 0):
    """[获取最新事件列表](https://12.onebot.dev/interface/meta/actions/#get_latest_events)"""
    # TODO: long polling
    assert self.event_queue
    times = 1
    events = []
    while not self.event_queue.empty() and (limit == 0 or times <= limit):
        events.append((await self.event_queue.get()).dict())
        times += 1
    return events

HTTPWebhook

Bases: ClientConnection

HTTP Webhook 连接。

Attributes:

Name Type Description
access_token str | None

访问令牌
url str

Webhook 上报地址
timeout int

上报请求超时时间，单位: 毫秒，0 表示不超时
Source code in src/pylibob/connection.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
class HTTPWebhook(ClientConnection):
    """[HTTP Webhook 连接](https://12.onebot.dev/connect/communication/http-webhook/)。

    Attributes:
        access_token (str | None): 访问令牌
        url (str): Webhook 上报地址
        timeout (int): 上报请求超时时间，单位: 毫秒，0 表示不超时
    """

    def __init__(
        self,
        url: str,
        *,
        access_token: str | None = None,
        timeout: int = 5000,
    ) -> None:
        """初始化 HTTP Webhook 连接。

        Args:
            access_token (str | None): 访问令牌
            url (str): Webhook 上报地址
            timeout (int): 上报请求超时时间，单位: 毫秒，0 表示不超时
        """
        super().__init__(url, access_token=access_token)
        self.timeout = timeout
        self.logger = logging.getLogger("pylibob.connection.http_webhook")

    def _make_header(self) -> dict[str, str]:
        header = {
            "Content-Type": "application/json",
            "User-Agent": self.ua,
            "X-OneBot-Version": self.impl.onebot_version,
            "X-Impl": self.impl.name,
        }
        if self.access_token:
            header["Authorization"] = f"Bearer {self.access_token}"
        return header

    async def emit_event(self, event: Event) -> None:
        async with ClientSession(
            headers=self._make_header(),
        ) as session:
            event_json = event.dict()
            self.logger.debug(f"[SEND => {self.url}] {event_json}")
            async with session.post(
                self.url,
                timeout=ClientTimeout(total=self.timeout / 1000),
                json=event_json,
            ) as resp:
                if resp.status == HTTP_204_NO_CONTENT:
                    # 如果响应状态码为 204 No Content，
                    # 应认为事件推送成功，并不做更多处理。
                    return
                if resp.status != HTTP_200_OK:
                    # 如果响应状态码不是 204 或 200 中的任一个，
                    # 应认为事件推送失败。
                    self.logger.warning(
                        f"事件推送失败: {resp.status} {resp.reason}",
                    )
                    return

                # 如果响应状态码为 200 OK，也应认为事件推送成功，
                # 此时应该根据响应头中的 Content-Type
                # 将响应体解析为动作请求列表，依次处理动作请求，丢弃动作响应。
                if not (
                    content_type := detect_content_type(
                        resp.headers.get("Content-Type") or "",
                    )
                ):
                    self.logger.warning(
                        "Content-Type 不为 application/json "
                        "或 application/msgpack",
                    )

                body = await resp.read()
                if content_type == ContentType.JSON:
                    try:
                        data = json.loads(body)
                    except json.JSONDecodeError:
                        return
                else:
                    try:
                        data = msgpack.unpackb(body)
                    except msgpack.UnpackException:
                        return
                self.logger.debug(f"[RECEIVE <= {self.url}] {data}")
                for action in data:
                    await self.run_action(action)

__init__(url, *, access_token=None, timeout=5000)

初始化 HTTP Webhook 连接。

Parameters:

Name Type Description Default
access_token str | None

访问令牌

 None
url str

Webhook 上报地址

 required
timeout int

上报请求超时时间，单位: 毫秒，0 表示不超时

 5000
Source code in src/pylibob/connection.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def __init__(
    self,
    url: str,
    *,
    access_token: str | None = None,
    timeout: int = 5000,
) -> None:
    """初始化 HTTP Webhook 连接。

    Args:
        access_token (str | None): 访问令牌
        url (str): Webhook 上报地址
        timeout (int): 上报请求超时时间，单位: 毫秒，0 表示不超时
    """
    super().__init__(url, access_token=access_token)
    self.timeout = timeout
    self.logger = logging.getLogger("pylibob.connection.http_webhook")

ServerConnection

Bases: Connection

服务器连接（HTTP/正向 WebSocket）基类。

服务器连接会由 ServerRunner 运行（uvicorn）。

Attributes:

Name Type Description
access_token str | None

访问令牌
host str

服务器监听 IP
port int

服务器监听端口
Source code in src/pylibob/connection.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class ServerConnection(Connection):
    """服务器连接（HTTP/正向 WebSocket）基类。

    服务器连接会由 ServerRunner 运行（uvicorn）。

    Attributes:
        access_token (str | None): 访问令牌
        host (str): 服务器监听 IP
        port (int): 服务器监听端口
    """

    def __init__(
        self,
        *,
        access_token: str | None = None,
        host: str = "0.0.0.0",
        port: int = 8080,
    ) -> None:
        """初始化服务器连接。

        Args:
            access_token (str | None): 访问令牌
            host (str): 服务器监听 IP
            port (int): 服务器监听端口
        """
        super().__init__(access_token=access_token)
        self.host = host
        self.port = port

__init__(*, access_token=None, host='0.0.0.0', port=8080)

初始化服务器连接。

Parameters:

Name Type Description Default
access_token str | None

访问令牌

 None
host str

服务器监听 IP

 '0.0.0.0'
port int

服务器监听端口

 8080
Source code in src/pylibob/connection.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def __init__(
    self,
    *,
    access_token: str | None = None,
    host: str = "0.0.0.0",
    port: int = 8080,
) -> None:
    """初始化服务器连接。

    Args:
        access_token (str | None): 访问令牌
        host (str): 服务器监听 IP
        port (int): 服务器监听端口
    """
    super().__init__(access_token=access_token)
    self.host = host
    self.port = port