Feature: 为子依赖添加 PEP593 Annotated 支持 (#1832)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Johnny Hsieh
2023-03-22 20:54:46 +08:00
committed by GitHub
parent 9afaf3d516
commit 3e92fccd4e
3 changed files with 59 additions and 13 deletions

View File

@ -1,8 +1,10 @@
import asyncio
import inspect
from typing_extensions import Annotated
from contextlib import AsyncExitStack, contextmanager, asynccontextmanager
from typing import TYPE_CHECKING, Any, Type, Tuple, Literal, Callable, Optional, cast
from pydantic.typing import get_args, get_origin
from pydantic.fields import Required, Undefined, ModelField
from nonebot.dependencies.utils import check_field_type
@ -78,21 +80,33 @@ class DependParam(Param):
def _check_param(
cls, param: inspect.Parameter, allow_types: Tuple[Type[Param], ...]
) -> Optional["DependParam"]:
if isinstance(param.default, DependsInner):
dependency: T_Handler
if param.default.dependency is None:
assert param.annotation is not param.empty, "Dependency cannot be empty"
dependency = param.annotation
else:
dependency = param.default.dependency
sub_dependent = Dependent[Any].parse(
call=dependency,
allow_types=allow_types,
)
return cls(
Required, use_cache=param.default.use_cache, dependent=sub_dependent
type_annotation, depends_inner = param.annotation, None
if get_origin(param.annotation) is Annotated:
type_annotation, *extra_args = get_args(param.annotation)
depends_inner = next(
(x for x in extra_args if isinstance(x, DependsInner)), None
)
depends_inner = (
param.default if isinstance(param.default, DependsInner) else depends_inner
)
if depends_inner is None:
return
dependency: T_Handler
if depends_inner.dependency is None:
assert (
type_annotation is not inspect.Signature.empty
), "Dependency cannot be empty"
dependency = type_annotation
else:
dependency = depends_inner.dependency
sub_dependent = Dependent[Any].parse(
call=dependency,
allow_types=allow_types,
)
return cls(Required, use_cache=depends_inner.use_cache, dependent=sub_dependent)
@classmethod
def _check_parameterless(
cls, value: Any, allow_types: Tuple[Type[Param], ...]