--- sidebar_position: 6 description: 通过依赖注入获取上下文信息 options: menu: weight: 70 category: advanced --- # 依赖注入 import Tabs from "@theme/Tabs"; import TabItem from "@theme/TabItem"; 在事件处理流程中,事件响应器具有自己独立的上下文,例如:当前的事件、机器人等信息。在 NoneBot 中,这些信息通过依赖注入的方式提供给事件处理函数,可以让代码更加整洁可读、提升复用能力。 在了解如何使用依赖注入获取上下文信息之前,我们需要先了解两个概念: - `Dependent`:使用依赖注入的函数或其他任意可调用对象。如:事件处理函数、自定义的依赖函数等。 - `Dependency`:依赖注入的对象。如:当前事件、机器人等。 在之前的文档中,我们已经多次使用了依赖注入来获取事件信息。通过对函数参数依照一定规则填写类型注解,即可获得想要的上下文信息。任何一个事件处理函数在添加到事件处理流程时,都会根据一定规则提前将其解析成一个 `Dependent` 对象,方便运行时进行注入。如果遇到无法解析的参数,将会抛出 `ValueError("Unknown parameter")` 的异常。整个依赖注入系统可以分为两部分: - 参数解析 - 依据一定规则解析函数参数,识别 `Dependency` 依赖。 - 生成 `Dependent` 对象。 - 执行 - 根据已经解析的 `Dependency` 依赖,执行调用。 - 将所有 `Dependency` 的返回值根据参数名传入并调用 `Dependent` 。 :::danger 警告 在依赖注入中,类型注解是非常重要的,因为它不仅可以决定依赖注入的对象,还可以触发[重载机制](../appendices/overload.md#重载)。如果类型注解与实际获得数据类型不一致,将会跳过当前 `Dependent` 对象(即事件处理函数)。 ::: :::tip 提示 如果对于依赖注入的解析流程有疑问,可以调整[日志等级配置项](../appendices/config.mdx#log-level)为 `TRACE`,查看依赖解析日志。 ::: ## 同步支持 对于依赖注入系统中的 `Dependent` 或者 `Dependency` 对象,均支持同步类型的函数或可调用对象。例如: ```python {6,10} 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)): ... ``` ## 非依赖参数 在依赖注入解析中,任何无法解析的参数如果带有默认值,将会被视为非依赖参数。这些参数在依赖运行时将不会被注入而使用函数默认值。例如: ```python async def _(foo: str = "bar"): ... ``` ## 类型依赖注入 这一类的依赖注入仅需要在函数参数中添加对应的类型注解即可。 ### Bot 获取当前事件的 Bot 对象。 通过标注参数为 `Bot` 类型,或者一系列 `Bot` 类型,即可获取到当前事件的 Bot 对象。为兼容性考虑,如果参数名为 `bot` 且无类型注解,也会视为 `Bot` 依赖注入。 ```python 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): ... # 兼容性处理 ``` ```python from typing import Union 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: Union[ConsoleBot, OneBotV11Bot]): ... async def _(bot): ... # 兼容性处理 ``` ### Event 获取当前事件。 通过标注参数为 `Event` 类型,或者一系列 `Event` 类型,即可获取到当前事件。为兼容性考虑,如果参数名为 `event` 且无类型注解,也会视为 `Event` 依赖注入。 ```python 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): ... # 兼容性处理 ``` ```python from typing import Union from nonebot.adapters import Event from nonebot.adapters.onebot.v11 import PrivateMessageEvent, GroupMessageEvent async def _(foo: Event): ... async def _(foo: Union[PrivateMessageEvent, GroupMessageEvent]): ... async def _(event): ... # 兼容性处理 ``` ### State 获取当前[会话状态](../appendices/session-state.md)。 ```python from nonebot.typing import T_State async def _(foo: T_State): ... ``` ### Matcher 获取当前事件响应器实例。常用于使用[事件响应器操作](../appendices/session-control.mdx)。 ```python from nonebot.matcher import Matcher async def _(matcher: Matcher): ... ``` ### Exception 获取事件响应器运行中抛出的异常。该依赖注入目前仅在事件响应器运行后处理 Hook 中可用。 通过标注参数为异常类型,或者一系列异常类型,即可获取到事件响应器运行中抛出的异常。 ```python {5,8} 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): ... ``` ```python {6,9} from typing import Union from nonebot.message import run_postprocessor from nonebot.exception import ActionFailed, NetworkError @run_postprocessor async def _(e: Exception): ... @run_postprocessor async def _(e: Union[ActionFailed, NetworkError]): ... ``` ## 子依赖 在依赖注入系统中,我们可以定义一个子依赖,来执行自定义的操作,提高代码复用性以及处理性能。 ### 定义子依赖 子依赖使用 `Depends` 标记进行定义,其参数即依赖的函数或可调用对象,同样会被解析为 `Dependent` 对象,将会在依赖注入期间执行。我们来看一个例子: ```python {4,16} from typing import Annotated from nonebot import on_command from nonebot.params import Depends from nonebot.matcher import Matcher from nonebot.adapters.console import MessageEvent test = on_command("test") async def check(event: MessageEvent, matcher: Matcher) -> MessageEvent: if event.get_user_id() in BLACKLIST: await matcher.finish() return event @test.handle() async def _(event: Annotated[MessageEvent, Depends(check)]): ... ``` ```python {2,14} from nonebot import on_command from nonebot.params import Depends from nonebot.matcher import Matcher from nonebot.adapters.console import MessageEvent test = on_command("test") async def check(event: MessageEvent, matcher: Matcher) -> MessageEvent: if event.get_user_id() in BLACKLIST: await matcher.finish() return event @test.handle() async def _(event: MessageEvent = Depends(check)): ... ``` 在上面的代码中,我们使用 `Depends` 标记定义了一个子依赖 `check`。它判断事件主体用户是否在黑名单中,如果在,则直接结束事件处理流程。如果不在,则返回事件对象,以便事件处理函数可以继续执行。 通过将 `Depends` 包裹的子依赖作为参数的默认值,我们就可以在执行事件处理函数之前执行子依赖,并将其返回值作为参数传入事件处理函数。子依赖和普通的事件处理函数并没有区别,同样可以使用依赖注入,并且可以返回任何类型的值。但需要注意的是,如果事件处理函数参数的类型注解与子依赖返回值的类型**不一致**,将会触发[重载](../appendices/overload.md)而跳过当前事件处理函数。 ### 依赖缓存 NoneBot 在执行子依赖时,会将其返回值缓存起来。当我们在使用子依赖时,`Depends` 具有一个参数 `use_cache`,默认为 `True`。此时在事件处理流程中,多次使用同一个子依赖时,将会使用缓存中的结果而不会重复执行。这在很多情景中非常有用,例如: ```python {7} 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) ``` ```python {6} import random async def random_result() -> int: return random.randint(1, 100) async def _(x: int = Depends(random_result)): print(x) ``` 此时,在同一事件处理流程中,这个随机函数的返回值将会保持一致。如果我们希望每次都重新执行子依赖,可以将 `use_cache` 设置为 `False`。 ```python {7} 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) ``` ```python {6} import random async def random_result() -> int: return random.randint(1, 100) async def _(x: int = Depends(random_result, use_cache=False)): print(x) ``` :::tip 提示 缓存的生命周期与当前接收到的事件相同。接收到事件后,子依赖在首次执行时缓存,在该事件处理完成后,缓存就会被清除。 ::: ### 类作为依赖 在前面的事例中,我们使用了函数作为子依赖。实际上,我们还可以使用类作为依赖。当我们在实例化一个类的时候,其实我们就在调用它,类本身也是一个可调用对象。例如: ```python {16} 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) ``` ```python {15} 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: ClassDependency = Depends(ClassDependency)): print(data.event, data.context) ``` 可以看到,我们使用 `dataclass` 定义了一个类。由于这个类的 `__init__` 方法可以被依赖注入系统解析,因此,我们可以将其作为子依赖进行声明。特别地,对于类依赖,`Depends` 的参数可以为空,NoneBot 将会使用参数的类型注解进行解析与推断: ```python from typing import Annotated async def _(data: Annotated[ClassDependency, Depends()]): print(data.event, data.context) ``` ```python async def _(data: ClassDependency = Depends()): print(data.event, data.context) ``` ### 生成器作为依赖 NoneBot 的依赖注入支持依赖项在事件处理流程结束后进行一些额外的工作,比如数据库 session 或者网络 IO 的关闭,互斥锁的解锁等等。同时,由于[依赖缓存](#依赖缓存)的存在,我们可以通过这种方式来实现共享一个 session 等功能。 要实现上述功能,我们可以用生成器函数作为依赖项,我们用 `yield` 关键字取代 `return` 关键字,并在 `yield` 之后进行额外的工作。 我们可以看下述代码段, 使用 `httpx.AsyncClient` 异步网络 IO,并在事件处理流程中共用一个 client: ```python {15} from typing import Annotated, 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") ``` ```python {15} from typing 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: httpx.AsyncClient = Depends(get_client)): resp = await x.get("https://nonebot.dev") ``` :::warning 注意 生成器作为依赖时,其中只能进行一次 `yield`,否则将会触发异常。如果对此有疑问并想探究原因,可以参考 [contextmanager](https://docs.python.org/zh-cn/3/library/contextlib.html#contextlib.contextmanager) 和 [asynccontextmanager](https://docs.python.org/zh-cn/3/library/contextlib.html#contextlib.asynccontextmanager) 文档。事实上,NoneBot 内部就使用了这两个装饰器。 ::: ### 可调用对象作为依赖 在 Python 里,为类定义 `__call__` 方法就可以使得这个类的实例成为一个可调用对象。因此,我们也可以将定义了 `__call__` 方法的类的实例作为依赖。事实上,NoneBot 的[内置响应规则](./matcher.md#内置响应规则)就广泛使用了这种方式,以 `is_type` 规则为例: ```python from typing import 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 获取当前事件的类型。 ```python {4} from typing import Annotated from nonebot.params import EventType async def _(foo: Annotated[str, EventType()]): ... ``` ```python {3} from nonebot.params import EventType async def _(foo: str = EventType()): ... ``` ### EventMessage 获取当前事件的消息。 ```python {5} from typing import Annotated from nonebot.adapters import Message from nonebot.params import EventMessage async def _(foo: Annotated[Message, EventMessage()]): ... ``` ```python {4} from nonebot.adapters import Message from nonebot.params import EventMessage async def _(foo: Message = EventMessage()): ... ``` ### EventPlainText 获取当前事件的消息纯文本部分。 ```python {4} from typing import Annotated from nonebot.params import EventPlainText async def _(foo: Annotated[str, EventPlainText()]): ... ``` ```python {3} from nonebot.params import EventPlainText async def _(foo: str = EventPlainText()): ... ``` ### EventToMe 获取当前事件是否与机器人相关。 ```python {4} from typing import Annotated from nonebot.params import EventToMe async def _(foo: Annotated[bool, EventToMe()]): ... ``` ```python {3} from nonebot.params import EventToMe async def _(foo: bool = EventToMe()): ... ``` ### Command 获取当前命令型消息的元组形式命令名。 ```python {4} from typing import Annotated from nonebot.params import Command async def _(foo: Annotated[tuple[str, ...], Command()]): ... ``` ```python {4} from typing import Tuple from nonebot.params import Command async def _(foo: Tuple[str, ...] = Command()): ... ``` :::tip 提示 命令详情只能在**触发命令型事件响应器时**获取。如果在事件处理后续流程中获取,则会获取到不同的值。 ::: ### RawCommand 获取当前命令型消息的文本形式命令名。 ```python {4} from typing import Annotated from nonebot.params import RawCommand async def _(foo: Annotated[str, RawCommand()]): ... ``` ```python {3} from nonebot.params import RawCommand async def _(foo: str = RawCommand()): ... ``` :::tip 提示 命令详情只能在**触发命令型事件响应器时**获取。如果在事件处理后续流程中获取,则会获取到不同的值。 ::: ### CommandArg 获取命令型消息命令后跟随的参数。 ```python {5} from typing import Annotated from nonebot.adapters import Message from nonebot.params import CommandArg async def _(foo: Annotated[Message, CommandArg()]): ... ``` ```python {4} from nonebot.adapters import Message from nonebot.params import CommandArg async def _(foo: Message = CommandArg()): ... ``` :::tip 提示 命令详情只能在**触发命令型事件响应器时**获取。如果在事件处理后续流程中获取,则会获取到不同的值。 ::: ### CommandStart 获取命令型消息命令前缀。 ```python {4} from typing import Annotated from nonebot.params import CommandStart async def _(foo: Annotated[str, CommandStart()]): ... ``` ```python {3} from nonebot.params import CommandStart async def _(foo: str = CommandStart()): ... ``` :::tip 提示 命令详情只能在**触发命令型事件响应器时**获取。如果在事件处理后续流程中获取,则会获取到不同的值。 ::: ### CommandWhitespace 获取命令型消息命令与参数间空白符。 ```python {4} from typing import Annotated from nonebot.params import CommandWhitespace async def _(foo: Annotated[str, CommandWhitespace()]): ... ``` ```python {3} from nonebot.params import CommandWhitespace async def _(foo: str = CommandWhitespace()): ... ``` :::tip 提示 命令详情只能在**触发命令型事件响应器时**获取。如果在事件处理后续流程中获取,则会获取到不同的值。 ::: ### ShellCommandArgv 获取 shell 命令解析前的参数列表,列表中可能包含文本字符串和富文本消息段(如:图片)。 ```python {4} from typing import Annotated from nonebot.params import ShellCommandArgs async def _(foo: Annotated[list[str | MessageSegment], ShellCommandArgv()]): ... ``` ```python {4} from typing import Union, Annotated from nonebot.params import ShellCommandArgs async def _(foo: Annotated[list[Union[str, MessageSegment]], ShellCommandArgv()]): ... ``` ```python {4} from typing import List, Union from nonebot.params import ShellCommandArgs async def _(foo: List[Union[str, MessageSegment]] = ShellCommandArgv()): ... ``` ### ShellCommandArgs 获取 shell 命令解析后的参数 Namespace,支持 MessageSegment 富文本(如:图片)。 :::tip 提示 如果参数解析成功,则为 parser 返回的 Namespace;如果参数解析失败,则为 [`ParserExit`](../api/exception.md#ParserExit) 异常,并携带错误码与错误信息。通过重载机制即可处理两种不同的情况。 由于 `ArgumentParser` 在解析到 `--help` 参数时也会抛出异常,这种情况下错误码为 `0` 且错误信息即为帮助信息。 ::: ```python {14,22} 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) ``` ```python {12,20} 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: ParserExit = ShellCommandArgs()): if foo.status == 0: foo.message # help message else: foo.message # error message # 解析成功 @matcher.handle() async def _(foo: Namespace = ShellCommandArgs()): arg_dict = vars(foo) ``` ### RegexStr 获取正则匹配结果的文本。 ```python {4} from typing import Annotated from nonebot.params import RegexStr async def _(foo: Annotated[str, RegexStr()]): ... ``` ```python {3} from nonebot.params import RegexStr async def _(foo: str = RegexStr()): ... ``` ### RegexGroup 获取正则匹配结果的 group 元组。 ```python {4} from typing import Any, Annotated from nonebot.params import RegexGroup async def _(foo: Annotated[tuple[Any, ...], RegexGroup()]): ... ``` ```python {4} from typing import Tuple, Any from nonebot.params import RegexGroup async def _(foo: Tuple[Any, ...] = RegexGroup()): ... ``` ### RegexDict 获取正则匹配结果的 group 字典。 ```python {4} from typing import Any, Annotated from nonebot.params import RegexDict async def _(foo: Annotated[dict[str, Any], RegexDict()]): ... ``` ```python {4} from typing import Any, Dict from nonebot.params import RegexDict async def _(foo: Dict[str, Any] = RegexDict()): ... ``` ### Startswith 获取触发响应器的消息前缀字符串。 ```python {4} from typing import Annotated from nonebot.params import Startswith async def _(foo: Annotated[str, Startswith()]): ... ``` ```python {3} from nonebot.params import Startswith async def _(foo: str = Startswith()): ... ``` ### Endswith 获取触发响应器的消息后缀字符串。 ```python {4} from typing import Annotated from nonebot.params import Endswith async def _(foo: Annotated[str, Endswith()]): ... ``` ```python {3} from nonebot.params import Endswith async def _(foo: str = Endswith()): ... ``` ### Fullmatch 获取触发响应器的消息字符串。 ```python {4} from typing import Annotated from nonebot.params import Fullmatch async def _(foo: Annotated[str, Fullmatch()]): ... ``` ```python {3} from nonebot.params import Fullmatch async def _(foo: str = Fullmatch()): ... ``` ### Keyword 获取触发响应器的关键字字符串。 ```python {4} from typing import Annotated from nonebot.params import Keyword async def _(foo: Annotated[str, Keyword()]): ... ``` ```python {3} from nonebot.params import Keyword async def _(foo: str = Keyword()): ... ``` ### Received 获取某次 `receive` 接收的事件。 ```python {7} from typing import Annotated from nonebot.adapters import Event from nonebot.params import Received @matcher.receive("id") async def _(foo: Annotated[Event, Received("id")]): ... ``` ```python {5} from nonebot.adapters import Event from nonebot.params import Received @matcher.receive("id") async def _(foo: Event = Received("id")): ... ``` ### LastReceived 获取最近一次 `receive` 接收的事件。 ```python {7} from typing import Annotated from nonebot.adapters import Event from nonebot.params import LastReceived @matcher.receive("any") async def _(foo: Annotated[Event, LastReceived()]): ... ``` ```python {5} from nonebot.adapters import Event from nonebot.params import LastReceived @matcher.receive("any") async def _(foo: Event = LastReceived()): ... ``` ### Arg 获取某次 `got` 接收的参数。如果 `Arg` 参数留空,则使用函数的参数名作为要获取的参数。 ```python {7,8} 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")]): ... ``` ```python {5,6} from nonebot.params import Arg from nonebot.adapters import Message @matcher.got("key") async def _(key: Message = Arg()): ... async def _(foo: Message = Arg("key")): ... ``` ### ArgStr 获取某次 `got` 接收的参数,并转换为字符串。如果 `Arg` 参数留空,则使用函数的参数名作为要获取的参数。 ```python {6,7} from typing import Annotated from nonebot.params import ArgStr @matcher.got("key") async def _(key: str = ArgStr()): ... async def _(foo: str = ArgStr("key")): ... ``` ```python {4,5} from nonebot.params import ArgStr @matcher.got("key") async def _(key: Annotated[str, ArgStr()]): ... async def _(foo: Annotated[str, ArgStr("key")]): ... ``` ### ArgPlainText 获取某次 `got` 接收的参数的纯文本部分。如果 `Arg` 参数留空,则使用函数的参数名作为要获取的参数。 ```python {6,7} 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")]): ... ``` ```python {4,5} from nonebot.params import ArgPlainText @matcher.got("key") async def _(key: str = ArgPlainText()): ... async def _(foo: str = ArgPlainText("key")): ... ```