You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用aiocache的@cached装饰器导致子类方法签名继承失效

问题描述

子类重写父类的异步方法时,父类方法有明确返回类型,子类返回类型与父类一致,但添加aiocache的@cached装饰器后,Pylance提示类型错误:“get_all”重写“IBaseRepository”类中的同名方法,但类型“_Wrapped[..., Unknown, ..., CoroutineType[Any, Any, Unknown]]”不兼容。即使将缓存逻辑移到仓储外的服务层,类型检查器仍会认为函数签名被破坏(缓存采用Redis实现)。

代码示例:

from aiocache import cached

# 基类仓储
class IBaseRepository(Generic[T]):
    async def get_all(self) -> List[T]:
        ...

# 具体仓储实现
class CategoryRepositoryImpl(IBaseRepository[CategoryDomain]):
    @cached(ttl=600, key_builder=classaware_key_builder)
    async def get_all(self) -> List[CategoryDomain]:
        return await super().get_all()
解决方案

1. 强制类型转换修复签名

利用typing.cast告诉类型检查器,装饰后的方法与原方法签名完全一致,直接修正类型推断的偏差:

from typing import cast, List, Generic, TypeVar, Callable, Coroutine, Any
from aiocache import cached

T = TypeVar('T')
CategoryDomain = ...  # 替换为你的CategoryDomain定义

class IBaseRepository(Generic[T]):
    async def get_all(self) -> List[T]:
        ...

class CategoryRepositoryImpl(IBaseRepository[CategoryDomain]):
    @cached(ttl=600, key_builder=classaware_key_builder)
    async def get_all(self) -> List[CategoryDomain]:
        return await super().get_all()

# 强制转换方法类型,让类型检查器认可
CategoryRepositoryImpl.get_all = cast(
    Callable[[CategoryRepositoryImpl], Coroutine[Any, Any, List[CategoryDomain]]],
    CategoryRepositoryImpl.get_all
)

2. 封装类型友好的缓存装饰器

自己封装一层装饰器,确保正确保留原异步方法的类型信息,避免aiocache原生装饰器的类型推断问题:

from aiocache import cached as aiocache_cached
from functools import wraps
from typing import Callable, TypeVar, Coroutine, Any

T = TypeVar('T')
R = TypeVar('R')

def cached(ttl: int, key_builder: Callable) -> Callable[[Callable[..., Coroutine[Any, Any, R]]], Callable[..., Coroutine[Any, Any, R]]]:
    def decorator(func: Callable[..., Coroutine[Any, Any, R]]) -> Callable[..., Coroutine[Any, Any, R]]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> R:
            cached_func = aiocache_cached(ttl=ttl, key_builder=key_builder)(func)
            return await cached_func(*args, **kwargs)
        return wrapper
    return decorator

# 使用自定义装饰器
class CategoryRepositoryImpl(IBaseRepository[CategoryDomain]):
    @cached(ttl=600, key_builder=classaware_key_builder)
    async def get_all(self) -> List[CategoryDomain]:
        return await super().get_all()

3. 组合模式实现缓存(推荐)

遵循单一职责原则,将缓存逻辑与仓储逻辑分离,创建一个实现相同接口的缓存包装类,包裹原始仓储实例,完全规避继承中的类型冲突:

class CachedCategoryRepository(IBaseRepository[CategoryDomain]):
    def __init__(self, inner_repo: CategoryRepositoryImpl):
        self.inner_repo = inner_repo

    @cached(ttl=600, key_builder=classaware_key_builder)
    async def get_all(self) -> List[CategoryDomain]:
        return await self.inner_repo.get_all()

# 使用方式
original_repo = CategoryRepositoryImpl()
# 对外提供带缓存的仓储实例
cached_repo = CachedCategoryRepository(original_repo)

这种方式既保证了类型安全,又让仓储类只专注于数据访问,缓存类专注于缓存逻辑,代码结构更清晰。

4. 临时忽略类型错误(不推荐)

如果需要快速绕过类型检查,可以在装饰器行添加# type: ignore,但这会跳过该位置的所有类型检查,可能隐藏其他潜在问题:

class CategoryRepositoryImpl(IBaseRepository[CategoryDomain]):
    @cached(ttl=600, key_builder=classaware_key_builder)  # type: ignore
    async def get_all(self) -> List[CategoryDomain]:
        return await super().get_all()

内容的提问来源于stack exchange,提问作者Дима Ощепков

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.12 23:16:16