NoneBot
Theme
  • 进阶
  • 依赖注入
Version: 2.4.0

依赖注入

在事件处理流程中，事件响应器具有自己独立的上下文，例如：当前的事件、机器人等信息。在 NoneBot 中，这些信息通过依赖注入的方式提供给事件处理函数，可以让代码更加整洁可读、提升复用能力。

在了解如何使用依赖注入获取上下文信息之前，我们需要先了解两个概念：

  • Dependent：使用依赖注入的函数或其他任意可调用对象。如：事件处理函数、自定义的依赖函数等。
  • Dependency：依赖注入的对象。如：当前事件、机器人等。

在之前的文档中，我们已经多次使用了依赖注入来获取事件信息。通过对函数参数依照一定规则填写类型注解，即可获得想要的上下文信息。任何一个事件处理函数在添加到事件处理流程时，都会根据一定规则提前将其解析成一个 Dependent 对象，方便运行时进行注入。如果遇到无法解析的参数，将会抛出 ValueError("Unknown parameter") 的异常。整个依赖注入系统可以分为两部分：

  • 参数解析
    • 依据一定规则解析函数参数，识别 Dependency 依赖。
    • 生成 Dependent 对象。
  • 执行
    • 根据已经解析的 Dependency 依赖，执行调用。
    • 将所有 Dependency 的返回值根据参数名传入并调用 Dependent 。
警告

在依赖注入中，类型注解是非常重要的，因为它不仅可以决定依赖注入的对象，还可以触发重载机制。如果类型注解与实际获得数据类型不一致，将会跳过当前 Dependent 对象（即事件处理函数）。

提示

如果对于依赖注入的解析流程有疑问，可以调整日志等级配置项为 TRACE，查看依赖解析日志。

同步支持​

对于依赖注入系统中的 Dependent 或者 Dependency 对象，均支持同步类型的函数或可调用对象。例如：

from nonebot import on_command
from nonebot.params import Depends

matcher = on_command("foo")

def dependency() -> str:
    return "something"

@matcher.handle()
def _(result: str = Depends(dependency)):
    ...

非依赖参数​

在依赖注入解析中，任何无法解析的参数如果带有默认值，将会被视为非依赖参数。这些参数在依赖运行时将不会被注入而使用函数默认值。例如：

async def _(foo: str = "bar"): ...

类型依赖注入​

这一类的依赖注入仅需要在函数参数中添加对应的类型注解即可。

Bot​

获取当前事件的 Bot 对象。

通过标注参数为 Bot 类型，或者一系列 Bot 类型，即可获取到当前事件的 Bot 对象。为兼容性考虑，如果参数名为 bot 且无类型注解，也会视为 Bot 依赖注入。

Bot 依赖注入支持重载（即：可以标注参数为子类型）且具有重载优先检查权。

from nonebot.adapters import Bot
from nonebot.adapters.console import Bot as ConsoleBot
from nonebot.adapters.onebot.v11 import Bot as OneBotV11Bot

async def _(foo: Bot): ...
async def _(foo: ConsoleBot | OneBotV11Bot): ...
async def _(bot): ...  # 兼容性处理

Event​

获取当前事件。

通过标注参数为 Event 类型，或者一系列 Event 类型，即可获取到当前事件。为兼容性考虑，如果参数名为 event 且无类型注解，也会视为 Event 依赖注入。

Event 依赖注入支持重载（即：可以标注参数为子类型）且具有重载优先检查权。

from nonebot.adapters import Event
from nonebot.adapters.onebot.v11 import PrivateMessageEvent, GroupMessageEvent

async def _(foo: Event): ...
async def _(foo: PrivateMessageEvent | GroupMessageEvent): ...
async def _(event): ...  # 兼容性处理

State​

获取当前会话状态。

通过标注参数为 T_State 类型，即可获取到当前会话状态。为兼容性考虑，如果参数名为 state 且无类型注解，也会视为 State 依赖注入。

from nonebot.typing import T_State

async def _(foo: T_State): ...

Matcher​

获取当前事件响应器实例。常用于使用事件响应器操作。

通过标注参数为 Matcher 类型，或者一系列 Matcher 类型，即可获取到当前事件。为兼容性考虑，如果参数名为 matcher 且无类型注解，也会视为 Matcher 依赖注入。

Matcher 依赖注入支持重载（即：可以标注参数为子类型）且具有重载优先检查权。

from nonebot.matcher import Matcher

async def _(foo: Matcher): ...
async def _(matcher): ...  # 兼容性处理

Exception​

获取事件响应器运行中抛出的异常。该依赖注入目前仅在事件响应器运行后处理 Hook 中可用。

