使用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,提问作者Дима Ощепков
相关产品推荐
相关产品推荐

