如何为Python多层装饰器添加类型注解？

以下是修正后的完整代码，可解决所有mypy报错：

import functools
import inspect
from typing import Any, Callable, TypeVar, ParamSpec, Protocol
from typing_extensions import Awaitable


Type = TypeVar('Type')
Param = ParamSpec('Param')
_INSTANCES: dict[str, object] = {}


def make_injectable(instance_name: str, instance: object) -> None:
    _INSTANCES[instance_name] = instance


# 定义带有_injectable_args属性的可调用协议
class InjectableCallable(Protocol[Param, Type]):
    _injectable_args: tuple[str, ...]
    def __call__(self, *args: Param.args, **kwargs: Param.kwargs) -> Type | Awaitable[Type]: ...


def inject(*instances: str) -> Callable[[Callable[Param, Type | Awaitable[Type]]], InjectableCallable[Param, Type]]:
    def get_function_with_instances(fn: Callable[Param, Type | Awaitable[Type]]) -> InjectableCallable[Param, Type]:
        # 给函数添加注入标记属性，通过协议让类型检查器识别
        fn._injectable_args = instances  # type: ignore[attr-defined]

        async def async_handler(*args: Param.args, **kwargs: Param.kwargs) -> Type:
            new_kwargs: dict[str, Any] = dict(kwargs).copy()
            for instance in instances:
                if instance in new_kwargs:
                    continue
                if instance not in _INSTANCES:
                    raise ValueError(f"Instance {instance} was not initialized yet")
                new_kwargs[instance] = _INSTANCES[instance]
            result = fn(*args, **new_kwargs)
            return await result if inspect.isawaitable(result) else result

        def sync_handler(*args: Param.args, **kwargs: Param.kwargs) -> Type:
            new_kwargs: dict[str, Any] = dict(kwargs).copy()
            for instance in instances:
                if instance in new_kwargs:
                    continue
                if instance not in _INSTANCES:
                    raise ValueError(f"Instance {instance} was not initialized yet")
                new_kwargs[instance] = _INSTANCES[instance]
            result = fn(*args, **new_kwargs)
            if inspect.isawaitable(result):
                raise RuntimeError("Sync handler received async function")
            return result

        if inspect.iscoroutinefunction(fn):
            @functools.wraps(fn)
            async def wrapper(*args: Param.args, **kwargs: Param.kwargs) -> Type:
                return await async_handler(*args, **kwargs)
            wrapper._injectable_args = instances
            return wrapper  # type: ignore[return-value]
        else:
            @functools.wraps(fn)
            def wrapper(*args: Param.args, **kwargs: Param.kwargs) -> Type:
                return sync_handler(*args, **kwargs)
            wrapper._injectable_args = instances
            return wrapper  # type: ignore[return-value]

    return get_function_with_instances

错误修复细节 #

1. "Callable[Param, Type]" has no attribute "_injectable_args"

   • 问题：默认Callable类型不包含自定义的_injectable_args属性，类型检查器无法识别动态添加的属性。
   • 修复：定义InjectableCallable协议，明确声明该属性；同时添加type: ignore[attr-defined]忽略原函数的属性不存在警告。

2. 注入器工厂返回类型错误

   • 问题：原inject函数错误标注返回Callable[Param, Type]，但它实际是装饰器工厂，返回的是接受函数并返回函数的回调。
   • 修复：修正返回类型为Callable[[Callable[Param, Type | Awaitable[Type]]], InjectableCallable[Param, Type]]，明确它支持同步/异步函数输入，返回带注入属性的可调用对象。

3. Wrapper返回类型不匹配

   • 问题：原代码中同步/异步wrapper错误标注返回Callable[Param, Type]，但实际返回的是被装饰函数的结果Type（异步场景是Awaitable[Type]）。
   • 修复：拆分handler为同步、异步两个版本，分别处理对应类型的函数；将同步wrapper返回类型改为Type，异步wrapper返回类型改为Type（await后得到实际结果）。

4. 协程await类型不兼容

   • 问题：原代码中await handler(...)返回值为Type，但wrapper标注返回Callable，导致类型冲突。
   • 修复：异步handler专门处理协程函数，确保await后返回正确的Type类型，wrapper直接返回该结果。

内容的提问来源于stack exchange，提问作者José Horta