依赖注入
在事件处理流程中,事件响应器具有自己独立的上下文,例如:当前的事件、机器人等信息。在 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 依赖注入支持重载(即:可以标注参数为子类型)且具有重载优先检查权。
- Python 3.10+
- Python 3.9
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): ... # 兼容性处理
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 依赖注入。
Event 依赖注入支持重载(即:可以标注参数为子类型)且具有重载优先检查权。
- Python 3.10+
- Python 3.9
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): ... # 兼容性处理
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
获取当前会话状态。
通过标注参数为 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 中可用。
通过标注参数为异常类型,或者一系列异常类型,即可获取到事件响应器运行中抛出的异常。
- Python 3.10+
- Python 3.9
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): ...
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 对象,将会在依赖注入期间执行。我们来看一个例子:
- Use Annotated
- Without Annotated
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)]):
...
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: 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。此时在事件处理流程中,多次使用同一个子依赖时,将会使用缓存中的结果而不会重复执行。这在很多情景中非常有用,例如:
- Use Annotated
- Without Annotated
import random
from typing import Annotated