ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2025-09-06 20:16:47 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

💥 Remove: 移除 Python 3.8 支持 (#2641)

Browse Source
This commit is contained in:
Ju4tCode
2024-04-16 00:33:48 +08:00
committed by GitHub
parent e93ee1ffec
commit 4a02dde83f
69 changed files with 1811 additions and 1848 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
nonebot/__init__.py
View File

@ -45,7 +45,7 @@ FrontMatter:
import os
from importlib.metadata import version
from typing import Any, Dict, Type, Union, TypeVar, Optional, overload
from typing import Any, Union, TypeVar, Optional, overload
import loguru
@ -100,7 +100,7 @@ def get_adapter(name: str) -> Adapter:
@overload
def get_adapter(name: Type[A]) -> A:
def get_adapter(name: type[A]) -> A:
"""
参数:
name: 适配器类型
@ -110,7 +110,7 @@ def get_adapter(name: Type[A]) -> A:
"""
def get_adapter(name: Union[str, Type[Adapter]]) -> Adapter:
def get_adapter(name: Union[str, type[Adapter]]) -> Adapter:
"""获取已注册的 {ref}`nonebot.adapters.Adapter` 实例。
异常:
@ -131,7 +131,7 @@ def get_adapter(name: Union[str, Type[Adapter]]) -> Adapter:
return adapters[target]
def get_adapters() -> Dict[str, Adapter]:
def get_adapters() -> dict[str, Adapter]:
"""获取所有已注册的 {ref}`nonebot.adapters.Adapter` 实例。
返回:
@ -230,7 +230,7 @@ def get_bot(self_id: Optional[str] = None) -> Bot:
raise ValueError("There are no bots to get.")
def get_bots() -> Dict[str, Bot]:
def get_bots() -> dict[str, Bot]:
"""获取所有连接到 NoneBot 的 {ref}`nonebot.adapters.Bot` 对象。
返回:
@ -249,7 +249,7 @@ def get_bots() -> Dict[str, Bot]:
return get_driver().bots
def _resolve_combine_expr(obj_str: str) -> Type[Driver]:
def _resolve_combine_expr(obj_str: str) -> type[Driver]:
drivers = obj_str.split("+")
DriverClass = resolve_dot_notation(
drivers[0], "Driver", default_prefix="nonebot.drivers."

51
nonebot/compat.py
View File

@ -7,21 +7,18 @@ FrontMatter:
description: nonebot.compat 模块
"""
from collections.abc import Generator
from dataclasses import dataclass, is_dataclass
from typing_extensions import Self, Annotated, get_args, get_origin, is_typeddict
from typing_extensions import Self, get_args, get_origin, is_typeddict
from typing import (
TYPE_CHECKING,
Any,
Set,
Dict,
List,
Type,
Union,
TypeVar,
Callable,
Optional,
Protocol,
Generator,
Annotated,
)
from pydantic import VERSION, BaseModel
@ -94,7 +91,7 @@ if PYDANTIC_V2: # pragma: pydantic-v2
super().__init__(default=default, **kwargs)
@property
def extra(self) -> Dict[str, Any]:
def extra(self) -> dict[str, Any]:
"""Extra data that is not part of the standard pydantic fields.
For compatibility with pydantic v1.
@ -160,7 +157,7 @@ if PYDANTIC_V2: # pragma: pydantic-v2
# to allow store them in a set.
return id(self)
def extract_field_info(field_info: BaseFieldInfo) -> Dict[str, Any]:
def extract_field_info(field_info: BaseFieldInfo) -> dict[str, Any]:
"""Get FieldInfo init kwargs from a FieldInfo instance."""
kwargs = field_info._attributes_set.copy()
@ -176,7 +173,7 @@ if PYDANTIC_V2: # pragma: pydantic-v2
type, config=None if model_field._annotation_has_config() else config
).validate_python(value)
def model_fields(model: Type[BaseModel]) -> List[ModelField]:
def model_fields(model: type[BaseModel]) -> list[ModelField]:
"""Get field list of a model."""
return [
@ -188,19 +185,19 @@ if PYDANTIC_V2: # pragma: pydantic-v2
for name, field_info in model.model_fields.items()
]
def model_config(model: Type[BaseModel]) -> Any:
def model_config(model: type[BaseModel]) -> Any:
"""Get config of a model."""
return model.model_config
def model_dump(
model: BaseModel,
include: Optional[Set[str]] = None,
exclude: Optional[Set[str]] = None,
include: Optional[set[str]] = None,
exclude: Optional[set[str]] = None,
by_alias: bool = False,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
) -> Dict[str, Any]:
) -> dict[str, Any]:
return model.model_dump(
include=include,
exclude=exclude,
@ -210,16 +207,16 @@ if PYDANTIC_V2: # pragma: pydantic-v2
exclude_none=exclude_none,
)
def type_validate_python(type_: Type[T], data: Any) -> T:
def type_validate_python(type_: type[T], data: Any) -> T:
"""Validate data with given type."""
return TypeAdapter(type_).validate_python(data)
def type_validate_json(type_: Type[T], data: Union[str, bytes]) -> T:
def type_validate_json(type_: type[T], data: Union[str, bytes]) -> T:
"""Validate JSON with given type."""
return TypeAdapter(type_).validate_json(data)
def __get_pydantic_core_schema__(
cls: Type["_CustomValidationClass"],
cls: type["_CustomValidationClass"],
source_type: Any,
handler: GetCoreSchemaHandler,
) -> CoreSchema:
@ -230,7 +227,7 @@ if PYDANTIC_V2: # pragma: pydantic-v2
[core_schema.no_info_plain_validator_function(func) for func in validators]
)
def custom_validation(class_: Type["CVC"]) -> Type["CVC"]:
def custom_validation(class_: type["CVC"]) -> type["CVC"]:
"""Use pydantic v1 like validator generator in pydantic v2"""
setattr(
@ -308,7 +305,7 @@ else: # pragma: pydantic-v1
)
return cls._construct(name, annotation, field_info or FieldInfo())
def extract_field_info(field_info: BaseFieldInfo) -> Dict[str, Any]:
def extract_field_info(field_info: BaseFieldInfo) -> dict[str, Any]:
"""Get FieldInfo init kwargs from a FieldInfo instance."""
kwargs = {
@ -318,7 +315,7 @@ else: # pragma: pydantic-v1
return kwargs
def model_field_validate(
model_field: ModelField, value: Any, config: Optional[Type[ConfigDict]] = None
model_field: ModelField, value: Any, config: Optional[type[ConfigDict]] = None
) -> Any:
"""Validate the value pass to the field.
@ -333,7 +330,7 @@ else: # pragma: pydantic-v1
raise ValueError(value, model_field)
return v
def model_fields(model: Type[BaseModel]) -> List[ModelField]:
def model_fields(model: type[BaseModel]) -> list[ModelField]:
"""Get field list of a model."""
# construct the model field without preprocess to avoid error
@ -348,19 +345,19 @@ else: # pragma: pydantic-v1
for model_field in model.__fields__.values()
]
def model_config(model: Type[BaseModel]) -> Any:
def model_config(model: type[BaseModel]) -> Any:
"""Get config of a model."""
return model.__config__
def model_dump(
model: BaseModel,
include: Optional[Set[str]] = None,
exclude: Optional[Set[str]] = None,
include: Optional[set[str]] = None,
exclude: Optional[set[str]] = None,
by_alias: bool = False,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
) -> Dict[str, Any]:
) -> dict[str, Any]:
return model.dict(
include=include,
exclude=exclude,
@ -370,14 +367,14 @@ else: # pragma: pydantic-v1
exclude_none=exclude_none,
)
def type_validate_python(type_: Type[T], data: Any) -> T:
def type_validate_python(type_: type[T], data: Any) -> T:
"""Validate data with given type."""
return parse_obj_as(type_, data)
def type_validate_json(type_: Type[T], data: Union[str, bytes]) -> T:
def type_validate_json(type_: type[T], data: Union[str, bytes]) -> T:
"""Validate JSON with given type."""
return parse_raw_as(type_, data)
def custom_validation(class_: Type["CVC"]) -> Type["CVC"]:
def custom_validation(class_: type["CVC"]) -> type["CVC"]:
"""Do nothing in pydantic v1"""
return class_

66
nonebot/config.py
View File

