在异步环境中缓存结果

Question

我正在 FastAPI 端点中工作，该端点进行 I/O 绑定操作，该操作是异步的以提高效率。但是，这需要时间，所以我想缓存结果以在一段时间内重复使用。

我目前有这个：

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def _get_expensive_resource(key) -> None:
    await asyncio.sleep(2)
    return True

@app.get('/')
async def get(key):
    return await _get_expensive_resource(key)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("test:app")

我正在尝试使用该cachetools包来缓存结果，并且我尝试了如下操作：

import asyncio
from cachetools import TTLCache
from fastapi import FastAPI
  
app = FastAPI()

async def _get_expensive_resource(key) -> None:
    await asyncio.sleep(2)
    return True

class ResourceCache(TTLCache):
    def __missing__(self, key):
        loop = asyncio.get_event_loop()
        resource = loop.run_until_complete(_get_expensive_resource(key))
        self[key] = resource
        return resource

resource_cache = ResourceCache(124, 300)

@app.get('/')
async def get(key: str):
    return resource_cache[key]

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("test2:app")

但是，这失败了，因为据我了解，该__missing__方法是同步的，并且您无法从异步的同步中调用异步。错误是：

RuntimeError: this event loop is already running.

如果我使用普通的 asyncio 而不是 uvloop，也会发生类似的错误。

对于 asyncio 事件循环，我尝试使用nest_asynciopackage，但它没有修补uvloop，而且即使与 asyncio 一起使用，服务似乎在第一次使用后就会冻结。

你知道我怎样才能做到这一点吗？

Answer 1

自动回答遇到此问题的其他人（包括十五天内的我自己）：

TTLCache就像普通的 python 字典一样工作，访问丢失的键将调用该__missing__方法。因此，我们希望使用字典中的值（如果存在），如果不存在，我们可以在此方法中收集资源。此方法还应该在缓存中设置键（以便下次它会出现）并返回这次使用的值。

class ResourceCache(TTLCache):
    def __missing__(self, key) -> asyncio.Task:
        # Create a task 
        resource_future = asyncio.create_task(_get_expensive_resource(key))
        self[key] = resource_future
        return resource_future

因此，我们有一个将键映射到asyncio.Task 的缓存（本质上是一个字典） 。这些任务将在事件循环中异步执行（已由 FastAPI 启动！）。当我们需要结果时，我们可以await在端点代码中或实际上任何地方获取结果，只要它和异步函数！

@app.get("/")
async def get(key:str) -> bool:
    return await resource_cache[key]

第二次调用此端点（在缓存超时内）将使用缓存的资源（在我们的示例中用“true”模拟）。