🚨 Develop: 添加 ruff linter (#2114)

This commit is contained in:
Ju4tCode
2023-06-24 14:47:35 +08:00
committed by GitHub
parent fe21cbfa1d
commit 3d5dd5969c
53 changed files with 813 additions and 758 deletions

View File

@ -24,12 +24,17 @@
- `load_all_plugins` => {ref}``load_all_plugins` <nonebot.plugin.load.load_all_plugins>`
- `load_from_json` => {ref}``load_from_json` <nonebot.plugin.load.load_from_json>`
- `load_from_toml` => {ref}``load_from_toml` <nonebot.plugin.load.load_from_toml>`
- `load_builtin_plugin` => {ref}``load_builtin_plugin` <nonebot.plugin.load.load_builtin_plugin>`
- `load_builtin_plugins` => {ref}``load_builtin_plugins` <nonebot.plugin.load.load_builtin_plugins>`
- `load_builtin_plugin` =>
{ref}``load_builtin_plugin` <nonebot.plugin.load.load_builtin_plugin>`
- `load_builtin_plugins` =>
{ref}``load_builtin_plugins` <nonebot.plugin.load.load_builtin_plugins>`
- `get_plugin` => {ref}``get_plugin` <nonebot.plugin.get_plugin>`
- `get_plugin_by_module_name` => {ref}``get_plugin_by_module_name` <nonebot.plugin.get_plugin_by_module_name>`
- `get_loaded_plugins` => {ref}``get_loaded_plugins` <nonebot.plugin.get_loaded_plugins>`
- `get_available_plugin_names` => {ref}``get_available_plugin_names` <nonebot.plugin.get_available_plugin_names>`
- `get_plugin_by_module_name` =>
{ref}``get_plugin_by_module_name` <nonebot.plugin.get_plugin_by_module_name>`
- `get_loaded_plugins` =>
{ref}``get_loaded_plugins` <nonebot.plugin.get_loaded_plugins>`
- `get_available_plugin_names` =>
{ref}``get_available_plugin_names` <nonebot.plugin.get_available_plugin_names>`
- `require` => {ref}``require` <nonebot.plugin.load.require>`
FrontMatter:
@ -69,7 +74,8 @@ def get_driver() -> Driver:
全局 {ref}`nonebot.drivers.Driver` 对象
异常:
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python
@ -83,23 +89,33 @@ def get_driver() -> Driver:
@overload
def get_adapter(name: str) -> Adapter:
...
"""
参数:
name: 适配器名称
返回:
指定名称的 {ref}`nonebot.adapters.Adapter` 对象
"""
@overload
def get_adapter(name: Type[A]) -> A:
...
"""
参数:
name: 适配器类型
返回:
指定类型的 {ref}`nonebot.adapters.Adapter` 对象
"""
def get_adapter(name: Union[str, Type[Adapter]]) -> Adapter:
"""获取已注册的 {ref}`nonebot.adapters.Adapter` 实例。
返回:
指定名称或类型的 {ref}`nonebot.adapters.Adapter` 对象
异常:
ValueError: 指定的 {ref}`nonebot.adapters.Adapter` 未注册
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python
@ -121,7 +137,8 @@ def get_adapters() -> Dict[str, Adapter]:
所有 {ref}`nonebot.adapters.Adapter` 实例字典
异常:
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python
@ -139,7 +156,8 @@ def get_app() -> Any:
异常:
AssertionError: 全局 Driver 对象不是 {ref}`nonebot.drivers.ReverseDriver` 类型
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python
@ -154,14 +172,16 @@ def get_app() -> Any:
def get_asgi() -> Any:
"""获取全局 {ref}`nonebot.drivers.ReverseDriver` 对应 [ASGI](https://asgi.readthedocs.io/) 对象。
"""获取全局 {ref}`nonebot.drivers.ReverseDriver` 对应
[ASGI](https://asgi.readthedocs.io/) 对象。
返回:
ASGI 对象
异常:
AssertionError: 全局 Driver 对象不是 {ref}`nonebot.drivers.ReverseDriver` 类型
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python
@ -182,7 +202,8 @@ def get_bot(self_id: Optional[str] = None) -> Bot:
当不提供时,返回一个 {ref}`nonebot.adapters.Bot`。
参数:
self_id: 用来识别 {ref}`nonebot.adapters.Bot` 的 {ref}`nonebot.adapters.Bot.self_id` 属性
self_id: 用来识别 {ref}`nonebot.adapters.Bot` 的
{ref}`nonebot.adapters.Bot.self_id` 属性
返回:
{ref}`nonebot.adapters.Bot` 对象
@ -190,7 +211,8 @@ def get_bot(self_id: Optional[str] = None) -> Bot:
异常:
KeyError: 对应 self_id 的 Bot 不存在
ValueError: 没有传入 self_id 且没有 Bot 可用
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python
@ -213,10 +235,12 @@ def get_bots() -> Dict[str, Bot]:
"""获取所有连接到 NoneBot 的 {ref}`nonebot.adapters.Bot` 对象。
返回:
一个以 {ref}`nonebot.adapters.Bot.self_id` 为键{ref}`nonebot.adapters.Bot` 对象为值的字典
一个以 {ref}`nonebot.adapters.Bot.self_id` 为键
{ref}`nonebot.adapters.Bot` 对象为值的字典
异常:
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化 ({ref}`nonebot.init <nonebot.init>` 尚未调用)
ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
({ref}`nonebot.init <nonebot.init>` 尚未调用)
用法:
```python

View File

@ -1,13 +1,16 @@
"""本模块定义了 NoneBot 本身运行所需的配置项。
NoneBot 使用 [`pydantic`](https://pydantic-docs.helpmanual.io/) 以及 [`python-dotenv`](https://saurabh-kumar.com/python-dotenv/) 来读取配置。
NoneBot 使用 [`pydantic`](https://pydantic-docs.helpmanual.io/) 以及
[`python-dotenv`](https://saurabh-kumar.com/python-dotenv/) 来读取配置。
配置项需符合特殊格式或 json 序列化格式。详情见 [`pydantic Field Type`](https://pydantic-docs.helpmanual.io/usage/types/) 文档。
配置项需符合特殊格式或 json 序列化格式
详情见 [`pydantic Field Type`](https://pydantic-docs.helpmanual.io/usage/types/) 文档。
FrontMatter:
sidebar_position: 1
description: nonebot.config 模块
"""
import os
from datetime import timedelta
from ipaddress import IPv4Address
@ -28,9 +31,8 @@ from nonebot.log import logger
class CustomEnvSettings(EnvSettingsSource):
def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
"""
Build environment variables suitable for passing to the Model.
"""
"""从环境变量和 dotenv 配置文件中读取配置项。"""
d: Dict[str, Any] = {}
if settings.__config__.case_sensitive:
@ -56,7 +58,8 @@ class CustomEnvSettings(EnvSettingsSource):
if env_val_built := self.explode_env_vars(field, env_vars):
d[field.alias] = env_val_built
else:
# field is complex and there's a value, decode that as JSON, then add explode_env_vars
# field is complex and there's a value
# decode that as JSON, then add explode_env_vars
try:
env_val = settings.__config__.parse_env_var(field.name, env_val)
except ValueError as e:
@ -72,7 +75,8 @@ class CustomEnvSettings(EnvSettingsSource):
else:
d[field.alias] = env_val
elif env_val is not None:
# simplest case, field is not complex, we only need to add the value if it was found
# simplest case, field is not complex
# we only need to add the value if it was found
d[field.alias] = env_val
# remain user custom config
@ -82,7 +86,7 @@ class CustomEnvSettings(EnvSettingsSource):
# there's a value, decode that as JSON
try:
env_val = settings.__config__.parse_env_var(env_name, val_striped)
except ValueError as e:
except ValueError:
logger.trace(
"Error while parsing JSON for "
f"{env_name!r}={val_striped!r}. "
@ -139,7 +143,7 @@ class BaseConfig(BaseSettings):
class Env(BaseConfig):
"""运行环境配置。大小写不敏感。
将会从 `环境变量` > `.env 环境配置文件` 的优先级读取环境信息。
将会从 **环境变量** > **dotenv 配置文件** 的优先级读取环境信息。
"""
environment: str = "prod"
@ -170,15 +174,17 @@ class Config(BaseConfig):
配置格式为 `<module>[:<Driver>][+<module>[:<Mixin>]]*`。
`~` 为 `nonebot.drivers.` 的缩写。
配置方法参考: [配置驱动器](https://nonebot.dev/docs/advanced/driver#%E9%85%8D%E7%BD%AE%E9%A9%B1%E5%8A%A8%E5%99%A8)
"""
host: IPvAnyAddress = IPv4Address("127.0.0.1") # type: ignore
"""NoneBot {ref}`nonebot.drivers.ReverseDriver` 服务端监听的 IP/主机名。"""
port: int = Field(default=8080, ge=1, le=65535)
"""NoneBot {ref}`nonebot.drivers.ReverseDriver` 服务端监听的端口。"""
log_level: Union[int, str] = "INFO"
"""NoneBot 日志输出等级,可以为 `int` 类型等级或等级名称
"""NoneBot 日志输出等级,可以为 `int` 类型等级或等级名称
参考 [`loguru 日志等级`](https://loguru.readthedocs.io/en/stable/api/logger.html#levels)。
参考 [记录日志](https://nonebot.dev/docs/appendices/log)[loguru 日志等级](https://loguru.readthedocs.io/en/stable/api/logger.html#levels)。
:::tip 提示
日志等级名称应为大写,如 `INFO`。
@ -209,6 +215,8 @@ class Config(BaseConfig):
command_start: Set[str] = {"/"}
"""命令的起始标记,用于判断一条消息是不是命令。
参考[命令响应规则](https://nonebot.dev/docs/advanced/matcher#command)。
用法:
```conf
COMMAND_START=["/", ""]
@ -217,6 +225,8 @@ class Config(BaseConfig):
command_sep: Set[str] = {"."}
"""命令的分隔标记,用于将文本形式的命令切分为元组(实际的命令名)。
参考[命令响应规则](https://nonebot.dev/docs/advanced/matcher#command)。
用法:
```conf
COMMAND_SEP=["."]

View File

@ -4,6 +4,7 @@ FrontMatter:
sidebar_position: 9
description: nonebot.consts 模块
"""
import os
import sys
from typing import Literal

View File

@ -82,8 +82,8 @@ class Dependent(Generic[R]):
"""
call: _DependentCallable[R]
params: Tuple[ModelField] = field(default_factory=tuple)
parameterless: Tuple[Param] = field(default_factory=tuple)
params: Tuple[ModelField, ...] = field(default_factory=tuple)
parameterless: Tuple[Param, ...] = field(default_factory=tuple)
def __repr__(self) -> str:
if inspect.isfunction(self.call) or inspect.isclass(self.call):
@ -129,7 +129,8 @@ class Dependent(Generic[R]):
break
else:
raise ValueError(
f"Unknown parameter {param.name} for function {call} with type {param.annotation}"
f"Unknown parameter {param.name} "
f"for function {call} with type {param.annotation}"
)
default_value = field_info.default
@ -182,7 +183,7 @@ class Dependent(Generic[R]):
params = cls.parse_params(call, allow_types)
parameterless_params = (
tuple()
()
if parameterless is None
else cls.parse_parameterless(tuple(parameterless), allow_types)
)

View File

@ -3,6 +3,7 @@ FrontMatter:
sidebar_position: 1
description: nonebot.dependencies.utils 模块
"""
import inspect
from typing import Any, Dict, TypeVar, Callable, ForwardRef
@ -17,6 +18,7 @@ V = TypeVar("V")
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
"""获取可调用对象签名"""
signature = inspect.signature(call)
globalns = getattr(call, "__globals__", {})
typed_params = [
@ -33,6 +35,7 @@ def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
def get_typed_annotation(param: inspect.Parameter, globalns: Dict[str, Any]) -> Any:
"""获取参数的类型注解"""
annotation = param.annotation
if isinstance(annotation, str):
annotation = ForwardRef(annotation)
@ -47,6 +50,8 @@ def get_typed_annotation(param: inspect.Parameter, globalns: Dict[str, Any]) ->
def check_field_type(field: ModelField, value: V) -> V:
"""检查字段类型是否匹配"""
_, errs_ = field.validate(value, {}, loc=())
if errs_:
raise TypeMisMatch(field, value)

View File

@ -1,10 +1,11 @@
from typing_extensions import TypeAlias
from typing import Any, List, Union, Callable, Awaitable, cast
from nonebot.utils import run_sync, is_coroutine_callable
SYNC_LIFESPAN_FUNC = Callable[[], Any]
ASYNC_LIFESPAN_FUNC = Callable[[], Awaitable[Any]]
LIFESPAN_FUNC = Union[SYNC_LIFESPAN_FUNC, ASYNC_LIFESPAN_FUNC]
SYNC_LIFESPAN_FUNC: TypeAlias = Callable[[], Any]
ASYNC_LIFESPAN_FUNC: TypeAlias = Callable[[], Awaitable[Any]]
LIFESPAN_FUNC: TypeAlias = Union[SYNC_LIFESPAN_FUNC, ASYNC_LIFESPAN_FUNC]
class Lifespan:

View File

@ -29,7 +29,8 @@ try:
import aiohttp
except ModuleNotFoundError as e: # pragma: no cover
raise ImportError(
"Please install aiohttp first to use this driver. `pip install nonebot2[aiohttp]`"
"Please install aiohttp first to use this driver. "
"Install with pip: `pip install nonebot2[aiohttp]`"
) from e

View File

@ -41,7 +41,8 @@ try:
from starlette.websockets import WebSocket, WebSocketState, WebSocketDisconnect
except ModuleNotFoundError as e: # pragma: no cover
raise ImportError(
"Please install FastAPI by using `pip install nonebot2[fastapi]`"
"Please install FastAPI first to use this driver. "
"Install with pip: `pip install nonebot2[fastapi]`"
) from e
@ -90,7 +91,7 @@ class Driver(ReverseDriver):
"""FastAPI 驱动框架。"""
def __init__(self, env: Env, config: NoneBotConfig):
super(Driver, self).__init__(env, config)
super().__init__(env, config)
self.fastapi_config: Config = Config(**config.dict())

View File

@ -14,6 +14,7 @@ FrontMatter:
sidebar_position: 3
description: nonebot.drivers.httpx 模块
"""
from typing import Type, AsyncGenerator
from contextlib import asynccontextmanager
@ -33,7 +34,8 @@ try:
import httpx
except ModuleNotFoundError as e: # pragma: no cover
raise ImportError(
"Please install httpx by using `pip install nonebot2[httpx]`"
"Please install httpx first to use this driver. "
"Install with pip: `pip install nonebot2[httpx]`"
) from e

View File

@ -9,7 +9,6 @@ FrontMatter:
description: nonebot.drivers.none 模块
"""
import signal
import asyncio
import threading

View File

@ -39,7 +39,8 @@ try:
from quart import Websocket as QuartWebSocket
except ModuleNotFoundError as e: # pragma: no cover
raise ImportError(
"Please install Quart by using `pip install nonebot2[quart]`"
"Please install Quart first to use this driver. "
"Install with pip: `pip install nonebot2[quart]`"
) from e
_AsyncCallable = TypeVar("_AsyncCallable", bound=Callable[..., Coroutine])
@ -188,9 +189,7 @@ class Driver(ReverseDriver):
async def _handle_http(self, setup: HTTPServerSetup) -> Response:
request: Request = _request
json = None
if request.is_json:
json = await request.get_json()
json = await request.get_json() if request.is_json else None
data = await request.form
files_dict = await request.files

View File

@ -14,10 +14,12 @@ FrontMatter:
sidebar_position: 4
description: nonebot.drivers.websockets 模块
"""
import logging
from functools import wraps
from typing_extensions import ParamSpec
from contextlib import asynccontextmanager
from typing import Type, Union, AsyncGenerator
from typing import Type, Union, TypeVar, Callable, Awaitable, AsyncGenerator
from nonebot.typing import overrides
from nonebot.log import LoguruHandler
@ -32,16 +34,20 @@ try:
from websockets.legacy.client import Connect, WebSocketClientProtocol
except ModuleNotFoundError as e: # pragma: no cover
raise ImportError(
"Please install websockets by using `pip install nonebot2[websockets]`"
"Please install websockets first to use this driver. "
"Install with pip: `pip install nonebot2[websockets]`"
) from e
T = TypeVar("T")
P = ParamSpec("P")
logger = logging.Logger("websockets.client", "INFO")
logger.addHandler(LoguruHandler())
def catch_closed(func):
def catch_closed(func: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
@wraps(func)
async def decorator(*args, **kwargs):
async def decorator(*args: P.args, **kwargs: P.kwargs) -> T:
try:
return await func(*args, **kwargs)
except ConnectionClosed as e:

View File

@ -43,9 +43,9 @@ class NoneBotException(Exception):
# Rule Exception
class ParserExit(NoneBotException):
"""{ref}`nonebot.rule.shell_command` 处理消息失败时返回的异常"""
"""{ref}`nonebot.rule.shell_command` 处理消息失败时返回的异常"""
def __init__(self, status: int = 0, message: Optional[str] = None):
def __init__(self, status: int = 0, message: Optional[str] = None) -> None:
self.status = status
self.message = message
@ -69,7 +69,7 @@ class IgnoredException(ProcessException):
reason: 忽略事件的原因
"""
def __init__(self, reason: Any):
def __init__(self, reason: Any) -> None:
self.reason: Any = reason
def __repr__(self) -> str:
@ -96,7 +96,7 @@ class SkippedException(ProcessException):
class TypeMisMatch(SkippedException):
"""当前 `Handler` 的参数类型不匹配。"""
def __init__(self, param: ModelField, value: Any):
def __init__(self, param: ModelField, value: Any) -> None:
self.param: ModelField = param
self.value: Any = value
@ -108,7 +108,8 @@ class TypeMisMatch(SkippedException):
class MockApiException(ProcessException):
"""指示 NoneBot 阻止本次 API 调用或修改本次调用返回值,并返回自定义内容。可由 api hook 抛出。
"""指示 NoneBot 阻止本次 API 调用或修改本次调用返回值,并返回自定义内容。
可由 api hook 抛出。
参数:
result: 返回的内容
@ -144,7 +145,8 @@ class MatcherException(NoneBotException):
class PausedException(MatcherException):
"""指示 NoneBot 结束当前 `Handler` 并等待下一条消息后继续下一个 `Handler`。可用于用户输入新信息。
"""指示 NoneBot 结束当前 `Handler` 并等待下一条消息后继续下一个 `Handler`。
可用于用户输入新信息。
可以在 `Handler` 中通过 {ref}`nonebot.matcher.Matcher.pause` 抛出。
@ -158,7 +160,8 @@ class PausedException(MatcherException):
class RejectedException(MatcherException):
"""指示 NoneBot 结束当前 `Handler` 并等待下一条消息后重新运行当前 `Handler`。可用于用户重新输入。
"""指示 NoneBot 结束当前 `Handler` 并等待下一条消息后重新运行当前 `Handler`。
可用于用户重新输入。
可以在 `Handler` 中通过 {ref}`nonebot.matcher.Matcher.reject` 抛出。
@ -187,7 +190,7 @@ class FinishedException(MatcherException):
# Adapter Exceptions
class AdapterException(NoneBotException):
"""代表 `Adapter` 抛出的异常,所有的 `Adapter` 都要在内部继承自这个 `Exception`
"""代表 `Adapter` 抛出的异常,所有的 `Adapter` 都要在内部继承自这个 `Exception`
参数:
adapter_name: 标识 adapter
@ -210,7 +213,9 @@ class ApiNotAvailable(AdapterException):
class NetworkError(AdapterException):
"""在网络出现问题时抛出,如: API 请求地址不正确, API 请求无返回或返回状态非正常等。"""
"""在网络出现问题时抛出,
如: API 请求地址不正确, API 请求无返回或返回状态非正常等。
"""
class ActionFailed(AdapterException):
@ -219,13 +224,13 @@ class ActionFailed(AdapterException):
# Driver Exceptions
class DriverException(NoneBotException):
"""`Driver` 抛出的异常基类"""
"""`Driver` 抛出的异常基类"""
class WebSocketClosed(DriverException):
"""WebSocket 连接已关闭"""
"""WebSocket 连接已关闭"""
def __init__(self, code: int, reason: Optional[str] = None):
def __init__(self, code: int, reason: Optional[str] = None) -> None:
self.code = code
self.reason = reason

View File

@ -44,10 +44,11 @@ class Event(abc.ABC, BaseModel):
def get_log_string(self) -> str:
"""获取事件日志信息的方法。
通常你不需要修改这个方法,只有当希望 NoneBot 隐藏该事件日志时,可以抛出 `NoLogException` 异常。
通常你不需要修改这个方法,只有当希望 NoneBot 隐藏该事件日志时,
可以抛出 `NoLogException` 异常。
异常:
NoLogException:
NoLogException: 希望 NoneBot 隐藏该事件日志
"""
return f"[{self.get_event_name()}]: {self.get_event_description()}"
@ -58,7 +59,9 @@ class Event(abc.ABC, BaseModel):
@abc.abstractmethod
def get_session_id(self) -> str:
"""获取会话 id 的方法,用于判断当前事件属于哪一个会话,通常是用户 id、群组 id 组合。"""
"""获取会话 id 的方法,用于判断当前事件属于哪一个会话,
通常是用户 id、群组 id 组合。
"""
raise NotImplementedError
@abc.abstractmethod

View File

@ -98,7 +98,7 @@ class MessageSegment(abc.ABC, Generic[TM]):
class Message(List[TMS], abc.ABC):
"""消息数组
"""消息序列
参数:
message: 消息内容
@ -124,9 +124,9 @@ class Message(List[TMS], abc.ABC):
def template(cls, format_string: Union[str, TM]) -> MessageTemplate[Self]:
"""创建消息模板。
用法和 `str.format` 大致相同, 但是可以输出消息对象, 并且支持以 `Message` 对象作为消息模板
并且提供了拓展的格式化控制符, 可以用适用于该消息类型的 `MessageSegment` 工厂方法创建消息
用法和 `str.format` 大致相同支持以 `Message` 对象作为消息模板并输出消息对象。
并且提供了拓展的格式化控制符,
可以通过该消息类型的 `MessageSegment` 工厂方法创建消息
参数:
format_string: 格式化模板

View File

@ -1,5 +1,6 @@
import functools
from string import Formatter
from typing_extensions import TypeAlias
from typing import (
TYPE_CHECKING,
Any,
@ -25,7 +26,7 @@ if TYPE_CHECKING:
TM = TypeVar("TM", bound="Message")
TF = TypeVar("TF", str, "Message")
FormatSpecFunc = Callable[[Any], str]
FormatSpecFunc: TypeAlias = Callable[[Any], str]
FormatSpecFunc_T = TypeVar("FormatSpecFunc_T", bound=FormatSpecFunc)

View File

@ -89,9 +89,7 @@ class Driver(abc.ABC):
@abc.abstractmethod
def run(self, *args, **kwargs):
"""
启动驱动框架
"""
"""启动驱动框架"""
logger.opt(colors=True).debug(
f"<g>Loaded adapters: {escape_tag(', '.join(self._adapters))}</g>"
)
@ -152,8 +150,10 @@ class Driver(abc.ABC):
await asyncio.gather(*coros)
except Exception as e:
logger.opt(colors=True, exception=e).error(
"<r><bg #f8bbd0>Error when running WebSocketConnection hook. "
"Running cancelled!</bg #f8bbd0></r>"
"<r><bg #f8bbd0>"
"Error when running WebSocketConnection hook. "
"Running cancelled!"
"</bg #f8bbd0></r>"
)
asyncio.create_task(_run_hook(bot))
@ -177,8 +177,10 @@ class Driver(abc.ABC):
await asyncio.gather(*coros)
except Exception as e:
logger.opt(colors=True, exception=e).error(
"<r><bg #f8bbd0>Error when running WebSocketDisConnection hook. "
"Running cancelled!</bg #f8bbd0></r>"
"<r><bg #f8bbd0>"
"Error when running WebSocketDisConnection hook. "
"Running cancelled!"
"</bg #f8bbd0></r>"
)
asyncio.create_task(_run_hook(bot))
@ -241,7 +243,7 @@ def combine_driver(driver: Type[Driver], *mixins: Type[ForwardMixin]) -> Type[Dr
# check first
assert issubclass(driver, Driver), "`driver` must be subclass of Driver"
assert all(
map(lambda m: issubclass(m, ForwardMixin), mixins)
issubclass(m, ForwardMixin) for m in mixins
), "`mixins` must be subclass of ForwardMixin"
if not mixins:
@ -251,7 +253,9 @@ def combine_driver(driver: Type[Driver], *mixins: Type[ForwardMixin]) -> Type[Dr
return (
driver.type.__get__(self)
+ "+"
+ "+".join(map(lambda x: x.type.__get__(self), mixins))
+ "+".join(x.type.__get__(self) for x in mixins)
)
return type("CombinedDriver", (*mixins, driver, ForwardDriver), {"type": property(type_)}) # type: ignore
return type(
"CombinedDriver", (*mixins, driver, ForwardDriver), {"type": property(type_)}
) # type: ignore

View File

@ -2,6 +2,7 @@ import abc
import urllib.request
from enum import Enum
from dataclasses import dataclass
from typing_extensions import TypeAlias
from http.cookiejar import Cookie, CookieJar
from typing import (
IO,
@ -21,28 +22,30 @@ from typing import (
from yarl import URL as URL
from multidict import CIMultiDict
RawURL = Tuple[bytes, bytes, Optional[int], bytes]
RawURL: TypeAlias = Tuple[bytes, bytes, Optional[int], bytes]
SimpleQuery = Union[str, int, float]
QueryVariable = Union[SimpleQuery, List[SimpleQuery]]
QueryTypes = Union[
SimpleQuery: TypeAlias = Union[str, int, float]
QueryVariable: TypeAlias = Union[SimpleQuery, List[SimpleQuery]]
QueryTypes: TypeAlias = Union[
None, str, Mapping[str, QueryVariable], List[Tuple[str, QueryVariable]]
]
HeaderTypes = Union[
HeaderTypes: TypeAlias = Union[
None,
CIMultiDict[str],
Dict[str, str],
List[Tuple[str, str]],
]
CookieTypes = Union[None, "Cookies", CookieJar, Dict[str, str], List[Tuple[str, str]]]
CookieTypes: TypeAlias = Union[
None, "Cookies", CookieJar, Dict[str, str], List[Tuple[str, str]]
]
ContentTypes = Union[str, bytes, None]
DataTypes = Union[dict, None]
FileContent = Union[IO[bytes], bytes]
FileType = Tuple[Optional[str], FileContent, Optional[str]]
FileTypes = Union[
ContentTypes: TypeAlias = Union[str, bytes, None]
DataTypes: TypeAlias = Union[dict, None]
FileContent: TypeAlias = Union[IO[bytes], bytes]
FileType: TypeAlias = Tuple[Optional[str], FileContent, Optional[str]]
FileTypes: TypeAlias = Union[
# file (or bytes)
FileContent,
# (filename, file (or bytes))
@ -50,7 +53,7 @@ FileTypes = Union[
# (filename, file (or bytes), content_type)
FileType,
]
FilesTypes = Union[Dict[str, FileTypes], List[Tuple[str, FileTypes]], None]
FilesTypes: TypeAlias = Union[Dict[str, FileTypes], List[Tuple[str, FileTypes]], None]
class HTTPVersion(Enum):
@ -160,7 +163,6 @@ class Response:
class WebSocket(abc.ABC):
def __init__(self, *, request: Request):
# request
self.request: Request = request
def __repr__(self) -> str:
@ -169,9 +171,7 @@ class WebSocket(abc.ABC):
@property
@abc.abstractmethod
def closed(self) -> bool:
"""
连接是否已经关闭
"""
"""连接是否已经关闭"""
raise NotImplementedError
@abc.abstractmethod

View File

@ -29,6 +29,13 @@ from nonebot.internal.adapter import (
MessageSegment,
MessageTemplate,
)
from nonebot.typing import (
T_State,
T_Handler,
T_TypeUpdater,
T_DependencyCache,
T_PermissionUpdater,
)
from nonebot.consts import (
ARG_KEY,
RECEIVE_KEY,
@ -36,14 +43,6 @@ from nonebot.consts import (
LAST_RECEIVE_KEY,
REJECT_CACHE_TARGET,
)
from nonebot.typing import (
Any,
T_State,
T_Handler,
T_TypeUpdater,
T_DependencyCache,
T_PermissionUpdater,
)
from nonebot.exception import (
PausedException,
StopPropagation,
@ -376,7 +375,7 @@ class Matcher(metaclass=MatcherMeta):
return
await matcher.reject()
_parameterless = (Depends(_receive), *(parameterless or tuple()))
_parameterless = (Depends(_receive), *(parameterless or ()))
def _decorator(func: T_Handler) -> T_Handler:
if cls.handlers and cls.handlers[-1].call is func:
@ -406,7 +405,8 @@ class Matcher(metaclass=MatcherMeta):
) -> Callable[[T_Handler], T_Handler]:
"""装饰一个函数来指示 NoneBot 获取一个参数 `key`
当要获取的 `key` 不存在时接收用户新的一条消息再运行该函数,如果 `key` 已存在则直接继续运行
当要获取的 `key` 不存在时接收用户新的一条消息再运行该函数,
如果 `key` 已存在则直接继续运行
参数:
key: 参数名
@ -423,7 +423,7 @@ class Matcher(metaclass=MatcherMeta):
return
await matcher.reject(prompt)
_parameterless = (Depends(_key_getter), *(parameterless or tuple()))
_parameterless = (Depends(_key_getter), *(parameterless or ()))
def _decorator(func: T_Handler) -> T_Handler:
if cls.handlers and cls.handlers[-1].call is func:
@ -454,7 +454,8 @@ class Matcher(metaclass=MatcherMeta):
参数:
message: 消息内容
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,请参考对应 adapter 的 bot 对象 api
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,
请参考对应 adapter 的 bot 对象 api
"""
bot = current_bot.get()
event = current_event.get()
@ -475,7 +476,8 @@ class Matcher(metaclass=MatcherMeta):
参数:
message: 消息内容
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,请参考对应 adapter 的 bot 对象 api
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,
请参考对应 adapter 的 bot 对象 api
"""
if message is not None:
await cls.send(message, **kwargs)
@ -491,7 +493,8 @@ class Matcher(metaclass=MatcherMeta):
参数:
prompt: 消息内容
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,请参考对应 adapter 的 bot 对象 api
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,
请参考对应 adapter 的 bot 对象 api
"""
if prompt is not None:
await cls.send(prompt, **kwargs)
@ -508,7 +511,8 @@ class Matcher(metaclass=MatcherMeta):
参数:
prompt: 消息内容
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,请参考对应 adapter 的 bot 对象 api
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,
请参考对应 adapter 的 bot 对象 api
"""
if prompt is not None:
await cls.send(prompt, **kwargs)
@ -527,7 +531,8 @@ class Matcher(metaclass=MatcherMeta):
参数:
key: 参数名
prompt: 消息内容
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,请参考对应 adapter 的 bot 对象 api
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,
请参考对应 adapter 的 bot 对象 api
"""
matcher = current_matcher.get()
matcher.set_target(ARG_KEY.format(key=key))
@ -548,7 +553,8 @@ class Matcher(metaclass=MatcherMeta):
参数:
id: 消息 id
prompt: 消息内容
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,请参考对应 adapter 的 bot 对象 api
kwargs: {ref}`nonebot.adapters.Bot.send` 的参数,
请参考对应 adapter 的 bot 对象 api
"""
matcher = current_matcher.get()
matcher.set_target(RECEIVE_KEY.format(id=id))

View File

@ -63,7 +63,10 @@ def Depends(
finally:
...
async def handler(param_name: Any = Depends(depend_func), gen: Any = Depends(depend_gen_func)):
async def handler(
param_name: Any = Depends(depend_func),
gen: Any = Depends(depend_gen_func),
):
...
```
"""

View File

@ -88,5 +88,6 @@ logger_id = logger.add(
filter=default_filter,
format=default_format,
)
"""默认日志处理器 id"""
__autodoc__ = {"logger_id": False}

View File

@ -166,7 +166,7 @@ async def _apply_event_preprocessors(
for proc in _event_preprocessors
)
)
except IgnoredException as e:
except IgnoredException:
logger.opt(colors=True).info(
f"Event {escape_tag(event.get_event_name())} is <b>ignored</b>"
)
@ -293,7 +293,7 @@ async def _apply_run_postprocessors(
) -> None:
"""运行事件响应器运行后处理。
Args:
参数:
bot: Bot 对象
event: Event 对象
matcher: 事件响应器

View File

@ -1,7 +1,8 @@
"""本模块是 {ref}`nonebot.matcher.Matcher.permission` 的类型定义。
每个 {ref}`nonebot.matcher.Matcher` 拥有一个 {ref}`nonebot.permission.Permission`
其中是 `PermissionChecker` 的集合,只要有一个 `PermissionChecker` 检查结果为 `True` 时就会继续运行
每个{ref}`事件响应器 <nonebot.matcher.Matcher>`
拥有一个 {ref}`nonebot.permission.Permission`,其中是 `PermissionChecker` 的集合
只要有一个 `PermissionChecker` 检查结果为 `True` 时就会继续运行。
FrontMatter:
sidebar_position: 6

View File

@ -24,8 +24,10 @@
- `load_all_plugins` => {ref}``load_all_plugins` <nonebot.plugin.load.load_all_plugins>`
- `load_from_json` => {ref}``load_from_json` <nonebot.plugin.load.load_from_json>`
- `load_from_toml` => {ref}``load_from_toml` <nonebot.plugin.load.load_from_toml>`
- `load_builtin_plugin` => {ref}``load_builtin_plugin` <nonebot.plugin.load.load_builtin_plugin>`
- `load_builtin_plugins` => {ref}``load_builtin_plugins` <nonebot.plugin.load.load_builtin_plugins>`
- `load_builtin_plugin` =>
{ref}``load_builtin_plugin` <nonebot.plugin.load.load_builtin_plugin>`
- `load_builtin_plugins` =>
{ref}``load_builtin_plugins` <nonebot.plugin.load.load_builtin_plugins>`
- `require` => {ref}``require` <nonebot.plugin.load.require>`
- `PluginMetadata` => {ref}``PluginMetadata` <nonebot.plugin.plugin.PluginMetadata>`
@ -42,7 +44,7 @@ from typing import Set, Dict, List, Tuple, Optional
_plugins: Dict[str, "Plugin"] = {}
_managers: List["PluginManager"] = []
_current_plugin_chain: ContextVar[Tuple["Plugin", ...]] = ContextVar(
"_current_plugin_chain", default=tuple()
"_current_plugin_chain", default=()
)

View File

@ -4,6 +4,7 @@ FrontMatter:
sidebar_position: 1
description: nonebot.plugin.load 模块
"""
import json
from pathlib import Path
from types import ModuleType
@ -25,7 +26,8 @@ def load_plugin(module_path: Union[str, Path]) -> Optional[Plugin]:
"""加载单个插件,可以是本地插件或是通过 `pip` 安装的插件。
参数:
module_path: 插件名称 `path.to.your.plugin` 或插件路径 `pathlib.Path(path/to/your/plugin)`
module_path: 插件名称 `path.to.your.plugin`
或插件路径 `pathlib.Path(path/to/your/plugin)`
"""
module_path = (
path_to_module_name(module_path)
@ -63,7 +65,8 @@ def load_all_plugins(
def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
"""导入指定 json 文件中的 `plugins` 以及 `plugin_dirs` 下多个插件,以 `_` 开头的插件不会被导入!
"""导入指定 json 文件中的 `plugins` 以及 `plugin_dirs` 下多个插件
以 `_` 开头的插件不会被导入!
参数:
file_path: 指定 json 文件路径
@ -81,7 +84,7 @@ def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
nonebot.load_from_json("plugins.json")
```
"""
with open(file_path, "r", encoding=encoding) as f:
with open(file_path, encoding=encoding) as f:
data = json.load(f)
if not isinstance(data, dict):
raise TypeError("json file must contains a dict!")
@ -93,7 +96,9 @@ def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
def load_from_toml(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
"""导入指定 toml 文件 `[tool.nonebot]` 中的 `plugins` 以及 `plugin_dirs` 下多个插件,以 `_` 开头的插件不会被导入!
"""导入指定 toml 文件 `[tool.nonebot]` 中的
`plugins` 以及 `plugin_dirs` 下多个插件。
以 `_` 开头的插件不会被导入!
参数:
file_path: 指定 toml 文件路径
@ -110,7 +115,7 @@ def load_from_toml(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
nonebot.load_from_toml("pyproject.toml")
```
"""
with open(file_path, "r", encoding=encoding) as f:
with open(file_path, encoding=encoding) as f:
data = tomllib.loads(f.read())
nonebot_data = data.get("tool", {}).get("nonebot")

View File

@ -6,6 +6,7 @@ FrontMatter:
sidebar_position: 5
description: nonebot.plugin.manager 模块
"""
import sys
import pkgutil
import importlib

View File

@ -4,6 +4,7 @@ FrontMatter:
sidebar_position: 2
description: nonebot.plugin.on 模块
"""
import re
import inspect
from types import ModuleType
@ -322,7 +323,8 @@ def on_shell_command(
与普通的 `on_command` 不同的是,在添加 `parser` 参数时, 响应器会自动处理消息。
并将用户输入的原始参数列表保存在 `state["argv"]`, `parser` 处理的参数保存在 `state["args"]` 中
可以通过 {ref}`nonebot.params.ShellCommandArgv` 获取原始参数列表,
通过 {ref}`nonebot.params.ShellCommandArgs` 获取解析后的参数字典。
参数:
cmd: 指定命令内容
@ -712,7 +714,8 @@ class MatcherGroup(_Group):
与普通的 `on_command` 不同的是,在添加 `parser` 参数时, 响应器会自动处理消息。
并将用户输入的原始参数列表保存在 `state["argv"]`, `parser` 处理的参数保存在 `state["args"]` 中
可以通过 {ref}`nonebot.params.ShellCommandArgv` 获取原始参数列表,
通过 {ref}`nonebot.params.ShellCommandArgs` 获取解析后的参数字典。
参数:
cmd: 指定命令内容

View File

@ -1,7 +1,6 @@
import re
from types import ModuleType
from datetime import datetime, timedelta
from typing import Set, List, Type, Tuple, Union, Optional
from nonebot.adapters import Event
from nonebot.matcher import Matcher
@ -12,399 +11,399 @@ from nonebot.typing import T_State, T_Handler, T_RuleChecker, T_PermissionChecke
from .plugin import Plugin
def store_matcher(matcher: Type[Matcher]) -> None: ...
def get_matcher_plugin(depth: int = ...) -> Optional[Plugin]: ...
def get_matcher_module(depth: int = ...) -> Optional[ModuleType]: ...
def store_matcher(matcher: type[Matcher]) -> None: ...
def get_matcher_plugin(depth: int = ...) -> Plugin | None: ...
def get_matcher_module(depth: int = ...) -> ModuleType | None: ...
def on(
type: str = "",
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
*,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_metaevent(
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
*,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_message(
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
*,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_notice(
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
*,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_request(
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
*,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_startswith(
msg: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
msg: str | tuple[str, ...],
rule: Rule | T_RuleChecker | None = ...,
ignorecase: bool = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_endswith(
msg: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
msg: str | tuple[str, ...],
rule: Rule | T_RuleChecker | None = ...,
ignorecase: bool = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_fullmatch(
msg: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
msg: str | tuple[str, ...],
rule: Rule | T_RuleChecker | None = ...,
ignorecase: bool = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_keyword(
keywords: Set[str],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
keywords: set[str],
rule: Rule | T_RuleChecker | None = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_command(
cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
force_whitespace: Optional[Union[str, bool]] = ...,
cmd: str | tuple[str, ...],
rule: Rule | T_RuleChecker | None = ...,
aliases: set[str | tuple[str, ...]] | None = ...,
force_whitespace: str | bool | None = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_shell_command(
cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
parser: Optional[ArgumentParser] = ...,
cmd: str | tuple[str, ...],
rule: Rule | T_RuleChecker | None = ...,
aliases: set[str | tuple[str, ...]] | None = ...,
parser: ArgumentParser | None = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_regex(
pattern: str,
flags: Union[int, re.RegexFlag] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
flags: int | re.RegexFlag = ...,
rule: Rule | T_RuleChecker | None = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_type(
types: Union[Type[Event], Tuple[Type[Event], ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
types: type[Event] | tuple[type[Event], ...],
rule: Rule | T_RuleChecker | None = ...,
*,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
class CommandGroup:
def __init__(
self,
cmd: Union[str, Tuple[str, ...]],
cmd: str | tuple[str, ...],
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
state: T_State | None = ...,
): ...
def command(
self,
cmd: Union[str, Tuple[str, ...]],
cmd: str | tuple[str, ...],
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
force_whitespace: Optional[Union[str, bool]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
aliases: set[str | tuple[str, ...]] | None = ...,
force_whitespace: str | bool | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def shell_command(
self,
cmd: Union[str, Tuple[str, ...]],
cmd: str | tuple[str, ...],
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
parser: Optional[ArgumentParser] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
aliases: set[str | tuple[str, ...]] | None = ...,
parser: ArgumentParser | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
class MatcherGroup:
def __init__(
self,
*,
type: str = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
state: T_State | None = ...,
): ...
def on(
self,
*,
type: str = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_metaevent(
self,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_message(
self,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_notice(
self,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_request(
self,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_startswith(
self,
msg: Union[str, Tuple[str, ...]],
msg: str | tuple[str, ...],
*,
ignorecase: bool = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_endswith(
self,
msg: Union[str, Tuple[str, ...]],
msg: str | tuple[str, ...],
*,
ignorecase: bool = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_fullmatch(
self,
msg: Union[str, Tuple[str, ...]],
msg: str | tuple[str, ...],
*,
ignorecase: bool = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_keyword(
self,
keywords: Set[str],
keywords: set[str],
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_command(
self,
cmd: Union[str, Tuple[str, ...]],
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
force_whitespace: Optional[Union[str, bool]] = ...,
cmd: str | tuple[str, ...],
aliases: set[str | tuple[str, ...]] | None = ...,
force_whitespace: str | bool | None = ...,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_shell_command(
self,
cmd: Union[str, Tuple[str, ...]],
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
parser: Optional[ArgumentParser] = ...,
cmd: str | tuple[str, ...],
aliases: set[str | tuple[str, ...]] | None = ...,
parser: ArgumentParser | None = ...,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_regex(
self,
pattern: str,
flags: Union[int, re.RegexFlag] = ...,
flags: int | re.RegexFlag = ...,
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...
def on_type(
self,
types: Union[Type[Event], Tuple[Type[Event]]],
types: type[Event] | tuple[type[Event]],
*,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Union[Permission, T_PermissionChecker]] = ...,
handlers: Optional[List[Union[T_Handler, Dependent]]] = ...,
rule: Rule | T_RuleChecker | None = ...,
permission: Permission | T_PermissionChecker | None = ...,
handlers: list[T_Handler | Dependent] | None = ...,
temp: bool = ...,
expire_time: Optional[Union[datetime, timedelta]] = ...,
expire_time: datetime | timedelta | None = ...,
priority: int = ...,
block: bool = ...,
state: Optional[T_State] = ...,
) -> Type[Matcher]: ...
state: T_State | None = ...,
) -> type[Matcher]: ...

View File

@ -26,7 +26,7 @@ async def matcher_mutex(event: Event) -> AsyncGenerator[bool, None]:
yield result
else:
current_event_id = id(event)
if event_id := _running_matcher.get(session_id, None):
if event_id := _running_matcher.get(session_id):
result = event_id != current_event_id
else:
_running_matcher[session_id] = current_event_id

View File

@ -1,7 +1,8 @@
"""本模块是 {ref}`nonebot.matcher.Matcher.rule` 的类型定义。
每个事件响应器 {ref}`nonebot.matcher.Matcher` 拥有一个匹配规则 {ref}`nonebot.rule.Rule`
其中是 `RuleChecker` 的集合,只有当所有 `RuleChecker` 检查结果为 `True` 时继续运行
每个{ref}`事件响应器 <nonebot.matcher.Matcher>`拥有一个
{ref}`nonebot.rule.Rule`,其中是 `RuleChecker` 的集合
只有当所有 `RuleChecker` 检查结果为 `True` 时继续运行。
FrontMatter:
sidebar_position: 5
@ -60,20 +61,19 @@ from nonebot.consts import (
T = TypeVar("T")
CMD_RESULT = TypedDict(
"CMD_RESULT",
{
"command": Optional[Tuple[str, ...]],
"raw_command": Optional[str],
"command_arg": Optional[Message],
"command_start": Optional[str],
"command_whitespace": Optional[str],
},
)
TRIE_VALUE = NamedTuple(
"TRIE_VALUE", [("command_start", str), ("command", Tuple[str, ...])]
)
class CMD_RESULT(TypedDict):
command: Optional[Tuple[str, ...]]
raw_command: Optional[str]
command_arg: Optional[Message]
command_start: Optional[str]
command_whitespace: Optional[str]
class TRIE_VALUE(NamedTuple):
command_start: str
command: Tuple[str, ...]
parser_message: ContextVar[str] = ContextVar("parser_message")
@ -406,7 +406,7 @@ def command(
force_whitespace: 是否强制命令后必须有指定空白符
用法:
使用默认 `command_start`, `command_sep` 配置
使用默认 `command_start`, `command_sep` 配置情况下:
命令 `("test",)` 可以匹配: `/test` 开头的消息
命令 `("test", "sub")` 可以匹配: `/test.sub` 开头的消息
@ -441,6 +441,8 @@ def command(
class ArgumentParser(ArgParser):
"""`shell_like` 命令参数解析器,解析出错时不会退出程序。
支持 {ref}`nonebot.adapters.Message` 富文本解析。
用法:
用法与 `argparse.ArgumentParser` 相同,
参考文档: [argparse](https://docs.python.org/3/library/argparse.html)
@ -588,10 +590,14 @@ def shell_command(
根据配置里提供的 {ref}``command_start` <nonebot.config.Config.command_start>`,
{ref}``command_sep` <nonebot.config.Config.command_sep>` 判断消息是否为命令。
可以通过 {ref}`nonebot.params.Command` 获取匹配成功的命令(例: `("test",)`
通过 {ref}`nonebot.params.RawCommand` 获取匹配成功的原始命令文本(例: `"/test"`
通过 {ref}`nonebot.params.ShellCommandArgv` 获取解析前的参数列表(例: `["arg", "-h"]`
通过 {ref}`nonebot.params.ShellCommandArgs` 获取解析后的参数字典(例: `{"arg": "arg", "h": True}`
可以通过 {ref}`nonebot.params.Command` 获取匹配成功的命令
(例: `("test",)`
通过 {ref}`nonebot.params.RawCommand` 获取匹配成功的原始命令文本
(例: `"/test"`
通过 {ref}`nonebot.params.ShellCommandArgv` 获取解析前的参数列表
(例: `["arg", "-h"]`
通过 {ref}`nonebot.params.ShellCommandArgs` 获取解析后的参数字典
(例: `{"arg": "arg", "h": True}`)。
:::warning 警告
如果参数解析失败,则通过 {ref}`nonebot.params.ShellCommandArgs`
@ -603,7 +609,8 @@ def shell_command(
parser: {ref}`nonebot.rule.ArgumentParser` 对象
用法:
使用默认 `command_start`, `command_sep` 配置,更多示例参考 `argparse` 标准库文档。
使用默认 `command_start`, `command_sep` 配置,更多示例参考
[argparse](https://docs.python.org/3/library/argparse.html) 标准库文档。
```python
from nonebot.rule import ArgumentParser
@ -698,7 +705,8 @@ def regex(regex: str, flags: Union[int, re.RegexFlag] = 0) -> Rule:
:::
:::tip 提示
正则表达式匹配使用 `EventMessage` 的 `str` 字符串,而非 `EventMessage` 的 `PlainText` 纯文本字符串
正则表达式匹配使用 `EventMessage` 的 `str` 字符串,
而非 `EventMessage` 的 `PlainText` 纯文本字符串
:::
"""

View File

@ -1,17 +1,16 @@
"""本模块定义了 NoneBot 模块中共享的一些类型。
下面的文档中,「类型」部分使用 Python 的 Type Hint 语法,
使用 Python 的 Type Hint 语法,
参考 [`PEP 484`](https://www.python.org/dev/peps/pep-0484/),
[`PEP 526`](https://www.python.org/dev/peps/pep-0526/) 和
[`typing`](https://docs.python.org/3/library/typing.html)。
除了 Python 内置的类型,下面还出现了如下 NoneBot 自定类型,实际上它们是 Python 内置类型的别名。
FrontMatter:
sidebar_position: 11
description: nonebot.typing 模块
"""
from typing_extensions import ParamSpec, TypeAlias
from typing import (
TYPE_CHECKING,
Any,
@ -30,8 +29,9 @@ if TYPE_CHECKING:
from nonebot.permission import Permission
T = TypeVar("T")
P = ParamSpec("P")
T_Wrapped = TypeVar("T_Wrapped", bound=Callable)
T_Wrapped: TypeAlias = Callable[P, T]
def overrides(InterfaceClass: object) -> Callable[[T_Wrapped], T_Wrapped]:
@ -45,13 +45,13 @@ def overrides(InterfaceClass: object) -> Callable[[T_Wrapped], T_Wrapped]:
# state
T_State = Dict[Any, Any]
T_State: TypeAlias = Dict[Any, Any]
"""事件处理状态 State 类型"""
_DependentCallable = Union[Callable[..., T], Callable[..., Awaitable[T]]]
_DependentCallable: TypeAlias = Union[Callable[..., T], Callable[..., Awaitable[T]]]
# driver hooks
T_BotConnectionHook = _DependentCallable[Any]
T_BotConnectionHook: TypeAlias = _DependentCallable[Any]
"""Bot 连接建立时钩子函数
依赖参数:
@ -60,7 +60,7 @@ T_BotConnectionHook = _DependentCallable[Any]
- BotParam: Bot 对象
- DefaultParam: 带有默认值的参数
"""
T_BotDisconnectionHook = _DependentCallable[Any]
T_BotDisconnectionHook: TypeAlias = _DependentCallable[Any]
"""Bot 连接断开时钩子函数
依赖参数:
@ -71,15 +71,15 @@ T_BotDisconnectionHook = _DependentCallable[Any]
"""
# api hooks
T_CallingAPIHook = Callable[["Bot", str, Dict[str, Any]], Awaitable[Any]]
T_CallingAPIHook: TypeAlias = Callable[["Bot", str, Dict[str, Any]], Awaitable[Any]]
"""`bot.call_api` 钩子函数"""
T_CalledAPIHook = Callable[
T_CalledAPIHook: TypeAlias = Callable[
["Bot", Optional[Exception], str, Dict[str, Any], Any], Awaitable[Any]
]
"""`bot.call_api` 后执行的函数,参数分别为 bot, exception, api, data, result"""
# event hooks
T_EventPreProcessor = _DependentCallable[Any]
T_EventPreProcessor: TypeAlias = _DependentCallable[Any]
"""事件预处理函数 EventPreProcessor 类型
依赖参数:
@ -90,7 +90,7 @@ T_EventPreProcessor = _DependentCallable[Any]
- StateParam: State 对象
- DefaultParam: 带有默认值的参数
"""
T_EventPostProcessor = _DependentCallable[Any]
T_EventPostProcessor: TypeAlias = _DependentCallable[Any]
"""事件预处理函数 EventPostProcessor 类型
依赖参数:
@ -103,7 +103,7 @@ T_EventPostProcessor = _DependentCallable[Any]
"""
# matcher run hooks
T_RunPreProcessor = _DependentCallable[Any]
T_RunPreProcessor: TypeAlias = _DependentCallable[Any]
"""事件响应器运行前预处理函数 RunPreProcessor 类型
依赖参数:
@ -115,7 +115,7 @@ T_RunPreProcessor = _DependentCallable[Any]
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_RunPostProcessor = _DependentCallable[Any]
T_RunPostProcessor: TypeAlias = _DependentCallable[Any]
"""事件响应器运行后后处理函数 RunPostProcessor 类型
依赖参数:
@ -130,7 +130,7 @@ T_RunPostProcessor = _DependentCallable[Any]
"""
# rule, permission
T_RuleChecker = _DependentCallable[bool]
T_RuleChecker: TypeAlias = _DependentCallable[bool]
"""RuleChecker 即判断是否响应事件的处理函数。
依赖参数:
@ -141,7 +141,7 @@ T_RuleChecker = _DependentCallable[bool]
- StateParam: State 对象
- DefaultParam: 带有默认值的参数
"""
T_PermissionChecker = _DependentCallable[bool]
T_PermissionChecker: TypeAlias = _DependentCallable[bool]
"""PermissionChecker 即判断事件是否满足权限的处理函数。
依赖参数:
@ -152,10 +152,11 @@ T_PermissionChecker = _DependentCallable[bool]
- DefaultParam: 带有默认值的参数
"""
T_Handler = _DependentCallable[Any]
T_Handler: TypeAlias = _DependentCallable[Any]
"""Handler 处理函数。"""
T_TypeUpdater = _DependentCallable[str]
"""TypeUpdater 在 Matcher.pause, Matcher.reject 时被运行,用于更新响应的事件类型。默认会更新为 `message`。
T_TypeUpdater: TypeAlias = _DependentCallable[str]
"""TypeUpdater 在 Matcher.pause, Matcher.reject 时被运行,用于更新响应的事件类型。
默认会更新为 `message`。
依赖参数:
@ -166,8 +167,9 @@ T_TypeUpdater = _DependentCallable[str]
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_PermissionUpdater = _DependentCallable["Permission"]
"""PermissionUpdater 在 Matcher.pause, Matcher.reject 时被运行,用于更新会话对象权限。默认会更新为当前事件的触发对象。
T_PermissionUpdater: TypeAlias = _DependentCallable["Permission"]
"""PermissionUpdater 在 Matcher.pause, Matcher.reject 时被运行,用于更新会话对象权限。
默认会更新为当前事件的触发对象。
依赖参数:
@ -178,5 +180,5 @@ T_PermissionUpdater = _DependentCallable["Permission"]
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_DependencyCache = Dict[_DependentCallable[Any], "Task[Any]"]
T_DependencyCache: TypeAlias = Dict[_DependentCallable[Any], "Task[Any]"]
"""依赖缓存, 用于存储依赖函数的返回值"""

View File

@ -63,7 +63,8 @@ def generic_check_issubclass(
- 如果 cls 是 `typing.Union` 或 `types.UnionType` 类型,
则会检查其中的所有类型是否是 class_or_tuple 中一个类型的子类或 None。
- 如果 cls 是 `typing.TypeVar` 类型,
则会检查其 `__bound__` 或 `__constraints__` 是否是 class_or_tuple 中一个类型的子类或 None。
则会检查其 `__bound__` 或 `__constraints__`
是否是 class_or_tuple 中一个类型的子类或 None。
"""
try:
return issubclass(cls, class_or_tuple)
@ -171,6 +172,17 @@ async def run_coro_with_catch(
exc: Tuple[Type[Exception], ...],
return_on_err: Optional[R] = None,
) -> Optional[Union[T, R]]:
"""运行协程并当遇到指定异常时返回指定值。
参数:
coro: 要运行的协程
exc: 要捕获的异常
return_on_err: 当发生异常时返回的值
返回:
协程的返回值或发生异常时的指定值
"""
try:
return await coro
except exc:
@ -210,7 +222,7 @@ def resolve_dot_notation(
class DataclassEncoder(json.JSONEncoder):
"""在JSON序列化 {ref}`nonebot.adapters.Message` (List[Dataclass]) 时使用的 `JSONEncoder`"""
"""可以序列化 {ref}`nonebot.adapters.Message`(List[Dataclass]) 的 `JSONEncoder`"""
@overrides(json.JSONEncoder)
def default(self, o):
@ -228,9 +240,11 @@ def logger_wrapper(logger_name: str):
返回:
日志记录函数
- level: 日志等级
- message: 日志信息
- exception: 异常信息
日志记录函数的参数:
- level: 日志等级
- message: 日志信息
- exception: 异常信息
"""
def log(level: str, message: str, exception: Optional[Exception] = None):