通过标注参数为异常类型，或者一系列异常类型，即可获取到事件响应器运行中抛出的异常。

from nonebot.message import run_postprocessor
from nonebot.exception import ActionFailed, NetworkError

@run_postprocessor
async def _(e: Exception): ...

@run_postprocessor
async def _(e: ActionFailed | NetworkError): ...

子依赖​

在依赖注入系统中，我们可以定义一个子依赖，来执行自定义的操作，提高代码复用性以及处理性能。

定义子依赖​

子依赖使用 Depends 标记进行定义，其参数即依赖的函数或可调用对象，同样会被解析为 Dependent 对象，将会在依赖注入期间执行。我们来看一个例子：

from typing import Annotated

from nonebot import on_command
from nonebot.adapters import Event
from nonebot.params import Depends

test = on_command("test")

async def check(event: Event) -> Event:
    if event.get_user_id() in BLACKLIST:
        await test.finish()
    return event

@test.handle()
async def _(event: Annotated[Event, Depends(check)]):
    ...

在上面的代码中，我们使用 Depends 标记定义了一个子依赖 check。它判断事件主体用户是否在黑名单中，如果在，则直接结束事件处理流程。如果不在，则返回事件对象，以便事件处理函数可以继续执行。

通过将 Depends 包裹的子依赖作为参数的默认值，我们就可以在执行事件处理函数之前执行子依赖，并将其返回值作为参数传入事件处理函数。子依赖和普通的事件处理函数并没有区别，同样可以使用依赖注入，并且可以返回任何类型的值。但需要注意的是，如果事件处理函数参数的类型注解与子依赖返回值的类型不一致，将会触发重载而跳过当前事件处理函数。

特别的，我们可以为 Dependent 对象定义一系列前置子依赖，它们会在参数执行前被顺序执行，且返回值将会被忽略，例如：

from nonebot import on_command
from nonebot.adapters import Event
from nonebot.params import Depends

test = on_command("test")

async def check(event: Event):
    if event.get_user_id() in BLACKLIST:
        await test.finish()

@test.handle(parameterless=[Depends(check)])
async def _():
    ...

依赖缓存​

NoneBot 在执行子依赖时，会将其返回值缓存起来。当我们在使用子依赖时，Depends 具有一个参数 use_cache，默认为 True。此时在事件处理流程中，多次使用同一个子依赖时，将会使用缓存中的结果而不会重复执行。这在很多情景中非常有用，例如：

import random
from typing import Annotated

async def random_result() -> int:
    return random.randint(1, 100)

async def _(x: Annotated[int, Depends(random_result)]):
    print(x)

此时，在同一事件处理流程中，这个随机函数的返回值将会保持一致。如果我们希望每次都重新执行子依赖，可以将 use_cache 设置为 False。

import random
from typing import Annotated

async def random_result() -> int:
    return random.randint(1, 100)

async def _(x: Annotated[int, Depends(random_result, use_cache=False)]):
    print(x)
提示

缓存的生命周期与当前接收到的事件相同。接收到事件后，子依赖在首次执行时缓存，在该事件处理完成后，缓存就会被清除。

类型转换与校验​

在依赖注入系统中，我们可以对子依赖的返回值进行自动类型转换与校验。这个功能由 Pydantic 支持，因此我们通过参数类型注解自动使用 Pydantic 支持的类型转换。例如：

from typing import Annotated

from nonebot.params import Depends
from nonebot.adapters import Event

def get_user_id(event: Event) -> str:
    return event.get_user_id()

async def _(user_id: Annotated[int, Depends(get_user_id, validate=True)]):
    print(user_id)

在进行类型自动转换的同时，Pydantic 还支持对数据进行更多的限制，如：大于、小于、长度等。使用方法如下：

from typing import Annotated

from pydantic import Field
from nonebot.params import Depends
from nonebot.adapters import Event

def get_user_id(event: Event) -> str:
    return event.get_user_id()

async def _(user_id: Annotated[int, Depends(get_user_id, validate=Field(gt=100))]):
    print(user_id)

类作为依赖​

在前面的事例中，我们使用了函数作为子依赖。实际上，我们还可以使用类作为依赖。当我们在实例化一个类的时候，其实我们就在调用它，类本身也是一个可调用对象。例如：

from typing import Annotated
from dataclasses import dataclass

from nonebot.params import Depends
from nonebot.adapters import Event
from nonebot.typing import T_State

def get_context(state: T_State) -> dict:
    return state.setdefault("context", {})

@dataclass
class ClassDependency:
    event: Event
    context: dict = Depends(get_context)

async def _(data: Annotated[ClassDependency, Depends(ClassDependency)]):
    print(data.event, data.context)

