pytest参数化时asyncio协程复用触发RuntimeError的解决方案问询

问题核心 #

你遇到的RuntimeError: cannot reuse already awaited coroutine，根源是@lru_cache直接装饰异步函数时，缓存的是协程对象实例——而协程对象只能被await一次。当参数化的不同输入（整数28815543和字符串"28815543"）生成同一个IdReference实例时，会命中同一个缓存的协程对象，第二次await就触发了错误。

解决办法（无需拆分参数化） #

1. 缓存异步函数的执行结果而非协程对象 #

替换原生@lru_cache为自定义装饰器，缓存函数的返回值而非协程实例：

from functools import wraps, lru_cache

def async_lru_cache(maxsize=128):
    def decorator(func):
        # 用lru_cache缓存异步函数的最终结果
        @lru_cache(maxsize=maxsize)
        async def cached_execution(*args, **kwargs):
            return await func(*args, **kwargs)
        
        @wraps(func)
        async def wrapper(*args, **kwargs):
            return await cached_execution(*args, **kwargs)
        
        # 暴露缓存清理方法，方便测试时重置
        wrapper.cache_clear = cached_execution.cache_clear
        return wrapper
    return decorator

用这个装饰器替代@lru_cache修饰fetch.summary：

@async_lru_cache()
async def summary(id_ref):
    # 原函数逻辑

这样每次调用都会生成新的协程对象，缓存的是最终计算结果，彻底避免重复await同一个协程的问题。

2. 测试用例执行后自动清理缓存 #

如果不想修改业务代码，可以通过pytest的自动fixture在每个测试用例结束后清空缓存：

import pytest

@pytest.fixture(autouse=True)
def clear_fetch_cache():
    yield
    # 清空fetch.summary的缓存
    fetch.summary.cache_clear()

这个fixture会自动在每个测试用例执行完毕后重置缓存，确保不同参数化用例的缓存互相隔离。

3. 调整缓存key的生成逻辑 #

如果允许修改IdReference的实现，可以调整其__hash__和__eq__方法，让原始输入的类型（整数/字符串）也参与缓存key的计算，确保不同输入生成不同的缓存条目：

class IdReference:
    def __init__(self, raw_id):
        self.external_id = str(raw_id)
        self.raw_type = type(raw_id)  # 保留原始输入类型
    
    def __hash__(self):
        return hash((self.external_id, self.raw_type))
    
    def __eq__(self, other):
        return (self.external_id == other.external_id) and (self.raw_type == other.raw_type)

这样整数和字符串的raw_id会生成不同的IdReference实例，避免命中同一个缓存。

总结 #

不需要强制拆分参数化，上述三种方法都可以解决问题：

• 优先推荐缓存异步函数结果的方案，从根源避免协程重复await的问题；
• 测试层面的缓存清理是快速临时解决方案；
• 调整IdReference的哈希逻辑适合允许修改业务模型的场景。

内容的提问来源于stack exchange，提问作者Theolodus