@ -17,19 +17,9 @@ import json
from pathlib import Path
from datetime import timedelta
from ipaddress import IPv4Address
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Union, Optional
from typing_extensions import TypeAlias, get_args, get_origin
from typing import (
TYPE_CHECKING,
Any,
Set,
Dict,
List,
Type,
Tuple,
Union,
Mapping,
Optional,
)
from dotenv import dotenv_values
from pydantic import Field, BaseModel
@ -49,7 +39,7 @@ from nonebot.compat import (
)
DOTENV_TYPE: TypeAlias = Union[
Path, str, List[Union[Path, str]], Tuple[Union[Path, str], ...]
Path, str, list[Union[Path, str]], tuple[Union[Path, str], ...]
]
ENV_FILE_SENTINEL = Path("")
@ -59,7 +49,7 @@ class SettingsError(ValueError): ...
class BaseSettingsSource(abc.ABC):
def __init__(self, settings_cls: Type["BaseSettings"]) -> None:
def __init__(self, settings_cls: type["BaseSettings"]) -> None:
self.settings_cls = settings_cls
@property
@ -67,7 +57,7 @@ class BaseSettingsSource(abc.ABC):
return model_config(self.settings_cls)
@abc.abstractmethod
def __call__(self) -> Dict[str, Any]:
def __call__(self) -> dict[str, Any]:
raise NotImplementedError
@ -75,12 +65,12 @@ class InitSettingsSource(BaseSettingsSource):
__slots__ = ("init_kwargs",)
def __init__(
self, settings_cls: Type["BaseSettings"], init_kwargs: Dict[str, Any]
self, settings_cls: type["BaseSettings"], init_kwargs: dict[str, Any]
) -> None:
self.init_kwargs = init_kwargs
super().__init__(settings_cls)
def __call__(self) -> Dict[str, Any]:
def __call__(self) -> dict[str, Any]:
return self.init_kwargs
def __repr__(self) -> str:
@ -90,7 +80,7 @@ class InitSettingsSource(BaseSettingsSource):
class DotEnvSettingsSource(BaseSettingsSource):
def __init__(
self,
settings_cls: Type["BaseSettings"],
settings_cls: type["BaseSettings"],
env_file: Optional[DOTENV_TYPE] = ENV_FILE_SENTINEL,
env_file_encoding: Optional[str] = None,
case_sensitive: Optional[bool] = None,
@ -121,7 +111,7 @@ class DotEnvSettingsSource(BaseSettingsSource):
def _apply_case_sensitive(self, var_name: str) -> str:
return var_name if self.case_sensitive else var_name.lower()
def _field_is_complex(self, field: ModelField) -> Tuple[bool, bool]:
def _field_is_complex(self, field: ModelField) -> tuple[bool, bool]:
if type_is_complex(field.annotation):
return True, False
elif origin_is_union(get_origin(field.annotation)) and any(
@ -132,16 +122,16 @@ class DotEnvSettingsSource(BaseSettingsSource):
def _parse_env_vars(
self, env_vars: Mapping[str, Optional[str]]
) -> Dict[str, Optional[str]]:
) -> dict[str, Optional[str]]:
return {
self._apply_case_sensitive(key): value for key, value in env_vars.items()
}
def _read_env_file(self, file_path: Path) -> Dict[str, Optional[str]]:
def _read_env_file(self, file_path: Path) -> dict[str, Optional[str]]:
file_vars = dotenv_values(file_path, encoding=self.env_file_encoding)
return self._parse_env_vars(file_vars)
def _read_env_files(self) -> Dict[str, Optional[str]]:
def _read_env_files(self) -> dict[str, Optional[str]]:
env_files = self.env_file
if env_files is None:
return {}
@ -149,7 +139,7 @@ class DotEnvSettingsSource(BaseSettingsSource):
if isinstance(env_files, (str, os.PathLike)):
env_files = [env_files]
dotenv_vars: Dict[str, Optional[str]] = {}
dotenv_vars: dict[str, Optional[str]] = {}
for env_file in env_files:
env_path = Path(env_file).expanduser()
if env_path.is_file():
@ -170,14 +160,14 @@ class DotEnvSettingsSource(BaseSettingsSource):
def _explode_env_vars(
self,
field: ModelField,
env_vars: Dict[str, Optional[str]],
env_file_vars: Dict[str, Optional[str]],
) -> Dict[str, Any]:
env_vars: dict[str, Optional[str]],
env_file_vars: dict[str, Optional[str]],
) -> dict[str, Any]:
if self.env_nested_delimiter is None:
return {}
prefix = f"{field.name}{self.env_nested_delimiter}"
result: Dict[str, Any] = {}
result: dict[str, Any] = {}
for env_name, env_val in env_vars.items():
if not env_name.startswith(prefix):
continue
@ -209,10 +199,10 @@ class DotEnvSettingsSource(BaseSettingsSource):
return result
def __call__(self) -> Dict[str, Any]:
def __call__(self) -> dict[str, Any]:
"""从环境变量和 dotenv 配置文件中读取配置项。"""
d: Dict[str, Any] = {}
d: dict[str, Any] = {}
env_vars = self._parse_env_vars(os.environ)
env_file_vars = self._read_env_files()
@ -317,7 +307,7 @@ class BaseSettings(BaseModel):
return self.__dict__.get(name)
if PYDANTIC_V2: # pragma: pydantic-v2
model_config: SettingsConfig = SettingsConfig(
model_config = SettingsConfig(
extra="allow",
env_file=".env",
env_file_encoding="utf-8",
@ -351,11 +341,11 @@ class BaseSettings(BaseModel):
def _settings_build_values(
self,
init_kwargs: Dict[str, Any],
init_kwargs: dict[str, Any],
env_file: Optional[DOTENV_TYPE] = None,
env_file_encoding: Optional[str] = None,
env_nested_delimiter: Optional[str] = None,
) -> Dict[str, Any]:
) -> dict[str, Any]:
init_settings = InitSettingsSource(self.__class__, init_kwargs=init_kwargs)
env_settings = DotEnvSettingsSource(
self.__class__,
@ -426,7 +416,7 @@ class Config(BaseSettings):
"""API 请求超时时间，单位: 秒。"""
# bot runtime configs
superusers: Set[str] = set()
superusers: set[str] = set()
"""机器人超级用户。
用法:
@ -434,9 +424,9 @@ class Config(BaseSettings):
SUPERUSERS=["12345789"]
```
"""
nickname: Set[str] = set()
nickname: set[str] = set()
"""机器人昵称。"""
command_start: Set[str] = {"/"}
command_start: set[str] = {"/"}
"""命令的起始标记，用于判断一条消息是不是命令。
参考[命令响应规则](https://nonebot.dev/docs/advanced/matcher#command)。
@ -446,7 +436,7 @@ class Config(BaseSettings):
COMMAND_START=["/", ""]
```
"""
command_sep: Set[str] = {"."}
command_sep: set[str] = {"."}
"""命令的分隔标记，用于将文本形式的命令切分为元组（实际的命令名）。
参考[命令响应规则](https://nonebot.dev/docs/advanced/matcher#command)。
@ -477,7 +467,9 @@ class Config(BaseSettings):
model_config = SettingsConfig(env_file=(".env", ".env.prod"))
else: # pragma: pydantic-v1
class Config(SettingsConfig):
class Config( # pyright: ignore[reportIncompatibleVariableOverride]
SettingsConfig
):
env_file = ".env", ".env.prod"

42
nonebot/dependencies/__init__.py
View File

@ -9,20 +9,8 @@ import abc
import asyncio
import inspect
from dataclasses import field, dataclass
from typing import (
Any,
Dict,
List,
Type,
Tuple,
Generic,
TypeVar,
Callable,
Iterable,
Optional,
Awaitable,
cast,
)
from collections.abc import Iterable, Awaitable
from typing import Any, Generic, TypeVar, Callable, Optional, cast
from nonebot.log import logger
from nonebot.typing import _DependentCallable
@ -48,13 +36,13 @@ class Param(abc.ABC, FieldInfo):
@classmethod
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type["Param"], ...]
cls, param: inspect.Parameter, allow_types: tuple[type["Param"], ...]
) -> Optional["Param"]:
return
@classmethod
def _check_parameterless(
cls, value: Any, allow_types: Tuple[Type["Param"], ...]
cls, value: Any, allow_types: tuple[type["Param"], ...]
) -> Optional["Param"]:
return
@ -79,8 +67,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):
@ -112,9 +100,9 @@ class Dependent(Generic[R]):
@staticmethod
def parse_params(
call: _DependentCallable[R], allow_types: Tuple[Type[Param], ...]
) -> Tuple[ModelField, ...]:
fields: List[ModelField] = []
call: _DependentCallable[R], allow_types: tuple[type[Param], ...]
) -> tuple[ModelField, ...]:
fields: list[ModelField] = []
params = get_typed_signature(call).parameters.values()
for param in params:
@ -144,9 +132,9 @@ class Dependent(Generic[R]):
@staticmethod
def parse_parameterless(
parameterless: Tuple[Any, ...], allow_types: Tuple[Type[Param], ...]
) -> Tuple[Param, ...]:
parameterless_params: List[Param] = []
parameterless: tuple[Any, ...], allow_types: tuple[type[Param], ...]
) -> tuple[Param, ...]:
parameterless_params: list[Param] = []
for value in parameterless:
for allow_type in allow_types:
if param := allow_type._check_parameterless(value, allow_types):
@ -162,7 +150,7 @@ class Dependent(Generic[R]):
*,
call: _DependentCallable[R],
parameterless: Optional[Iterable[Any]] = None,
allow_types: Iterable[Type[Param]],
allow_types: Iterable[type[Param]],
) -> "Dependent[R]":
allow_types = tuple(allow_types)
@ -181,7 +169,7 @@ class Dependent(Generic[R]):
*(cast(Param, param.field_info)._check(**params) for param in self.params)
)
async def _solve_field(self, field: ModelField, params: Dict[str, Any]) -> Any:
async def _solve_field(self, field: ModelField, params: dict[str, Any]) -> Any:
param = cast(Param, field.field_info)
value = await param._solve(**params)
if value is PydanticUndefined:
@ -189,7 +177,7 @@ class Dependent(Generic[R]):
v = check_field_type(field, value)
return v if param.validate else value
async def solve(self, **params: Any) -> Dict[str, Any]:
async def solve(self, **params: Any) -> dict[str, Any]:
# solve parameterless
for param in self.parameterless:
await param._solve(**params)

4
nonebot/dependencies/utils.py
View File

@ -5,7 +5,7 @@ FrontMatter:
"""
import inspect
from typing import Any, Dict, Callable, ForwardRef
from typing import Any, Callable, ForwardRef
from loguru import logger
@ -31,7 +31,7 @@ def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
return inspect.Signature(typed_params)
def get_typed_annotation(param: inspect.Parameter, globalns: Dict[str, Any]) -> Any:
def get_typed_annotation(param: inspect.Parameter, globalns: dict[str, Any]) -> Any:
"""获取参数的类型注解"""
annotation = param.annotation

7
nonebot/drivers/aiohttp.py
View File

@ -16,8 +16,9 @@ FrontMatter:
"""
from typing_extensions import override
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Union, Optional, AsyncGenerator
from typing import TYPE_CHECKING, Union, Optional
from multidict import CIMultiDict
@ -221,8 +222,8 @@ class WebSocket(BaseWebSocket):
raise NotImplementedError
@override
async def close(self, code: int = 1000):
await self.websocket.close(code=code)
async def close(self, code: int = 1000, reason: str = ""):
await self.websocket.close(code=code, message=reason.encode("utf-8"))
await self.session.close()
async def _receive(self) -> aiohttp.WSMessage:

14
nonebot/drivers/fastapi.py
View File

@ -19,7 +19,7 @@ import logging
import contextlib
from functools import wraps
from typing_extensions import override
from typing import Any, Dict, List, Tuple, Union, Optional
from typing import Any, Union, Optional
from pydantic import BaseModel
@ -72,15 +72,15 @@ class Config(BaseModel):
"""是否包含适配器路由的 schema，默认为 `True`"""
fastapi_reload: bool = False
"""开启/关闭冷重载"""
fastapi_reload_dirs: Optional[List[str]] = None
fastapi_reload_dirs: Optional[list[str]] = None
"""重载监控文件夹列表，默认为 uvicorn 默认值"""
fastapi_reload_delay: float = 0.25
"""重载延迟，默认为 uvicorn 默认值"""
fastapi_reload_includes: Optional[List[str]] = None
fastapi_reload_includes: Optional[list[str]] = None
"""要监听的文件列表，支持 glob pattern，默认为 uvicorn 默认值"""
fastapi_reload_excludes: Optional[List[str]] = None
fastapi_reload_excludes: Optional[list[str]] = None
"""不要监听的文件列表，支持 glob pattern，默认为 uvicorn 默认值"""
fastapi_extra: Dict[str, Any] = {}
fastapi_extra: dict[str, Any] = {}
"""传递给 `FastAPI` 的其他参数。"""
@ -161,7 +161,7 @@ class Driver(BaseDriver, ASGIMixin):
self,
host: Optional[str] = None,
port: Optional[int] = None,
*,
*args,
app: Optional[str] = None,
**kwargs,
):
@ -206,7 +206,7 @@ class Driver(BaseDriver, ASGIMixin):
json = await request.json()
data: Optional[dict] = None
files: Optional[List[Tuple[str, FileTypes]]] = None
files: Optional[list[tuple[str, FileTypes]]] = None
with contextlib.suppress(Exception):
form = await request.form()
data = {}

14
nonebot/drivers/quart.py
View File

@ -18,7 +18,7 @@ FrontMatter:
import asyncio
from functools import wraps
from typing_extensions import override
from typing import Any, Dict, List, Tuple, Union, Optional, cast
from typing import Any, Union, Optional, cast
from pydantic import BaseModel
@ -64,15 +64,15 @@ class Config(BaseModel):
quart_reload: bool = False
"""开启/关闭冷重载"""
quart_reload_dirs: Optional[List[str]] = None
quart_reload_dirs: Optional[list[str]] = None
"""重载监控文件夹列表，默认为 uvicorn 默认值"""
quart_reload_delay: float = 0.25
"""重载延迟，默认为 uvicorn 默认值"""
quart_reload_includes: Optional[List[str]] = None
quart_reload_includes: Optional[list[str]] = None
"""要监听的文件列表，支持 glob pattern，默认为 uvicorn 默认值"""
quart_reload_excludes: Optional[List[str]] = None
quart_reload_excludes: Optional[list[str]] = None
"""不要监听的文件列表，支持 glob pattern，默认为 uvicorn 默认值"""
quart_extra: Dict[str, Any] = {}
quart_extra: dict[str, Any] = {}
"""传递给 `Quart` 的其他参数。"""
@ -142,7 +142,7 @@ class Driver(BaseDriver, ASGIMixin):
self,
host: Optional[str] = None,
port: Optional[int] = None,
*,
*args,
app: Optional[str] = None,
**kwargs,
):
@ -184,7 +184,7 @@ class Driver(BaseDriver, ASGIMixin):
data = await request.form
files_dict = await request.files
files: List[Tuple[str, FileTypes]] = []
files: list[tuple[str, FileTypes]] = []
key: str
value: FileStorage
for key, value in files_dict.items():

7
nonebot/drivers/websockets.py
View File

@ -19,7 +19,8 @@ import logging
from functools import wraps
from contextlib import asynccontextmanager
from typing_extensions import ParamSpec, override
from typing import TYPE_CHECKING, Union, TypeVar, Callable, Awaitable, AsyncGenerator
from collections.abc import Coroutine, AsyncGenerator
from typing import TYPE_CHECKING, Any, Union, TypeVar, Callable
from nonebot.drivers import Request
from nonebot.log import LoguruHandler
@ -44,7 +45,9 @@ logger = logging.Logger("websockets.client", "INFO")
logger.addHandler(LoguruHandler())
def catch_closed(func: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
def catch_closed(
func: Callable[P, Coroutine[Any, Any, T]]
) -> Callable[P, Coroutine[Any, Any, T]]:
@wraps(func)
async def decorator(*args: P.args, **kwargs: P.kwargs) -> T:
try:

5
nonebot/internal/adapter/adapter.py
View File

@ -1,6 +1,7 @@
import abc
from typing import Any
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import Any, Dict, AsyncGenerator
from nonebot.config import Config
from nonebot.internal.driver._lifespan import LIFESPAN_FUNC
@ -32,7 +33,7 @@ class Adapter(abc.ABC):
def __init__(self, driver: Driver, **kwargs: Any):
self.driver: Driver = driver
"""{ref}`nonebot.drivers.Driver` 实例"""
self.bots: Dict[str, Bot] = {}
self.bots: dict[str, Bot] = {}
"""本协议适配器已建立连接的 {ref}`nonebot.adapters.Bot` 实例"""
def __repr__(self) -> str:

6
nonebot/internal/adapter/bot.py
View File

@ -1,7 +1,7 @@
import abc
import asyncio
from functools import partial
from typing import TYPE_CHECKING, Any, Set, Union, ClassVar, Optional, Protocol
from typing import TYPE_CHECKING, Any, Union, ClassVar, Optional, Protocol
from nonebot.log import logger
from nonebot.config import Config
@ -27,9 +27,9 @@ class Bot(abc.ABC):
self_id: 机器人 ID
"""
_calling_api_hook: ClassVar[Set[T_CallingAPIHook]] = set()
_calling_api_hook: ClassVar[set[T_CallingAPIHook]] = set()
"""call_api 时执行的函数"""
_called_api_hook: ClassVar[Set[T_CalledAPIHook]] = set()
_called_api_hook: ClassVar[set[T_CalledAPIHook]] = set()
"""call_api 后执行的函数"""
def __init__(self, adapter: "Adapter", self_id: str):

4
nonebot/internal/adapter/event.py
View File

@ -1,5 +1,5 @@
import abc
from typing import Any, Type, TypeVar
from typing import Any, TypeVar
from pydantic import BaseModel
@ -25,7 +25,7 @@ class Event(abc.ABC, BaseModel):
if not PYDANTIC_V2: # pragma: pydantic-v1
@classmethod
def validate(cls: Type["E"], value: Any) -> "E":
def validate(cls: type["E"], value: Any) -> "E":
if isinstance(value, Event) and not isinstance(value, cls):
raise TypeError(f"{value} is incompatible with Event type {cls}")
return super().validate(value)

45
nonebot/internal/adapter/message.py
View File

@ -1,17 +1,14 @@
import abc
from copy import deepcopy
from typing_extensions import Self
from collections.abc import Iterable
from dataclasses import field, asdict, dataclass
from typing import (
from typing import ( # noqa: UP035
Any,
Dict,
List,
Type,
Tuple,
Union,
Generic,
TypeVar,
Iterable,
Optional,
SupportsIndex,
overload,
@ -32,12 +29,12 @@ class MessageSegment(abc.ABC, Generic[TM]):
type: str
"""消息段类型"""
data: Dict[str, Any] = field(default_factory=dict)
data: dict[str, Any] = field(default_factory=dict)
"""消息段数据"""
@classmethod
@abc.abstractmethod
def get_message_class(cls) -> Type[TM]:
def get_message_class(cls) -> Type[TM]: # noqa: UP006
"""获取消息数组类型"""
raise NotImplementedError
@ -49,7 +46,9 @@ class MessageSegment(abc.ABC, Generic[TM]):
def __len__(self) -> int:
return len(str(self))
def __ne__(self, other: Self) -> bool:
def __ne__( # pyright: ignore[reportIncompatibleMethodOverride]
self, other: Self
) -> bool:
return not self == other
def __add__(self: TMS, other: Union[str, TMS, Iterable[TMS]]) -> TM:
@ -101,7 +100,7 @@ class MessageSegment(abc.ABC, Generic[TM]):
@custom_validation
class Message(List[TMS], abc.ABC):
class Message(list[TMS], abc.ABC):
"""消息序列
参数:
@ -142,7 +141,7 @@ class Message(List[TMS], abc.ABC):
@classmethod
@abc.abstractmethod
def get_segment_class(cls) -> Type[TMS]:
def get_segment_class(cls) -> type[TMS]:
"""获取消息段类型"""
raise NotImplementedError
@ -177,7 +176,9 @@ class Message(List[TMS], abc.ABC):
"""构造消息数组"""
raise NotImplementedError
def __add__(self, other: Union[str, TMS, Iterable[TMS]]) -> Self:
def __add__( # pyright: ignore[reportIncompatibleMethodOverride]
self, other: Union[str, TMS, Iterable[TMS]]
) -> Self:
result = self.copy()
result += other
return result
@ -209,7 +210,7 @@ class Message(List[TMS], abc.ABC):
"""
@overload
def __getitem__(self, args: Tuple[str, int]) -> TMS:
def __getitem__(self, args: tuple[str, int]) -> TMS:
"""索引指定类型的消息段
参数:
@ -220,7 +221,7 @@ class Message(List[TMS], abc.ABC):
"""
@overload
def __getitem__(self, args: Tuple[str, slice]) -> Self:
def __getitem__(self, args: tuple[str, slice]) -> Self:
"""切片指定类型的消息段
参数:
@ -252,12 +253,12 @@ class Message(List[TMS], abc.ABC):
消息切片 `args`
"""
def __getitem__(
def __getitem__( # pyright: ignore[reportIncompatibleMethodOverride]
self,
args: Union[
str,
Tuple[str, int],
Tuple[str, slice],
tuple[str, int],
tuple[str, slice],
int,
slice,
],
@ -276,7 +277,9 @@ class Message(List[TMS], abc.ABC):
else:
raise ValueError("Incorrect arguments to slice") # pragma: no cover
def __contains__(self, value: Union[TMS, str]) -> bool:
def __contains__( # pyright: ignore[reportIncompatibleMethodOverride]
self, value: Union[TMS, str]
) -> bool:
"""检查消息段是否存在
参数:
@ -359,7 +362,9 @@ class Message(List[TMS], abc.ABC):
return all(seg.type == value for seg in self)
return all(seg == value for seg in self)
def append(self, obj: Union[str, TMS]) -> Self:
def append( # pyright: ignore[reportIncompatibleMethodOverride]
self, obj: Union[str, TMS]
) -> Self:
"""添加一个消息段到消息数组末尾。
参数:
@ -373,7 +378,9 @@ class Message(List[TMS], abc.ABC):
raise ValueError(f"Unexpected type: {type(obj)} {obj}") # pragma: no cover
return self
def extend(self, obj: Union[Self, Iterable[TMS]]) -> Self:
def extend( # pyright: ignore[reportIncompatibleMethodOverride]
self, obj: Union[Self, Iterable[TMS]]
) -> Self:
"""拼接一个消息数组或多个消息段到消息数组末尾。
参数:

38
nonebot/internal/adapter/template.py
View File

@ -1,21 +1,15 @@
import functools
from string import Formatter
from typing_extensions import TypeAlias
from collections.abc import Mapping, Sequence
from typing import (
TYPE_CHECKING,
Any,
Set,
Dict,
List,
Type,
Tuple,
Union,
Generic,
Mapping,
TypeVar,
Callable,
Optional,
Sequence,
cast,
overload,
)
@ -27,7 +21,7 @@ if TYPE_CHECKING:
def formatter_field_name_split(
field_name: str,
) -> Tuple[str, List[Tuple[bool, str]]]: ...
) -> tuple[str, list[tuple[bool, str]]]: ...
TM = TypeVar("TM", bound="Message")
@ -50,7 +44,7 @@ class MessageTemplate(Formatter, Generic[TF]):
def __init__(
self: "MessageTemplate[str]",
template: str,
factory: Type[str] = str,
factory: type[str] = str,
private_getattr: bool = False,
) -> None: ...
@ -58,19 +52,19 @@ class MessageTemplate(Formatter, Generic[TF]):
def __init__(
self: "MessageTemplate[TM]",
template: Union[str, TM],
factory: Type[TM],
factory: type[TM],
private_getattr: bool = False,
) -> None: ...
def __init__(
self,
template: Union[str, TM],
factory: Union[Type[str], Type[TM]] = str,
factory: Union[type[str], type[TM]] = str,
private_getattr: bool = False,
) -> None:
self.template: TF = template # type: ignore
self.factory: Type[TF] = factory # type: ignore
self.format_specs: Dict[str, FormatSpecFunc] = {}
self.factory: type[TF] = factory # type: ignore
self.format_specs: dict[str, FormatSpecFunc] = {}
self.private_getattr = private_getattr
def __repr__(self) -> str:
@ -85,7 +79,9 @@ class MessageTemplate(Formatter, Generic[TF]):
self.format_specs[name] = spec
return spec
def format(self, *args, **kwargs):
def format( # pyright: ignore[reportIncompatibleMethodOverride]
self, *args, **kwargs
) -> TF:
"""根据传入参数和模板生成消息对象"""
return self._format(args, kwargs)
@ -118,7 +114,7 @@ class MessageTemplate(Formatter, Generic[TF]):
self.check_unused_args(used_args, args, kwargs)
return cast(TF, full_message)
def vformat(
def vformat( # pyright: ignore[reportIncompatibleMethodOverride]
self,
format_string: str,
args: Sequence[Any],
@ -126,15 +122,15 @@ class MessageTemplate(Formatter, Generic[TF]):
) -> TF:
raise NotImplementedError("`vformat` has merged into `_format`")
def _vformat(
def _vformat( # pyright: ignore[reportIncompatibleMethodOverride]
self,
format_string: str,
args: Sequence[Any],
kwargs: Mapping[str, Any],
used_args: Set[Union[int, str]],
used_args: set[Union[int, str]],
auto_arg_index: int = 0,
) -> Tuple[TF, int]:
results: List[Any] = [self.factory()]
) -> tuple[TF, int]:
results: list[Any] = [self.factory()]
for literal_text, field_name, format_spec, conversion in self.parse(
format_string
@ -185,7 +181,7 @@ class MessageTemplate(Formatter, Generic[TF]):
def get_field(
self, field_name: str, args: Sequence[Any], kwargs: Mapping[str, Any]
) -> Tuple[Any, Union[int, str]]:
) -> tuple[Any, Union[int, str]]:
first, rest = formatter_field_name_split(field_name)
obj = self.get_value(first, args, kwargs)
@ -199,7 +195,7 @@ class MessageTemplate(Formatter, Generic[TF]):
def format_field(self, value: Any, format_spec: str) -> Any:
formatter: Optional[FormatSpecFunc] = self.format_specs.get(format_spec)
if formatter is None and not issubclass(self.factory, str):
segment_class: Type["MessageSegment"] = self.factory.get_segment_class()
segment_class: type["MessageSegment"] = self.factory.get_segment_class()
method = getattr(segment_class, format_spec, None)
if callable(method) and not cast(str, method.__name__).startswith("_"):
formatter = getattr(segment_class, format_spec)

11
nonebot/internal/driver/_lifespan.py
View File

@ -1,5 +1,6 @@
from collections.abc import Awaitable
from typing_extensions import TypeAlias
from typing import Any, List, Union, Callable, Awaitable, cast
from typing import Any, Union, Callable, cast
from nonebot.utils import run_sync, is_coroutine_callable
@ -10,9 +11,9 @@ LIFESPAN_FUNC: TypeAlias = Union[SYNC_LIFESPAN_FUNC, ASYNC_LIFESPAN_FUNC]
class Lifespan:
def __init__(self) -> None:
self._startup_funcs: List[LIFESPAN_FUNC] = []
self._ready_funcs: List[LIFESPAN_FUNC] = []
self._shutdown_funcs: List[LIFESPAN_FUNC] = []
self._startup_funcs: list[LIFESPAN_FUNC] = []
self._ready_funcs: list[LIFESPAN_FUNC] = []
self._shutdown_funcs: list[LIFESPAN_FUNC] = []
def on_startup(self, func: LIFESPAN_FUNC) -> LIFESPAN_FUNC:
self._startup_funcs.append(func)
@ -28,7 +29,7 @@ class Lifespan:
@staticmethod
async def _run_lifespan_func(
funcs: List[LIFESPAN_FUNC],
funcs: list[LIFESPAN_FUNC],
) -> None:
for func in funcs:
if is_coroutine_callable(func):

29
nonebot/internal/driver/abstract.py
View File

@ -1,19 +1,10 @@
import abc
import asyncio
from types import TracebackType
from collections.abc import AsyncGenerator
from typing_extensions import Self, TypeAlias
from contextlib import AsyncExitStack, asynccontextmanager
from typing import (
TYPE_CHECKING,
Any,
Set,
Dict,
Type,
Union,
ClassVar,
Optional,
AsyncGenerator,
)
from typing import TYPE_CHECKING, Any, Union, ClassVar, Optional
from nonebot.log import logger
from nonebot.config import Env, Config
@ -57,11 +48,11 @@ class Driver(abc.ABC):
config: 包含配置信息的 Config 对象
"""
_adapters: ClassVar[Dict[str, "Adapter"]] = {}
_adapters: ClassVar[dict[str, "Adapter"]] = {}
"""已注册的适配器列表"""
_bot_connection_hook: ClassVar[Set[Dependent[Any]]] = set()
_bot_connection_hook: ClassVar[set[Dependent[Any]]] = set()
"""Bot 连接建立时执行的函数"""
_bot_disconnection_hook: ClassVar[Set[Dependent[Any]]] = set()
_bot_disconnection_hook: ClassVar[set[Dependent[Any]]] = set()
"""Bot 连接断开时执行的函数"""
def __init__(self, env: Env, config: Config):
@ -69,8 +60,8 @@ class Driver(abc.ABC):
"""环境名称"""
self.config: Config = config
"""全局配置对象"""
self._bots: Dict[str, "Bot"] = {}
self._bot_tasks: Set[asyncio.Task] = set()
self._bots: dict[str, "Bot"] = {}
self._bot_tasks: set[asyncio.Task] = set()
self._lifespan = Lifespan()
def __repr__(self) -> str:
@ -80,11 +71,11 @@ class Driver(abc.ABC):
)
@property
def bots(self) -> Dict[str, "Bot"]:
def bots(self) -> dict[str, "Bot"]:
"""获取当前所有已连接的 Bot"""
return self._bots
def register_adapter(self, adapter: Type["Adapter"], **kwargs) -> None:
def register_adapter(self, adapter: type["Adapter"], **kwargs) -> None:
"""注册一个协议适配器
参数:
@ -279,7 +270,7 @@ class HTTPClientSession(abc.ABC):
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc_type: Optional[type[BaseException]],
exc: Optional[BaseException],
tb: Optional[TracebackType],
) -> None:

12
nonebot/internal/driver/combine.py
View File

@ -1,4 +1,4 @@
from typing import TYPE_CHECKING, Type, Union, TypeVar, overload
from typing import TYPE_CHECKING, Union, TypeVar, overload
from .abstract import Mixin, Driver
@ -10,16 +10,18 @@ if TYPE_CHECKING:
@overload
def combine_driver(driver: Type[D]) -> Type[D]: ...
def combine_driver(driver: type[D]) -> type[D]: ...
@overload
def combine_driver(driver: Type[D], *mixins: Type[Mixin]) -> Type["CombinedDriver"]: ...
def combine_driver(
driver: type[D], _m: type[Mixin], *mixins: type[Mixin]
) -> type["CombinedDriver"]: ...
def combine_driver(
driver: Type[D], *mixins: Type[Mixin]
) -> Union[Type[D], Type["CombinedDriver"]]:
driver: type[D], *mixins: type[Mixin]
) -> Union[type[D], type["CombinedDriver"]]:
"""将一个驱动器和多个混入类合并。"""
# check first
if not issubclass(driver, Driver):

50
nonebot/internal/driver/model.py
View File

@ -4,56 +4,44 @@ from enum import Enum
from dataclasses import dataclass
from typing_extensions import TypeAlias
from http.cookiejar import Cookie, CookieJar
from typing import (
IO,
Any,
Dict,
List,
Tuple,
Union,
Mapping,
Callable,
Iterator,
Optional,
Awaitable,
MutableMapping,
)
from typing import IO, Any, Union, Callable, Optional
from collections.abc import Mapping, Iterator, Awaitable, MutableMapping
from yarl import URL as URL
from multidict import CIMultiDict
RawURL: TypeAlias = Tuple[bytes, bytes, Optional[int], bytes]
RawURL: TypeAlias = tuple[bytes, bytes, Optional[int], bytes]
SimpleQuery: TypeAlias = Union[str, int, float]
QueryVariable: TypeAlias = Union[SimpleQuery, List[SimpleQuery]]
QueryVariable: TypeAlias = Union[SimpleQuery, list[SimpleQuery]]
QueryTypes: TypeAlias = Union[
None, str, Mapping[str, QueryVariable], List[Tuple[str, SimpleQuery]]
None, str, Mapping[str, QueryVariable], list[tuple[str, SimpleQuery]]
]
HeaderTypes: TypeAlias = Union[
None,
CIMultiDict[str],
Dict[str, str],
List[Tuple[str, str]],
dict[str, str],
list[tuple[str, str]],
]
CookieTypes: TypeAlias = Union[
None, "Cookies", CookieJar, Dict[str, str], List[Tuple[str, str]]
None, "Cookies", CookieJar, dict[str, str], list[tuple[str, str]]
]
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]]
FileType: TypeAlias = tuple[Optional[str], FileContent, Optional[str]]
FileTypes: TypeAlias = Union[
# file (or bytes)
FileContent,
# (filename, file (or bytes))
Tuple[Optional[str], FileContent],
tuple[Optional[str], FileContent],
# (filename, file (or bytes), content_type)
FileType,
]
FilesTypes: TypeAlias = Union[Dict[str, FileTypes], List[Tuple[str, FileTypes]], None]
FilesTypes: TypeAlias = Union[dict[str, FileTypes], list[tuple[str, FileTypes]], None]
class HTTPVersion(Enum):
@ -119,7 +107,7 @@ class Request:
self.content: ContentTypes = content
self.data: DataTypes = data
self.json: Any = json
self.files: Optional[List[Tuple[str, FileType]]] = None
self.files: Optional[list[tuple[str, FileType]]] = None
if files:
self.files = []
files_ = files.items() if isinstance(files, dict) else files
@ -257,7 +245,7 @@ class Cookies(MutableMapping):
)
self.jar.set_cookie(cookie)
def get(
def get( # pyright: ignore[reportIncompatibleMethodOverride]
self,
name: str,
default: Optional[str] = None,
@ -298,12 +286,14 @@ class Cookies(MutableMapping):
def clear(self, domain: Optional[str] = None, path: Optional[str] = None) -> None:
self.jar.clear(domain, path)
def update(self, cookies: CookieTypes = None) -> None:
def update( # pyright: ignore[reportIncompatibleMethodOverride]
self, cookies: CookieTypes = None
) -> None:
cookies = Cookies(cookies)
for cookie in cookies.jar:
self.jar.set_cookie(cookie)
def as_header(self, request: Request) -> Dict[str, str]:
def as_header(self, request: Request) -> dict[str, str]:
urllib_request = self._CookieCompatRequest(request)
self.jar.add_cookie_header(urllib_request)
return urllib_request.added_headers
@ -341,9 +331,11 @@ class Cookies(MutableMapping):
method=request.method,
)
self.request = request
self.added_headers: Dict[str, str] = {}
self.added_headers: dict[str, str] = {}
def add_unredirected_header(self, key: str, value: str) -> None:
def add_unredirected_header( # pyright: ignore[reportIncompatibleMethodOverride]
self, key: str, value: str
) -> None:
super().add_unredirected_header(key, value)
self.added_headers[key] = value

49
nonebot/internal/matcher/manager.py
View File

@ -1,18 +1,5 @@
from typing import (
TYPE_CHECKING,
List,
Type,
Tuple,
Union,
TypeVar,
Iterator,
KeysView,
Optional,
ItemsView,
ValuesView,
MutableMapping,
overload,
)
from typing import TYPE_CHECKING, Union, TypeVar, Optional, overload
from collections.abc import Iterator, KeysView, ItemsView, ValuesView, MutableMapping
from .provider import DEFAULT_PROVIDER_CLASS, MatcherProvider
@ -22,7 +9,7 @@ if TYPE_CHECKING:
T = TypeVar("T")
class MatcherManager(MutableMapping[int, List[Type["Matcher"]]]):
class MatcherManager(MutableMapping[int, list[type["Matcher"]]]):
"""事件响应器管理器
实现了常用字典操作，用于管理事件响应器。
@ -43,10 +30,10 @@ class MatcherManager(MutableMapping[int, List[Type["Matcher"]]]):
def __len__(self) -> int:
return len(self.provider)
def __getitem__(self, key: int) -> List[Type["Matcher"]]:
def __getitem__(self, key: int) -> list[type["Matcher"]]:
return self.provider[key]
def __setitem__(self, key: int, value: List[Type["Matcher"]]) -> None:
def __setitem__(self, key: int, value: list[type["Matcher"]]) -> None:
self.provider[key] = value
def __delitem__(self, key: int) -> None:
@ -58,41 +45,45 @@ class MatcherManager(MutableMapping[int, List[Type["Matcher"]]]):
def keys(self) -> KeysView[int]:
return self.provider.keys()
def values(self) -> ValuesView[List[Type["Matcher"]]]:
def values(self) -> ValuesView[list[type["Matcher"]]]:
return self.provider.values()
def items(self) -> ItemsView[int, List[Type["Matcher"]]]:
def items(self) -> ItemsView[int, list[type["Matcher"]]]:
return self.provider.items()
@overload
def get(self, key: int) -> Optional[List[Type["Matcher"]]]: ...
def get(self, key: int) -> Optional[list[type["Matcher"]]]: ...
@overload
def get(self, key: int, default: T) -> Union[List[Type["Matcher"]], T]: ...
def get(self, key: int, default: T) -> Union[list[type["Matcher"]], T]: ...
def get(
self, key: int, default: Optional[T] = None
) -> Optional[Union[List[Type["Matcher"]], T]]:
) -> Optional[Union[list[type["Matcher"]], T]]:
return self.provider.get(key, default)
def pop(self, key: int) -> List[Type["Matcher"]]:
def pop( # pyright: ignore[reportIncompatibleMethodOverride]
self, key: int
) -> list[type["Matcher"]]:
return self.provider.pop(key)
def popitem(self) -> Tuple[int, List[Type["Matcher"]]]:
def popitem(self) -> tuple[int, list[type["Matcher"]]]:
return self.provider.popitem()
def clear(self) -> None:
self.provider.clear()
def update(self, __m: MutableMapping[int, List[Type["Matcher"]]]) -> None:
def update( # pyright: ignore[reportIncompatibleMethodOverride]
self, __m: MutableMapping[int, list[type["Matcher"]]]
) -> None:
self.provider.update(__m)
def setdefault(
self, key: int, default: List[Type["Matcher"]]
) -> List[Type["Matcher"]]:
self, key: int, default: list[type["Matcher"]]
) -> list[type["Matcher"]]:
return self.provider.setdefault(key, default)
def set_provider(self, provider_class: Type[MatcherProvider]) -> None:
def set_provider(self, provider_class: type[MatcherProvider]) -> None:
"""设置事件响应器存储器
参数:

16
nonebot/internal/matcher/matcher.py
View File

@ -6,19 +6,17 @@ from types import ModuleType
from dataclasses import dataclass
from contextvars import ContextVar
from typing_extensions import Self
from collections.abc import Iterable
from datetime import datetime, timedelta
from contextlib import AsyncExitStack, contextmanager
from typing import (
from typing import ( # noqa: UP035
TYPE_CHECKING,
Any,
List,
Type,
Tuple,
Union,
TypeVar,
Callable,
ClassVar,
Iterable,
NoReturn,
Optional,
overload,
@ -141,7 +139,7 @@ class Matcher(metaclass=MatcherMeta):
"""事件响应器匹配规则"""
permission: ClassVar[Permission] = Permission()
"""事件响应器触发权限"""
handlers: ClassVar[List[Dependent[Any]]] = []
handlers: ClassVar[list[Dependent[Any]]] = []
"""事件响应器拥有的事件处理函数列表"""
priority: ClassVar[int] = 1
"""事件响应器优先级"""
@ -160,7 +158,7 @@ class Matcher(metaclass=MatcherMeta):
_default_permission_updater: ClassVar[Optional[Dependent[Permission]]] = None
"""事件响应器权限更新函数"""
HANDLER_PARAM_TYPES: ClassVar[Tuple[Type[Param], ...]] = (
HANDLER_PARAM_TYPES: ClassVar[tuple[Type[Param], ...]] = ( # noqa: UP006
DependParam,
BotParam,
EventParam,
@ -171,7 +169,7 @@ class Matcher(metaclass=MatcherMeta):
)
def __init__(self):
self.remain_handlers: List[Dependent[Any]] = self.handlers.copy()
self.remain_handlers: list[Dependent[Any]] = self.handlers.copy()
self.state = self._default_state.copy()
def __repr__(self) -> str:
@ -192,7 +190,7 @@ class Matcher(metaclass=MatcherMeta):
type_: str = "",
rule: Optional[Rule] = None,
permission: Optional[Permission] = None,
handlers: Optional[List[Union[T_Handler, Dependent[Any]]]] = None,
handlers: Optional[list[Union[T_Handler, Dependent[Any]]]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
@ -206,7 +204,7 @@ class Matcher(metaclass=MatcherMeta):
default_permission_updater: Optional[
Union[T_PermissionUpdater, Dependent[Permission]]
] = None,
) -> Type[Self]:
) -> Type[Self]: # noqa: UP006
"""
创建一个新的事件响应器，并存储至 `matchers <#matchers>`_

9
nonebot/internal/matcher/provider.py
View File

@ -1,12 +1,13 @@
import abc
from typing import TYPE_CHECKING
from collections import defaultdict
from typing import TYPE_CHECKING, List, Type, Mapping, MutableMapping
from collections.abc import Mapping, MutableMapping
if TYPE_CHECKING:
from .matcher import Matcher
class MatcherProvider(abc.ABC, MutableMapping[int, List[Type["Matcher"]]]):
class MatcherProvider(abc.ABC, MutableMapping[int, list[type["Matcher"]]]):
"""事件响应器存储器基类
参数:
@ -14,12 +15,12 @@ class MatcherProvider(abc.ABC, MutableMapping[int, List[Type["Matcher"]]]):
"""
@abc.abstractmethod
def __init__(self, matchers: Mapping[int, List[Type["Matcher"]]]):
def __init__(self, matchers: Mapping[int, list[type["Matcher"]]]):
raise NotImplementedError
class _DictProvider(defaultdict, MatcherProvider):
def __init__(self, matchers: Mapping[int, List[Type["Matcher"]]]):
def __init__(self, matchers: Mapping[int, list[type["Matcher"]]]):
super().__init__(list, matchers)

55
nonebot/internal/params.py
View File

@ -1,16 +1,15 @@
import asyncio
import inspect
from typing_extensions import Self, get_args, override, get_origin
from contextlib import AsyncExitStack, contextmanager, asynccontextmanager
from typing_extensions import Self, Annotated, get_args, override, get_origin
from typing import (
TYPE_CHECKING,
Any,
Type,
Tuple,
Union,
Literal,
Callable,
Optional,
Annotated,
cast,
)
@ -126,7 +125,7 @@ class DependParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
type_annotation, depends_inner = param.annotation, None
# extract type annotation and dependency from Annotated
@ -166,7 +165,7 @@ class DependParam(Param):
@classmethod
@override
def _check_parameterless(
cls, value: Any, allow_types: Tuple[Type[Param], ...]
cls, value: Any, allow_types: tuple[type[Param], ...]
) -> Optional["Param"]:
if isinstance(value, DependsInner):
assert value.dependency, "Dependency cannot be empty"
@ -249,7 +248,7 @@ class BotParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
from nonebot.adapters import Bot
@ -266,11 +265,15 @@ class BotParam(Param):
return cls()
@override
async def _solve(self, bot: "Bot", **kwargs: Any) -> Any:
async def _solve( # pyright: ignore[reportIncompatibleMethodOverride]
self, bot: "Bot", **kwargs: Any
) -> Any:
return bot
@override
async def _check(self, bot: "Bot", **kwargs: Any) -> None:
async def _check( # pyright: ignore[reportIncompatibleMethodOverride]
self, bot: "Bot", **kwargs: Any
) -> None:
if self.checker is not None:
check_field_type(self.checker, bot)
@ -299,7 +302,7 @@ class EventParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
from nonebot.adapters import Event
@ -316,11 +319,15 @@ class EventParam(Param):
return cls()
@override
async def _solve(self, event: "Event", **kwargs: Any) -> Any:
async def _solve( # pyright: ignore[reportIncompatibleMethodOverride]
self, event: "Event", **kwargs: Any
) -> Any:
return event
@override
async def _check(self, event: "Event", **kwargs: Any) -> Any:
async def _check( # pyright: ignore[reportIncompatibleMethodOverride]
self, event: "Event", **kwargs: Any
) -> Any:
if self.checker is not None:
check_field_type(self.checker, event)
@ -339,7 +346,7 @@ class StateParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
# param type is T_State
if param.annotation is T_State:
@ -349,7 +356,9 @@ class StateParam(Param):
return cls()
@override
async def _solve(self, state: T_State, **kwargs: Any) -> Any:
async def _solve( # pyright: ignore[reportIncompatibleMethodOverride]
self, state: T_State, **kwargs: Any
) -> Any:
return state
@ -377,7 +386,7 @@ class MatcherParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
from nonebot.matcher import Matcher
@ -394,11 +403,15 @@ class MatcherParam(Param):
return cls()
@override
async def _solve(self, matcher: "Matcher", **kwargs: Any) -> Any:
async def _solve( # pyright: ignore[reportIncompatibleMethodOverride]
self, matcher: "Matcher", **kwargs: Any
) -> Any:
return matcher
@override
async def _check(self, matcher: "Matcher", **kwargs: Any) -> Any:
async def _check( # pyright: ignore[reportIncompatibleMethodOverride]
self, matcher: "Matcher", **kwargs: Any
) -> Any:
if self.checker is not None:
check_field_type(self.checker, matcher)
@ -455,7 +468,7 @@ class ArgParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
if isinstance(param.default, ArgInner):
return cls(key=param.default.key or param.name, type=param.default.type)
@ -464,7 +477,9 @@ class ArgParam(Param):
if isinstance(arg, ArgInner):
return cls(key=arg.key or param.name, type=arg.type)
async def _solve(self, matcher: "Matcher", **kwargs: Any) -> Any:
async def _solve( # pyright: ignore[reportIncompatibleMethodOverride]
self, matcher: "Matcher", **kwargs: Any
) -> Any:
message = matcher.get_arg(self.key)
if message is None:
return message
@ -490,7 +505,7 @@ class ExceptionParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
# param type is Exception(s) or subclass(es) of Exception or None
if generic_check_issubclass(param.annotation, Exception):
@ -518,7 +533,7 @@ class DefaultParam(Param):
@classmethod
@override
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
cls, param: inspect.Parameter, allow_types: tuple[type[Param], ...]
) -> Optional[Self]:
if param.default != param.empty:
return cls(default=param.default)

8
nonebot/internal/permission.py
View File

@ -1,7 +1,7 @@
import asyncio
from typing_extensions import Self
from contextlib import AsyncExitStack
from typing import Set, List, Type, Tuple, Union, ClassVar, NoReturn, Optional
from typing import Union, ClassVar, NoReturn, Optional
from nonebot.dependencies import Dependent
from nonebot.utils import run_coro_with_catch
@ -30,7 +30,7 @@ class Permission:
__slots__ = ("checkers",)
HANDLER_PARAM_TYPES: ClassVar[List[Type[Param]]] = [
HANDLER_PARAM_TYPES: ClassVar[list[type[Param]]] = [
DependParam,
BotParam,
EventParam,
@ -38,7 +38,7 @@ class Permission:
]
def __init__(self, *checkers: Union[T_PermissionChecker, Dependent[bool]]) -> None:
self.checkers: Set[Dependent[bool]] = {
self.checkers: set[Dependent[bool]] = {
(
checker
if isinstance(checker, Dependent)
@ -122,7 +122,7 @@ class User:
__slots__ = ("users", "perm")
def __init__(
self, users: Tuple[str, ...], perm: Optional[Permission] = None
self, users: tuple[str, ...], perm: Optional[Permission] = None
) -> None:
self.users = users
self.perm = perm

6
nonebot/internal/rule.py
View File

@ -1,6 +1,6 @@
import asyncio
from contextlib import AsyncExitStack
from typing import Set, List, Type, Union, ClassVar, NoReturn, Optional
from typing import Union, ClassVar, NoReturn, Optional
from nonebot.dependencies import Dependent
from nonebot.exception import SkippedException
@ -28,7 +28,7 @@ class Rule:
__slots__ = ("checkers",)
HANDLER_PARAM_TYPES: ClassVar[List[Type[Param]]] = [
HANDLER_PARAM_TYPES: ClassVar[list[type[Param]]] = [
DependParam,
BotParam,
EventParam,
@ -37,7 +37,7 @@ class Rule:
]
def __init__(self, *checkers: Union[T_RuleChecker, Dependent[bool]]) -> None:
self.checkers: Set[Dependent[bool]] = {
self.checkers: set[Dependent[bool]] = {
(
checker
if isinstance(checker, Dependent)

18
nonebot/message.py
View File

@ -11,7 +11,7 @@ import asyncio
import contextlib
from datetime import datetime
from contextlib import AsyncExitStack
from typing import TYPE_CHECKING, Any, Set, Dict, Type, Optional
from typing import TYPE_CHECKING, Any, Optional
from nonebot.log import logger
from nonebot.rule import TrieRule
@ -46,10 +46,10 @@ from nonebot.internal.params import (
if TYPE_CHECKING:
from nonebot.adapters import Bot, Event
_event_preprocessors: Set[Dependent[Any]] = set()
_event_postprocessors: Set[Dependent[Any]] = set()
_run_preprocessors: Set[Dependent[Any]] = set()
_run_postprocessors: Set[Dependent[Any]] = set()
_event_preprocessors: set[Dependent[Any]] = set()
_event_postprocessors: set[Dependent[Any]] = set()
_run_preprocessors: set[Dependent[Any]] = set()
_run_postprocessors: set[Dependent[Any]] = set()
EVENT_PCS_PARAMS = (
DependParam,
@ -330,7 +330,7 @@ async def _apply_run_postprocessors(
async def _check_matcher(
Matcher: Type[Matcher],
Matcher: type[Matcher],
bot: "Bot",
event: "Event",
state: T_State,
@ -381,7 +381,7 @@ async def _check_matcher(
async def _run_matcher(
Matcher: Type[Matcher],
Matcher: type[Matcher],
bot: "Bot",
event: "Event",
state: T_State,
@ -446,7 +446,7 @@ async def _run_matcher(
async def check_and_run_matcher(
Matcher: Type[Matcher],
Matcher: type[Matcher],
bot: "Bot",
event: "Event",
state: T_State,
@ -505,7 +505,7 @@ async def handle_event(bot: "Bot", event: "Event") -> None:
if show_log:
logger.opt(colors=True).success(log_msg)
state: Dict[Any, Any] = {}
state: dict[Any, Any] = {}
dependency_cache: T_DependencyCache = {}
# create event scope context

36
nonebot/params.py
View File

@ -5,18 +5,8 @@ FrontMatter:
description: nonebot.params 模块
"""
from typing import (
Any,
Dict,
List,
Match,
Tuple,
Union,
Literal,
Callable,
Optional,
overload,
)
from re import Match
from typing import Any, Union, Literal, Callable, Optional, overload
from nonebot.typing import T_State
from nonebot.matcher import Matcher
@ -90,7 +80,7 @@ def _command(state: T_State) -> Message:
return state[PREFIX_KEY][CMD_KEY]
def Command() -> Tuple[str, ...]:
def Command() -> tuple[str, ...]:
"""消息命令元组"""
return Depends(_command)
@ -140,7 +130,7 @@ def ShellCommandArgs() -> Any:
return Depends(_shell_command_args, use_cache=False)
def _shell_command_argv(state: T_State) -> List[Union[str, MessageSegment]]:
def _shell_command_argv(state: T_State) -> list[Union[str, MessageSegment]]:
return state[SHELL_ARGV]
@ -159,11 +149,11 @@ def RegexMatched() -> Match[str]:
def _regex_str(
groups: Tuple[Union[str, int], ...]
) -> Callable[[T_State], Union[str, Tuple[Union[str, Any], ...], Any]]:
groups: tuple[Union[str, int], ...]
) -> Callable[[T_State], Union[str, tuple[Union[str, Any], ...], Any]]:
def _regex_str_dependency(
state: T_State,
) -> Union[str, Tuple[Union[str, Any], ...], Any]:
) -> Union[str, tuple[Union[str, Any], ...], Any]:
return _regex_matched(state).group(*groups)
return _regex_str_dependency
@ -180,28 +170,28 @@ def RegexStr(__group: Union[str, int]) -> Union[str, Any]: ...
@overload
def RegexStr(
__group1: Union[str, int], __group2: Union[str, int], *groups: Union[str, int]
) -> Tuple[Union[str, Any], ...]: ...
) -> tuple[Union[str, Any], ...]: ...
def RegexStr(*groups: Union[str, int]) -> Union[str, Tuple[Union[str, Any], ...], Any]:
def RegexStr(*groups: Union[str, int]) -> Union[str, tuple[Union[str, Any], ...], Any]:
"""正则匹配结果文本"""
return Depends(_regex_str(groups), use_cache=False)
def _regex_group(state: T_State) -> Tuple[Any, ...]:
def _regex_group(state: T_State) -> tuple[Any, ...]:
return _regex_matched(state).groups()
def RegexGroup() -> Tuple[Any, ...]:
def RegexGroup() -> tuple[Any, ...]:
"""正则匹配结果 group 元组"""
return Depends(_regex_group, use_cache=False)
def _regex_dict(state: T_State) -> Dict[str, Any]:
def _regex_dict(state: T_State) -> dict[str, Any]:
return _regex_matched(state).groupdict()
def RegexDict() -> Dict[str, Any]:
def RegexDict() -> dict[str, Any]:
"""正则匹配结果 group 字典"""
return Depends(_regex_dict, use_cache=False)

14
nonebot/plugin/__init__.py
View File

@ -39,7 +39,7 @@ FrontMatter:
from itertools import chain
from types import ModuleType
from contextvars import ContextVar
from typing import Set, Dict, List, Type, Tuple, TypeVar, Optional
from typing import TypeVar, Optional
from pydantic import BaseModel
@ -48,9 +48,9 @@ from nonebot.compat import model_dump, type_validate_python
C = TypeVar("C", bound=BaseModel)
_plugins: Dict[str, "Plugin"] = {}
_managers: List["PluginManager"] = []
_current_plugin_chain: ContextVar[Tuple["Plugin", ...]] = ContextVar(
_plugins: dict[str, "Plugin"] = {}
_managers: list["PluginManager"] = []
_current_plugin_chain: ContextVar[tuple["Plugin", ...]] = ContextVar(
"_current_plugin_chain", default=()
)
@ -105,17 +105,17 @@ def get_plugin_by_module_name(module_name: str) -> Optional["Plugin"]:
module_name, *has_parent = module_name.rsplit(".", 1)
def get_loaded_plugins() -> Set["Plugin"]:
def get_loaded_plugins() -> set["Plugin"]:
"""获取当前已导入的所有插件。"""
return set(_plugins.values())
def get_available_plugin_names() -> Set[str]:
def get_available_plugin_names() -> set[str]:
"""获取当前所有可用的插件名（包含尚未加载的插件）。"""
return {*chain.from_iterable(manager.available_plugins for manager in _managers)}
def get_plugin_config(config: Type[C]) -> C:
def get_plugin_config(config: type[C]) -> C:
"""从全局配置获取当前插件需要的配置项。"""
return type_validate_python(config, model_dump(get_driver().config))

17
nonebot/plugin/load.py
View File

@ -8,7 +8,8 @@ FrontMatter:
import json
from pathlib import Path
from types import ModuleType
from typing import Set, Union, Iterable, Optional
from typing import Union, Optional
from collections.abc import Iterable
from nonebot.utils import path_to_module_name
@ -39,7 +40,7 @@ def load_plugin(module_path: Union[str, Path]) -> Optional[Plugin]:
return manager.load_plugin(module_path)
def load_plugins(*plugin_dir: str) -> Set[Plugin]:
def load_plugins(*plugin_dir: str) -> set[Plugin]:
"""导入文件夹下多个插件，以 `_` 开头的插件不会被导入!
参数:
@ -52,7 +53,7 @@ def load_plugins(*plugin_dir: str) -> Set[Plugin]:
def load_all_plugins(
module_path: Iterable[str], plugin_dir: Iterable[str]
) -> Set[Plugin]:
) -> set[Plugin]:
"""导入指定列表中的插件以及指定目录下多个插件，以 `_` 开头的插件不会被导入!
参数:
@ -64,7 +65,7 @@ def load_all_plugins(
return manager.load_all_plugins()
def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
def load_from_json(file_path: str, encoding: str = "utf-8") -> set[Plugin]:
"""导入指定 json 文件中的 `plugins` 以及 `plugin_dirs` 下多个插件。
以 `_` 开头的插件不会被导入!
@ -95,7 +96,7 @@ def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
return load_all_plugins(set(plugins), set(plugin_dirs))
def load_from_toml(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` 下多个插件。
以 `_` 开头的插件不会被导入!
@ -139,7 +140,7 @@ def load_builtin_plugin(name: str) -> Optional[Plugin]:
return load_plugin(f"nonebot.plugins.{name}")
def load_builtin_plugins(*plugins: str) -> Set[Plugin]:
def load_builtin_plugins(*plugins: str) -> set[Plugin]:
"""导入多个 NoneBot 内置插件。
参数:
@ -184,7 +185,7 @@ def require(name: str) -> ModuleType:
return plugin.module
def inherit_supported_adapters(*names: str) -> Optional[Set[str]]:
def inherit_supported_adapters(*names: str) -> Optional[set[str]]:
"""获取已加载插件的适配器支持状态集合。
如果传入了多个插件名称，返回值会自动取交集。
@ -196,7 +197,7 @@ def inherit_supported_adapters(*names: str) -> Optional[Set[str]]:
RuntimeError: 插件未加载
ValueError: 插件缺少元数据
"""
final_supported: Optional[Set[str]] = None
final_supported: Optional[set[str]] = None
for name in names:
plugin = get_plugin(_module_name_to_plugin_name(name))

29
nonebot/plugin/manager.py
View File

@ -12,10 +12,11 @@ import pkgutil
import importlib
from pathlib import Path
from itertools import chain
from typing import Optional
from types import ModuleType
from importlib.abc import MetaPathFinder
from collections.abc import Iterable, Sequence
from importlib.machinery import PathFinder, SourceFileLoader
from typing import Set, Dict, List, Iterable, Optional, Sequence
from nonebot.log import logger
from nonebot.utils import escape_tag, path_to_module_name
@ -44,34 +45,34 @@ class PluginManager:
search_path: Optional[Iterable[str]] = None,
):
# simple plugin not in search path
self.plugins: Set[str] = set(plugins or [])
self.search_path: Set[str] = set(search_path or [])
self.plugins: set[str] = set(plugins or [])
self.search_path: set[str] = set(search_path or [])
# cache plugins
self._third_party_plugin_names: Dict[str, str] = {}
self._searched_plugin_names: Dict[str, Path] = {}
self._third_party_plugin_names: dict[str, str] = {}
self._searched_plugin_names: dict[str, Path] = {}
self.prepare_plugins()
def __repr__(self) -> str:
return f"PluginManager(plugins={self.plugins}, search_path={self.search_path})"
@property
def third_party_plugins(self) -> Set[str]:
def third_party_plugins(self) -> set[str]:
"""返回所有独立插件名称。"""
return set(self._third_party_plugin_names.keys())
@property
def searched_plugins(self) -> Set[str]:
def searched_plugins(self) -> set[str]:
"""返回已搜索到的插件名称。"""
return set(self._searched_plugin_names.keys())
@property
def available_plugins(self) -> Set[str]:
def available_plugins(self) -> set[str]:
"""返回当前插件管理器中可用的插件名称。"""
return self.third_party_plugins | self.searched_plugins
def _previous_plugins(self) -> Set[str]:
_pre_managers: List[PluginManager]
def _previous_plugins(self) -> set[str]:
_pre_managers: list[PluginManager]
if self in _managers:
_pre_managers = _managers[: _managers.index(self)]
else:
@ -81,12 +82,12 @@ class PluginManager:
*chain.from_iterable(manager.available_plugins for manager in _pre_managers)
}
def prepare_plugins(self) -> Set[str]:
def prepare_plugins(self) -> set[str]:
"""搜索插件并缓存插件名称。"""
# get all previous ready to load plugins
previous_plugins = self._previous_plugins()
searched_plugins: Dict[str, Path] = {}
third_party_plugins: Dict[str, str] = {}
searched_plugins: dict[str, Path] = {}
third_party_plugins: dict[str, str] = {}
# check third party plugins
for plugin in self.plugins:
@ -170,7 +171,7 @@ class PluginManager:
f'<r><bg #f8bbd0>Failed to import "{escape_tag(name)}"</bg #f8bbd0></r>'
)
def load_all_plugins(self) -> Set[Plugin]:
def load_all_plugins(self) -> set[Plugin]:
"""加载所有可用插件。"""
return set(

14
nonebot/plugin/model.py
View File

@ -8,7 +8,7 @@ FrontMatter:
import contextlib
from types import ModuleType
from dataclasses import field, dataclass
from typing import TYPE_CHECKING, Any, Set, Dict, Type, Optional
from typing import TYPE_CHECKING, Any, Type, Optional # noqa: UP035
from pydantic import BaseModel
@ -35,19 +35,19 @@ class PluginMetadata:
"""插件类型，用于商店分类"""
homepage: Optional[str] = None
"""插件主页"""
config: Optional[Type[BaseModel]] = None
config: Optional[Type[BaseModel]] = None # noqa: UP006
"""插件配置项"""
supported_adapters: Optional[Set[str]] = None
supported_adapters: Optional[set[str]] = None
"""插件支持的适配器模块路径
格式为 `<module>[:<Adapter>]`，`~` 为 `nonebot.adapters.` 的缩写。
`None` 表示支持**所有适配器**。
"""
extra: Dict[Any, Any] = field(default_factory=dict)
extra: dict[Any, Any] = field(default_factory=dict)
"""插件额外信息，可由插件编写者自由扩展定义"""
def get_supported_adapters(self) -> Optional[Set[Type["Adapter"]]]:
def get_supported_adapters(self) -> Optional[set[Type["Adapter"]]]: # noqa: UP006
"""获取当前已安装的插件支持适配器类列表"""
if self.supported_adapters is None:
return None
@ -73,10 +73,10 @@ class Plugin:
"""点分割模块路径"""
manager: "PluginManager"
"""导入该插件的插件管理器"""
matcher: Set[Type[Matcher]] = field(default_factory=set)
matcher: set[type[Matcher]] = field(default_factory=set)
"""插件加载时定义的 `Matcher`"""
parent_plugin: Optional["Plugin"] = None
"""父插件"""
sub_plugins: Set["Plugin"] = field(default_factory=set)
sub_plugins: set["Plugin"] = field(default_factory=set)
"""子插件集合"""
metadata: Optional[PluginMetadata] = None

106
nonebot/plugin/on.py
View File

@ -9,8 +9,8 @@ import re
import inspect
import warnings
from types import ModuleType
from typing import Any, Union, Optional
from datetime import datetime, timedelta
from typing import Any, Set, Dict, List, Type, Tuple, Union, Optional
from nonebot.adapters import Event
from nonebot.permission import Permission
@ -35,7 +35,7 @@ from . import get_plugin_by_module_name
from .manager import _current_plugin_chain
def store_matcher(matcher: Type[Matcher]) -> None:
def store_matcher(matcher: type[Matcher]) -> None:
"""存储一个事件响应器到插件。
参数:
@ -109,14 +109,14 @@ def on(
rule: Optional[Union[Rule, T_RuleChecker]] = None,
permission: Optional[Union[Permission, T_PermissionChecker]] = None,
*,
handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
handlers: Optional[list[Union[T_Handler, Dependent]]] = None,
temp: bool = False,
expire_time: Optional[Union[datetime, timedelta]] = None,
priority: int = 1,
block: bool = False,
state: Optional[T_State] = None,
_depth: int = 0,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个基础事件响应器，可自定义类型。
参数:
@ -146,7 +146,7 @@ def on(
return matcher
def on_metaevent(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
def on_metaevent(*args, _depth: int = 0, **kwargs) -> type[Matcher]:
"""注册一个元事件响应器。
参数:
@ -162,7 +162,7 @@ def on_metaevent(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
return on("meta_event", *args, **kwargs, _depth=_depth + 1)
def on_message(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
def on_message(*args, _depth: int = 0, **kwargs) -> type[Matcher]:
"""注册一个消息事件响应器。
参数:
@ -179,7 +179,7 @@ def on_message(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
return on("message", *args, **kwargs, _depth=_depth + 1)
def on_notice(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
def on_notice(*args, _depth: int = 0, **kwargs) -> type[Matcher]:
"""注册一个通知事件响应器。
参数:
@ -195,7 +195,7 @@ def on_notice(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
return on("notice", *args, **kwargs, _depth=_depth + 1)
def on_request(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
def on_request(*args, _depth: int = 0, **kwargs) -> type[Matcher]:
"""注册一个请求事件响应器。
参数:
@ -212,12 +212,12 @@ def on_request(*args, _depth: int = 0, **kwargs) -> Type[Matcher]:
def on_startswith(
msg: Union[str, Tuple[str, ...]],
msg: Union[str, tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
ignorecase: bool = False,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息的**文本部分**以指定内容开头时响应。
参数:
@ -236,12 +236,12 @@ def on_startswith(
def on_endswith(
msg: Union[str, Tuple[str, ...]],
msg: Union[str, tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
ignorecase: bool = False,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息的**文本部分**以指定内容结尾时响应。
参数:
@ -260,12 +260,12 @@ def on_endswith(
def on_fullmatch(
msg: Union[str, Tuple[str, ...]],
msg: Union[str, tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
ignorecase: bool = False,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息的**文本部分**与指定内容完全一致时响应。
参数:
@ -284,11 +284,11 @@ def on_fullmatch(
def on_keyword(
keywords: Set[str],
keywords: set[str],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息纯文本部分包含关键词时响应。
参数:
@ -306,13 +306,13 @@ def on_keyword(
def on_command(
cmd: Union[str, Tuple[str, ...]],
cmd: Union[str, tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None,
aliases: Optional[set[Union[str, tuple[str, ...]]]] = None,
force_whitespace: Optional[Union[str, bool]] = None,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息以指定命令开头时响应。
命令匹配规则参考: `命令形式匹配 <rule.md#command-command>`_
@ -341,13 +341,13 @@ def on_command(
def on_shell_command(
cmd: Union[str, Tuple[str, ...]],
cmd: Union[str, tuple[str, ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None,
aliases: Optional[set[Union[str, tuple[str, ...]]]] = None,
parser: Optional[ArgumentParser] = None,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个支持 `shell_like` 解析参数的命令消息事件响应器。
与普通的 `on_command` 不同的是，在添加 `parser` 参数时, 响应器会自动处理消息。
@ -383,7 +383,7 @@ def on_regex(
rule: Optional[Union[Rule, T_RuleChecker]] = None,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息匹配正则表达式时响应。
命令匹配规则参考: `正则匹配 <rule.md#regex-regex-flags-0>`_
@ -404,12 +404,12 @@ def on_regex(
def on_type(
types: Union[Type[Event], Tuple[Type[Event], ...]],
types: Union[type[Event], tuple[type[Event], ...]],
rule: Optional[Union[Rule, T_RuleChecker]] = None,
*,
_depth: int = 0,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个事件响应器，并且当事件为指定类型时响应。
参数:
@ -430,14 +430,14 @@ def on_type(
class _Group:
def __init__(self, **kwargs):
"""创建一个事件响应器组合，参数为默认值，与 `on` 一致"""
self.matchers: List[Type[Matcher]] = []
self.matchers: list[type[Matcher]] = []
"""组内事件响应器列表"""
self.base_kwargs: Dict[str, Any] = kwargs
self.base_kwargs: dict[str, Any] = kwargs
"""其他传递给 `on` 的参数默认值"""
def _get_final_kwargs(
self, update: Dict[str, Any], *, exclude: Optional[Set[str]] = None
) -> Dict[str, Any]:
self, update: dict[str, Any], *, exclude: Optional[set[str]] = None
) -> dict[str, Any]:
"""获取最终传递给 `on` 的参数
参数:
@ -470,18 +470,18 @@ class CommandGroup(_Group):
"""
def __init__(
self, cmd: Union[str, Tuple[str, ...]], prefix_aliases: bool = False, **kwargs
self, cmd: Union[str, tuple[str, ...]], prefix_aliases: bool = False, **kwargs
):
"""命令前缀"""
super().__init__(**kwargs)
self.basecmd: Tuple[str, ...] = (cmd,) if isinstance(cmd, str) else cmd
self.basecmd: tuple[str, ...] = (cmd,) if isinstance(cmd, str) else cmd
self.base_kwargs.pop("aliases", None)
self.prefix_aliases = prefix_aliases
def __repr__(self) -> str:
return f"CommandGroup(cmd={self.basecmd}, matchers={len(self.matchers)})"
def command(self, cmd: Union[str, Tuple[str, ...]], **kwargs) -> Type[Matcher]:
def command(self, cmd: Union[str, tuple[str, ...]], **kwargs) -> type[Matcher]:
"""注册一个新的命令。新参数将会覆盖命令组默认值
参数:
@ -509,8 +509,8 @@ class CommandGroup(_Group):
return matcher
def shell_command(
self, cmd: Union[str, Tuple[str, ...]], **kwargs
) -> Type[Matcher]:
self, cmd: Union[str, tuple[str, ...]], **kwargs
) -> type[Matcher]:
"""注册一个新的 `shell_like` 命令。新参数将会覆盖命令组默认值
参数:
@ -544,7 +544,7 @@ class MatcherGroup(_Group):
def __repr__(self) -> str:
return f"MatcherGroup(matchers={len(self.matchers)})"
def on(self, **kwargs) -> Type[Matcher]:
def on(self, **kwargs) -> type[Matcher]:
"""注册一个基础事件响应器，可自定义类型。
参数:
@ -562,7 +562,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_metaevent(self, **kwargs) -> Type[Matcher]:
def on_metaevent(self, **kwargs) -> type[Matcher]:
"""注册一个元事件响应器。
参数:
@ -580,7 +580,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_message(self, **kwargs) -> Type[Matcher]:
def on_message(self, **kwargs) -> type[Matcher]:
"""注册一个消息事件响应器。
参数:
@ -598,7 +598,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_notice(self, **kwargs) -> Type[Matcher]:
def on_notice(self, **kwargs) -> type[Matcher]:
"""注册一个通知事件响应器。
参数:
@ -616,7 +616,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_request(self, **kwargs) -> Type[Matcher]:
def on_request(self, **kwargs) -> type[Matcher]:
"""注册一个请求事件响应器。
参数:
@ -635,8 +635,8 @@ class MatcherGroup(_Group):
return matcher
def on_startswith(
self, msg: Union[str, Tuple[str, ...]], **kwargs
) -> Type[Matcher]:
self, msg: Union[str, tuple[str, ...]], **kwargs
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息的**文本部分**以指定内容开头时响应。
参数:
@ -656,7 +656,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_endswith(self, msg: Union[str, Tuple[str, ...]], **kwargs) -> Type[Matcher]:
def on_endswith(self, msg: Union[str, tuple[str, ...]], **kwargs) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息的**文本部分**以指定内容结尾时响应。
参数:
@ -676,7 +676,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_fullmatch(self, msg: Union[str, Tuple[str, ...]], **kwargs) -> Type[Matcher]:
def on_fullmatch(self, msg: Union[str, tuple[str, ...]], **kwargs) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息的**文本部分**与指定内容完全一致时响应。
参数:
@ -696,7 +696,7 @@ class MatcherGroup(_Group):
self.matchers.append(matcher)
return matcher
def on_keyword(self, keywords: Set[str], **kwargs) -> Type[Matcher]:
def on_keyword(self, keywords: set[str], **kwargs) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息纯文本部分包含关键词时响应。
参数:
@ -717,11 +717,11 @@ class MatcherGroup(_Group):
def on_command(
self,
cmd: Union[str, Tuple[str, ...]],
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None,
cmd: Union[str, tuple[str, ...]],
aliases: Optional[set[Union[str, tuple[str, ...]]]] = None,
force_whitespace: Optional[Union[str, bool]] = None,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息以指定命令开头时响应。
命令匹配规则参考: `命令形式匹配 <rule.md#command-command>`_
@ -748,11 +748,11 @@ class MatcherGroup(_Group):
def on_shell_command(
self,
cmd: Union[str, Tuple[str, ...]],
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None,
cmd: Union[str, tuple[str, ...]],
aliases: Optional[set[Union[str, tuple[str, ...]]]] = None,
parser: Optional[ArgumentParser] = None,
**kwargs,
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个支持 `shell_like` 解析参数的命令消息事件响应器。
与普通的 `on_command` 不同的是，在添加 `parser` 参数时, 响应器会自动处理消息。
@ -780,7 +780,7 @@ class MatcherGroup(_Group):
def on_regex(
self, pattern: str, flags: Union[int, re.RegexFlag] = 0, **kwargs
) -> Type[Matcher]:
) -> type[Matcher]:
"""注册一个消息事件响应器，并且当消息匹配正则表达式时响应。
命令匹配规则参考: `正则匹配 <rule.md#regex-regex-flags-0>`_
@ -803,8 +803,8 @@ class MatcherGroup(_Group):
return matcher
def on_type(
self, types: Union[Type[Event], Tuple[Type[Event]]], **kwargs
) -> Type[Matcher]:
self, types: Union[type[Event], tuple[type[Event]]], **kwargs
) -> type[Matcher]:
"""注册一个事件响应器，并且当事件为指定类型时响应。
参数:

4
nonebot/plugins/single_session.py
View File

@ -1,4 +1,4 @@
from typing import Dict, AsyncGenerator
from collections.abc import AsyncGenerator
from nonebot.adapters import Event
from nonebot.params import Depends
@ -15,7 +15,7 @@ __plugin_meta__ = PluginMetadata(
supported_adapters=None,
)
_running_matcher: Dict[str, int] = {}
_running_matcher: dict[str, int] = {}
async def matcher_mutex(event: Event) -> AsyncGenerator[bool, None]:

55
nonebot/rule.py
View File

@ -15,19 +15,16 @@ from argparse import Action
from gettext import gettext
from argparse import ArgumentError
from contextvars import ContextVar
from collections.abc import Sequence
from itertools import chain, product
from argparse import Namespace as Namespace
from argparse import ArgumentParser as ArgParser
from typing import (
IO,
TYPE_CHECKING,
List,
Type,
Tuple,
Union,
TypeVar,
Optional,
Sequence,
TypedDict,
NamedTuple,
cast,
@ -63,7 +60,7 @@ T = TypeVar("T")
class CMD_RESULT(TypedDict):
command: Optional[Tuple[str, ...]]
command: Optional[tuple[str, ...]]
raw_command: Optional[str]
command_arg: Optional[Message]
command_start: Optional[str]
@ -72,7 +69,7 @@ class CMD_RESULT(TypedDict):
class TRIE_VALUE(NamedTuple):
command_start: str
command: Tuple[str, ...]
command: tuple[str, ...]
parser_message: ContextVar[str] = ContextVar("parser_message")
@ -149,7 +146,7 @@ class StartswithRule:
__slots__ = ("msg", "ignorecase")
def __init__(self, msg: Tuple[str, ...], ignorecase: bool = False):
def __init__(self, msg: tuple[str, ...], ignorecase: bool = False):
self.msg = msg
self.ignorecase = ignorecase
@ -181,7 +178,7 @@ class StartswithRule:
return False
def startswith(msg: Union[str, Tuple[str, ...]], ignorecase: bool = False) -> Rule:
def startswith(msg: Union[str, tuple[str, ...]], ignorecase: bool = False) -> Rule:
"""匹配消息纯文本开头。
参数:
@ -204,7 +201,7 @@ class EndswithRule:
__slots__ = ("msg", "ignorecase")
def __init__(self, msg: Tuple[str, ...], ignorecase: bool = False):
def __init__(self, msg: tuple[str, ...], ignorecase: bool = False):
self.msg = msg
self.ignorecase = ignorecase
@ -236,7 +233,7 @@ class EndswithRule:
return False
def endswith(msg: Union[str, Tuple[str, ...]], ignorecase: bool = False) -> Rule:
def endswith(msg: Union[str, tuple[str, ...]], ignorecase: bool = False) -> Rule:
"""匹配消息纯文本结尾。
参数:
@ -259,7 +256,7 @@ class FullmatchRule:
__slots__ = ("msg", "ignorecase")
def __init__(self, msg: Tuple[str, ...], ignorecase: bool = False):
def __init__(self, msg: tuple[str, ...], ignorecase: bool = False):
self.msg = tuple(map(str.casefold, msg) if ignorecase else msg)
self.ignorecase = ignorecase
@ -290,7 +287,7 @@ class FullmatchRule:
return False
def fullmatch(msg: Union[str, Tuple[str, ...]], ignorecase: bool = False) -> Rule:
def fullmatch(msg: Union[str, tuple[str, ...]], ignorecase: bool = False) -> Rule:
"""完全匹配消息。
参数:
@ -361,7 +358,7 @@ class CommandRule:
def __init__(
self,
cmds: List[Tuple[str, ...]],
cmds: list[tuple[str, ...]],
force_whitespace: Optional[Union[str, bool]] = None,
):
self.cmds = tuple(cmds)
@ -380,7 +377,7 @@ class CommandRule:
async def __call__(
self,
cmd: Optional[Tuple[str, ...]] = Command(),
cmd: Optional[tuple[str, ...]] = Command(),
cmd_arg: Optional[Message] = CommandArg(),
cmd_whitespace: Optional[str] = CommandWhitespace(),
) -> bool:
@ -394,7 +391,7 @@ class CommandRule:
def command(
*cmds: Union[str, Tuple[str, ...]],
*cmds: Union[str, tuple[str, ...]],
force_whitespace: Optional[Union[str, bool]] = None,
) -> Rule:
"""匹配消息命令。
@ -424,7 +421,7 @@ def command(
config = get_driver().config
command_start = config.command_start
command_sep = config.command_sep
commands: List[Tuple[str, ...]] = []
commands: list[tuple[str, ...]] = []
for command in cmds:
if isinstance(command, str):
command = (command,)
@ -460,23 +457,23 @@ class ArgumentParser(ArgParser):
self,
args: Optional[Sequence[Union[str, MessageSegment]]] = None,
namespace: None = None,
) -> Tuple[Namespace, List[Union[str, MessageSegment]]]: ...
) -> tuple[Namespace, list[Union[str, MessageSegment]]]: ...
@overload
def parse_known_args(
self, args: Optional[Sequence[Union[str, MessageSegment]]], namespace: T
) -> Tuple[T, List[Union[str, MessageSegment]]]: ...
) -> tuple[T, list[Union[str, MessageSegment]]]: ...
@overload
def parse_known_args(
self, *, namespace: T
) -> Tuple[T, List[Union[str, MessageSegment]]]: ...
) -> tuple[T, list[Union[str, MessageSegment]]]: ...
def parse_known_args(
def parse_known_args( # pyright: ignore[reportIncompatibleMethodOverride]
self,
args: Optional[Sequence[Union[str, MessageSegment]]] = None,
namespace: Optional[T] = None,
) -> Tuple[Union[Namespace, T], List[Union[str, MessageSegment]]]: ...
) -> tuple[Union[Namespace, T], list[Union[str, MessageSegment]]]: ...
@overload
def parse_args(
@ -506,7 +503,7 @@ class ArgumentParser(ArgParser):
def _parse_optional(
self, arg_string: Union[str, MessageSegment]
) -> Optional[Tuple[Optional[Action], str, Optional[str]]]:
) -> Optional[tuple[Optional[Action], str, Optional[str]]]:
return (
super()._parse_optional(arg_string) if isinstance(arg_string, str) else None
)
@ -533,7 +530,7 @@ class ShellCommandRule:
__slots__ = ("cmds", "parser")
def __init__(self, cmds: List[Tuple[str, ...]], parser: Optional[ArgumentParser]):
def __init__(self, cmds: list[tuple[str, ...]], parser: Optional[ArgumentParser]):
self.cmds = tuple(cmds)
self.parser = parser
@ -553,7 +550,7 @@ class ShellCommandRule:
async def __call__(
self,
state: T_State,
cmd: Optional[Tuple[str, ...]] = Command(),
cmd: Optional[tuple[str, ...]] = Command(),
msg: Optional[Message] = CommandArg(),
) -> bool:
if cmd not in self.cmds or msg is None:
@ -571,7 +568,7 @@ class ShellCommandRule:
try:
args = self.parser.parse_args(state[SHELL_ARGV])
state[SHELL_ARGS] = args
except ArgumentError as e: # pragma: py-gte-39
except ArgumentError as e:
state[SHELL_ARGS] = ParserExit(status=2, message=str(e))
except ParserExit as e:
state[SHELL_ARGS] = e
@ -581,7 +578,7 @@ class ShellCommandRule:
def shell_command(
*cmds: Union[str, Tuple[str, ...]], parser: Optional[ArgumentParser] = None
*cmds: Union[str, tuple[str, ...]], parser: Optional[ArgumentParser] = None
) -> Rule:
"""匹配 `shell_like` 形式的消息命令。
@ -629,7 +626,7 @@ def shell_command(
config = get_driver().config
command_start = config.command_start
command_sep = config.command_sep
commands: List[Tuple[str, ...]] = []
commands: list[tuple[str, ...]] = []
for command in cmds:
if isinstance(command, str):
command = (command,)
@ -740,7 +737,7 @@ class IsTypeRule:
__slots__ = ("types",)
def __init__(self, *types: Type[Event]):
def __init__(self, *types: type[Event]):
self.types = types
def __repr__(self) -> str:
@ -756,7 +753,7 @@ class IsTypeRule:
return isinstance(event, self.types)
def is_type(*types: Type[Event]) -> Rule:
def is_type(*types: type[Event]) -> Rule:
"""匹配事件类型。
参数:

51
nonebot/typing.py
View File

@ -45,26 +45,38 @@ def overrides(InterfaceClass: object):
if sys.version_info < (3, 10):
def origin_is_union(origin: t.Optional[t.Type[t.Any]]) -> bool:
def type_has_args(type_: type[t.Any]) -> bool:
"""判断类型是否有参数"""
return isinstance(type_, (t._GenericAlias, types.GenericAlias)) # type: ignore
else:
def type_has_args(type_: type[t.Any]) -> bool:
return isinstance(type_, (t._GenericAlias, types.GenericAlias, types.UnionType)) # type: ignore
if sys.version_info < (3, 10):
def origin_is_union(origin: t.Optional[type[t.Any]]) -> bool:
"""判断是否是 Union 类型"""
return origin is t.Union
else:
def origin_is_union(origin: t.Optional[t.Type[t.Any]]) -> bool:
def origin_is_union(origin: t.Optional[type[t.Any]]) -> bool:
return origin is t.Union or origin is types.UnionType
def origin_is_literal(origin: t.Optional[t.Type[t.Any]]) -> bool:
def origin_is_literal(origin: t.Optional[type[t.Any]]) -> bool:
"""判断是否是 Literal 类型"""
return origin is t.Literal or origin is t_ext.Literal
def _literal_values(type_: t.Type[t.Any]) -> t.Tuple[t.Any, ...]:
def _literal_values(type_: type[t.Any]) -> tuple[t.Any, ...]:
return get_args(type_)
def all_literal_values(type_: t.Type[t.Any]) -> t.List[t.Any]:
def all_literal_values(type_: type[t.Any]) -> list[t.Any]:
"""获取 Literal 类型包含的所有值"""
if not origin_is_literal(get_origin(type_)):
return [type_]
@ -72,7 +84,7 @@ def all_literal_values(type_: t.Type[t.Any]) -> t.List[t.Any]:
return [x for value in _literal_values(type_) for x in all_literal_values(value)]
def origin_is_annotated(origin: t.Optional[t.Type[t.Any]]) -> bool:
def origin_is_annotated(origin: t.Optional[type[t.Any]]) -> bool:
"""判断是否是 Annotated 类型"""
with contextlib.suppress(TypeError):
return origin is not None and issubclass(origin, t_ext.Annotated)
@ -84,28 +96,19 @@ if sys.version_info >= (3, 10):
NONE_TYPES.add(types.NoneType)
def is_none_type(type_: t.Type[t.Any]) -> bool:
def is_none_type(type_: type[t.Any]) -> bool:
"""判断是否是 None 类型"""
return type_ in NONE_TYPES
if sys.version_info < (3, 9): # pragma: py-lt-39
def evaluate_forwardref(
ref: t.ForwardRef, globalns: t.Dict[str, t.Any], localns: t.Dict[str, t.Any]
) -> t.Any:
return ref._evaluate(globalns, localns)
else: # pragma: py-gte-39
def evaluate_forwardref(
ref: t.ForwardRef, globalns: t.Dict[str, t.Any], localns: t.Dict[str, t.Any]
) -> t.Any:
return ref._evaluate(globalns, localns, frozenset())
def evaluate_forwardref(
ref: t.ForwardRef, globalns: dict[str, t.Any], localns: dict[str, t.Any]
) -> t.Any:
return ref._evaluate(globalns, localns, frozenset())
# state
T_State: TypeAlias = t.Dict[t.Any, t.Any]
T_State: TypeAlias = dict[t.Any, t.Any]
"""事件处理状态 State 类型"""
_DependentCallable: TypeAlias = t.Union[
@ -134,11 +137,11 @@ T_BotDisconnectionHook: TypeAlias = _DependentCallable[t.Any]
# api hooks
T_CallingAPIHook: TypeAlias = t.Callable[
["Bot", str, t.Dict[str, t.Any]], t.Awaitable[t.Any]
["Bot", str, dict[str, t.Any]], t.Awaitable[t.Any]
]
"""`bot.call_api` 钩子函数"""
T_CalledAPIHook: TypeAlias = t.Callable[
["Bot", t.Optional[Exception], str, t.Dict[str, t.Any], t.Any], t.Awaitable[t.Any]
["Bot", t.Optional[Exception], str, dict[str, t.Any], t.Any], t.Awaitable[t.Any]
]
"""`bot.call_api` 后执行的函数，参数分别为 bot, exception, api, data, result"""
@ -244,5 +247,5 @@ T_PermissionUpdater: TypeAlias = _DependentCallable["Permission"]
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_DependencyCache: TypeAlias = t.Dict[_DependentCallable[t.Any], "Task[t.Any]"]
T_DependencyCache: TypeAlias = dict[_DependentCallable[t.Any], "Task[t.Any]"]
"""依赖缓存, 用于存储依赖函数的返回值"""

101
nonebot/utils.py
View File

@ -10,36 +10,23 @@ import json
import asyncio
import inspect
import importlib
import contextlib
import dataclasses
from pathlib import Path
from collections import deque
from contextvars import copy_context
from functools import wraps, partial
from contextlib import asynccontextmanager
from contextlib import AbstractContextManager, asynccontextmanager
from typing_extensions import ParamSpec, get_args, override, get_origin
from typing import (
Any,
Dict,
Type,
Tuple,
Union,
Generic,
Mapping,
TypeVar,
Callable,
Optional,
Sequence,
Coroutine,
AsyncGenerator,
ContextManager,
overload,
)
from collections.abc import Mapping, Sequence, Coroutine, AsyncGenerator
from typing import Any, Union, Generic, TypeVar, Callable, Optional, overload
from pydantic import BaseModel
from nonebot.log import logger
from nonebot.typing import (
is_none_type,
type_has_args,
origin_is_union,
origin_is_literal,
all_literal_values,
@ -64,8 +51,8 @@ def escape_tag(s: str) -> str:
def deep_update(
mapping: Dict[K, Any], *updating_mappings: Dict[K, Any]
) -> Dict[K, Any]:
mapping: dict[K, Any], *updating_mappings: dict[K, Any]
) -> dict[K, Any]:
"""深度更新合并字典"""
updated_mapping = mapping.copy()
for updating_mapping in updating_mappings:
@ -82,7 +69,7 @@ def deep_update(
def lenient_issubclass(
cls: Any, class_or_tuple: Union[Type[Any], Tuple[Type[Any], ...]]
cls: Any, class_or_tuple: Union[type[Any], tuple[type[Any], ...]]
) -> bool:
"""检查 cls 是否是 class_or_tuple 中的一个类型子类并忽略类型错误。"""
try:
@ -92,7 +79,7 @@ def lenient_issubclass(
def generic_check_issubclass(
cls: Any, class_or_tuple: Union[Type[Any], Tuple[Type[Any], ...]]
cls: Any, class_or_tuple: Union[type[Any], tuple[type[Any], ...]]
) -> bool:
"""检查 cls 是否是 class_or_tuple 中的一个类型子类。
@ -106,46 +93,46 @@ def generic_check_issubclass(
则会检查其 `__bound__` 或 `__constraints__`
是否是 class_or_tuple 中一个类型的子类或 None。
"""
try:
return issubclass(cls, class_or_tuple)
except TypeError:
origin = get_origin(cls)
if origin_is_union(origin):
if not type_has_args(cls):
with contextlib.suppress(TypeError):
return issubclass(cls, class_or_tuple)
origin = get_origin(cls)
if origin_is_union(origin):
return all(
is_none_type(type_) or generic_check_issubclass(type_, class_or_tuple)
for type_ in get_args(cls)
)
elif origin_is_literal(origin):
return all(
is_none_type(value) or isinstance(value, class_or_tuple)
for value in all_literal_values(cls)
)
# ensure generic List, Dict can be checked
elif origin:
# avoid class check error (typing.Final, typing.ClassVar, etc...)
try:
return issubclass(origin, class_or_tuple)
except TypeError:
return False
elif isinstance(cls, TypeVar):
if cls.__constraints__:
return all(
is_none_type(type_) or generic_check_issubclass(type_, class_or_tuple)
for type_ in get_args(cls)
for type_ in cls.__constraints__
)
elif origin_is_literal(origin):
return all(
is_none_type(value) or isinstance(value, class_or_tuple)
for value in all_literal_values(cls)
)
# ensure generic List, Dict can be checked
elif origin:
# avoid class check error (typing.Final, typing.ClassVar, etc...)
try:
return issubclass(origin, class_or_tuple)
except TypeError:
return False
elif isinstance(cls, TypeVar):
if cls.__constraints__:
return all(
is_none_type(type_)
or generic_check_issubclass(type_, class_or_tuple)
for type_ in cls.__constraints__
)
elif cls.__bound__:
return generic_check_issubclass(cls.__bound__, class_or_tuple)
return False
elif cls.__bound__:
return generic_check_issubclass(cls.__bound__, class_or_tuple)
return False
def type_is_complex(type_: Type[Any]) -> bool:
def type_is_complex(type_: type[Any]) -> bool:
"""检查 type_ 是否是复杂类型"""
origin = get_origin(type_)
return _type_is_complex_inner(type_) or _type_is_complex_inner(origin)
def _type_is_complex_inner(type_: Optional[Type[Any]]) -> bool:
def _type_is_complex_inner(type_: Optional[type[Any]]) -> bool:
if lenient_issubclass(type_, (str, bytes)):
return False
@ -200,7 +187,7 @@ def run_sync(call: Callable[P, R]) -> Callable[P, Coroutine[None, None, R]]:
@asynccontextmanager
async def run_sync_ctx_manager(
cm: ContextManager[T],
cm: AbstractContextManager[T],
) -> AsyncGenerator[T, None]:
"""一个用于包装 sync context manager 为 async context manager 的执行函数"""
try:
@ -216,7 +203,7 @@ async def run_sync_ctx_manager(
@overload
async def run_coro_with_catch(
coro: Coroutine[Any, Any, T],
exc: Tuple[Type[Exception], ...],
exc: tuple[type[Exception], ...],
return_on_err: None = None,
) -> Union[T, None]: ...
@ -224,14 +211,14 @@ async def run_coro_with_catch(
@overload
async def run_coro_with_catch(
coro: Coroutine[Any, Any, T],
exc: Tuple[Type[Exception], ...],
exc: tuple[type[Exception], ...],
return_on_err: R,
) -> Union[T, R]: ...
async def run_coro_with_catch(
coro: Coroutine[Any, Any, T],
exc: Tuple[Type[Exception], ...],
exc: tuple[type[Exception], ...],
return_on_err: Optional[R] = None,
) -> Optional[Union[T, R]]:
"""运行协程并当遇到指定异常时返回指定值。
@ -289,7 +276,7 @@ class classproperty(Generic[T]):
def __init__(self, func: Callable[[Any], T]) -> None:
self.func = func
def __get__(self, instance: Any, owner: Optional[Type[Any]] = None) -> T:
def __get__(self, instance: Any, owner: Optional[type[Any]] = None) -> T:
return self.func(type(instance) if owner is None else owner)