可以看到，我们使用 dataclass 定义了一个类。由于这个类的 __init__ 方法可以被依赖注入系统解析，因此，我们可以将其作为子依赖进行声明。特别地，对于类依赖，Depends 的参数可以为空，NoneBot 将会使用参数的类型注解进行解析与推断：

from typing import Annotated

async def _(data: Annotated[ClassDependency, Depends()]):
    print(data.event, data.context)

生成器作为依赖​

NoneBot 的依赖注入支持依赖项在事件处理流程结束后进行一些额外的工作，比如数据库 session 或者网络 IO 的关闭，互斥锁的解锁等等。同时，由于依赖缓存的存在，我们可以通过这种方式来实现共享一个 session 等功能。

要实现上述功能，我们可以用生成器函数作为依赖项，我们用 yield 关键字取代 return 关键字，并在 yield 之后进行额外的工作。

我们可以看下述代码段, 使用 httpx.AsyncClient 异步网络 IO，并在事件处理流程中共用一个 client：

from typing import Annotated
from collections.abc import AsyncGenerator

import httpx
from nonebot.params import Depends

async def get_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    try:
        async with httpx.AsyncClient() as client:
            yield client
    finally:
        # 在这里进行额外的工作


@test.handle()
async def _(x: Annotated[httpx.AsyncClient, Depends(get_client)]):
    resp = await x.get("https://nonebot.dev")
注意

生成器作为依赖时，其中只能进行一次 yield，否则将会触发异常。如果对此有疑问并想探究原因，可以参考 contextmanager 和 asynccontextmanager 文档。事实上，NoneBot 内部就使用了这两个装饰器。

可调用对象作为依赖​

在 Python 里，为类定义 __call__ 方法就可以使得这个类的实例成为一个可调用对象。因此，我们也可以将定义了 __call__ 方法的类的实例作为依赖。事实上，NoneBot 的内置响应规则就广泛使用了这种方式，以 is_type 规则为例：

from nonebot.adapters import Event

class IsTypeRule:
    def __init__(self, *types: type[Event]):
        self.types = types

    async def __call__(self, event: Event) -> bool:
        return isinstance(event, self.types)

我们在使用 is_type 时，即实例化了 IsTypeRule 类，然后将实例作为响应规则依赖项传入。

其他依赖注入​

这一类的依赖注入通常基于子依赖编写，为我们开发者提供更方便的途径获取上下文信息。

EventType​

获取当前事件的类型。

from typing import Annotated
from nonebot.params import EventType

async def _(foo: Annotated[str, EventType()]): ...

EventMessage​

获取当前事件的消息。

from typing import Annotated
from nonebot.adapters import Message
from nonebot.params import EventMessage

async def _(foo: Annotated[Message, EventMessage()]): ...

EventPlainText​

获取当前事件的消息纯文本部分。

from typing import Annotated
from nonebot.params import EventPlainText

async def _(foo: Annotated[str, EventPlainText()]): ...

EventToMe​

获取当前事件是否与机器人相关。

from typing import Annotated
from nonebot.params import EventToMe

async def _(foo: Annotated[bool, EventToMe()]): ...

Command​

获取当前命令型消息的元组形式命令名。

from typing import Annotated
from nonebot.params import Command

async def _(foo: Annotated[tuple[str, ...], Command()]): ...
提示

命令详情只能在触发命令型事件响应器时获取。如果在事件处理后续流程中获取，则会获取到不同的值。

RawCommand​

获取当前命令型消息的文本形式命令名。

from typing import Annotated
from nonebot.params import RawCommand

async def _(foo: Annotated[str, RawCommand()]): ...
提示

命令详情只能在触发命令型事件响应器时获取。如果在事件处理后续流程中获取，则会获取到不同的值。

CommandArg​

获取命令型消息命令后跟随的参数。

from typing import Annotated
from nonebot.adapters import Message
from nonebot.params import CommandArg

async def _(foo: Annotated[Message, CommandArg()]): ...
提示

命令详情只能在触发命令型事件响应器时获取。如果在事件处理后续流程中获取，则会获取到不同的值。

CommandStart​

获取命令型消息命令前缀。

from typing import Annotated
from nonebot.params import CommandStart

async def _(foo: Annotated[str, CommandStart()]): ...
提示

命令详情只能在触发命令型事件响应器时获取。如果在事件处理后续流程中获取，则会获取到不同的值。

CommandWhitespace​

获取命令型消息命令与参数间空白符。

from typing import Annotated
from nonebot.params import CommandWhitespace

async def _(foo: Annotated[str, CommandWhitespace()]): ...
提示

命令详情只能在触发命令型事件响应器时获取。如果在事件处理后续流程中获取，则会获取到不同的值。

