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

如何正确为AsyncGenerator、AsyncContextManager指定类型提示修复mypy报错

问题原因

该报错不属于mypy本身的问题,是类型签名不匹配导致的:

contextlib.asynccontextmanager装饰器的作用是将一个单次yield的异步生成器函数,包装为返回异步上下文管理器的函数。被装饰的函数本身的返回值是AsyncIterator[T],但装饰后的方法实际调用时返回的是AsyncContextManager[T],你在子类方法上写的是装饰前的函数返回类型,自然和父类要求的AsyncContextManager类型不匹配。

正确写法

直接将子类subscribe方法的返回类型调整为和父类完全一致即可,mypy会自动识别asynccontextmanager装饰器的类型转换逻辑:

import contextlib
import abc
import asyncio

from typing import AsyncContextManager, AsyncGenerator, AsyncIterator


class Base:
    @abc.abstractmethod
    async def subscribe(self) -> AsyncContextManager[AsyncGenerator[int, None]]:
        pass

class Impl1(Base):
    @contextlib.asynccontextmanager
    async def subscribe(self) -> AsyncContextManager[AsyncGenerator[int, None]]:
        async def _generator():
            for i in range(5):
                await asyncio.sleep(1)
                yield i
                    
        yield _generator()
可选优化

如果觉得类型太长可读性差,可以定义类型别名统一管理:

from typing import TypeAlias

SubscribeResult: TypeAlias = AsyncGenerator[int, None]
SubscribeReturn: TypeAlias = AsyncContextManager[SubscribeResult]

class Base:
    @abc.abstractmethod
    async def subscribe(self) -> SubscribeReturn:
        pass

class Impl1(Base):
    @contextlib.asynccontextmanager
    async def subscribe(self) -> SubscribeReturn:
        async def _generator():
            for i in range(5):
                await asyncio.sleep(1)
                yield i
                    
        yield _generator()

内容的提问来源于stack exchange,提问作者Andreas H.

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.10.07 02:06:00