Python有选择地等待协同路由，而不引发RuntimeWarning

Question

我目前有一个类似缓存的对象，它有一个aget方法，该方法接收要返回的对象的名称，如果键不存在，则接收回调协程

import asyncio

class Cache:
    def __init__(self):
        self.cache = {}

    async def aget(self, name, default):
        try:
            return self.cache["name"]
        except KeyError:
            return await default

async def foo():
    pass

async def bar(cache):
    obj = await cache.aget("duh", foo())
    # If "duh" doesn't exist, then await foo() and return it

cache = Cache()

asyncio.run(bar(cache))

因此，假设密钥确实存在，那么foo()将永远不会被等待，从而引发一个RuntimeWarning抱怨没有调度协同路由

问题：是否有一种干净的方法可以选择等待协同路由，而无需解压协同路由及其参数（如^{）。我不认为将运行时警告作为一个选项存在是因为它的USECASE。

Answer 1

Question: Is there a clean way to optionally await a coroutine, without the need to unpack the coroutine and its arguments like aget("duh", foo, arg).

您可以创建任务，但在它有机会开始运行之前取消它：

    async def aget(self, name, default):
        try:
            obj = self.cache["name"]
        except KeyError:
            return await default
        else:
            asyncio.create_task(default).cancel()
            return obj

如果你走这条路，一定要测量创建和取消任务的开销。因为它发生在代码的热路径上（大概），所以可能会影响性能

Answer 2

我找到了另一个答案，创建任务并取消它们可能会导致取消问题和不必要的开销

    async def aget(self, name, default):
        try:
            obj = self.cache["name"]
        except KeyError:
            return await default
        else:
            default.close()
            return obj

由于某些原因，close()方法没有在任何地方记录，这是gc在协同路由对象上调用的方法

编辑：正如评论所说，它记录在https://docs.python.org/3/reference/datamodel.html#coroutine-objects