ShellCommandArgv​

获取 shell 命令解析前的参数列表，列表中可能包含文本字符串和富文本消息段（如：图片）。

from typing import Annotated
from nonebot.params import ShellCommandArgs

async def _(foo: Annotated[list[str | MessageSegment], ShellCommandArgv()]): ...
from nonebot.params import ShellCommandArgs

async def _(foo: list[str | MessageSegment] = ShellCommandArgv()): ...

ShellCommandArgs​

获取 shell 命令解析后的参数 Namespace，支持 MessageSegment 富文本（如：图片）。

提示

如果参数解析成功，则为 parser 返回的 Namespace；如果参数解析失败，则为 ParserExit 异常，并携带错误码与错误信息。通过重载机制即可处理两种不同的情况。

由于 ArgumentParser 在解析到 --help 参数时也会抛出异常，这种情况下错误码为 0 且错误信息即为帮助信息。

from typing import Annotated

from nonebot import on_shell_command
from nonebot.exception import ParserExit
from nonebot.params import ShellCommandArgs
from nonebot.rule import Namespace, ArgumentParser

parser = ArgumentParser("demo")
# parser.add_argument ...
matcher = on_shell_command("cmd", parser=parser)

# 解析失败
@matcher.handle()
async def _(foo: Annotated[ParserExit, ShellCommandArgs()]):
    if foo.status == 0:
        foo.message  # help message
    else:
        foo.message  # error message

# 解析成功
@matcher.handle()
async def _(foo: Annotated[Namespace, ShellCommandArgs()]):
    arg_dict = vars(foo)

RegexMatched​

获取正则匹配结果的对象。

from re import Match
from typing import Annotated
from nonebot.params import RegexMatched

async def _(foo: Annotated[Match[str], RegexMatched()]): ...

RegexStr​

获取正则匹配结果的文本。

from typing import Annotated
from nonebot.params import RegexStr

async def _(foo: Annotated[str, RegexStr()]): ...

RegexGroup​

获取正则匹配结果的 group 元组。

from typing import Any, Annotated
from nonebot.params import RegexGroup

async def _(foo: Annotated[tuple[Any, ...], RegexGroup()]): ...

RegexDict​

获取正则匹配结果的 group 字典。

from typing import Any, Annotated
from nonebot.params import RegexDict

async def _(foo: Annotated[dict[str, Any], RegexDict()]): ...

Startswith​

获取触发响应器的消息前缀字符串。

from typing import Annotated
from nonebot.params import Startswith

async def _(foo: Annotated[str, Startswith()]): ...

Endswith​

获取触发响应器的消息后缀字符串。

from typing import Annotated
from nonebot.params import Endswith

async def _(foo: Annotated[str, Endswith()]): ...

Fullmatch​

获取触发响应器的消息字符串。

from typing import Annotated
from nonebot.params import Fullmatch

async def _(foo: Annotated[str, Fullmatch()]): ...

Keyword​

获取触发响应器的关键字字符串。

from typing import Annotated
from nonebot.params import Keyword

async def _(foo: Annotated[str, Keyword()]): ...

Received​

获取某次 receive 接收的事件。

from typing import Annotated

from nonebot.adapters import Event
from nonebot.params import Received

@matcher.receive("id")
async def _(foo: Annotated[Event, Received("id")]): ...

LastReceived​

获取最近一次 receive 接收的事件。

from typing import Annotated

from nonebot.adapters import Event
from nonebot.params import LastReceived

@matcher.receive("any")
async def _(foo: Annotated[Event, LastReceived()]): ...

Arg​

获取某次 got 接收的参数。如果 Arg 参数留空，则使用函数的参数名作为要获取的参数。

from typing import Annotated

from nonebot.params import Arg
from nonebot.adapters import Message

@matcher.got("key")
async def _(key: Annotated[Message, Arg()]): ...
async def _(foo: Annotated[Message, Arg("key")]): ...

ArgStr​

获取某次 got 接收的参数，并转换为字符串。如果 Arg 参数留空，则使用函数的参数名作为要获取的参数。

from typing import Annotated

from nonebot.params import ArgStr

@matcher.got("key")
async def _(key: Annotated[str, ArgStr()]): ...
async def _(foo: Annotated[str, ArgStr("key")]): ...

ArgPlainText​

获取某次 got 接收的参数的纯文本部分。如果 Arg 参数留空，则使用函数的参数名作为要获取的参数。

from typing import Annotated

from nonebot.params import ArgPlainText

@matcher.got("key")
async def _(key: Annotated[str, ArgPlainText()]): ...
async def _(foo: Annotated[str, ArgPlainText("key")]): ...