2

我正在一个 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_asyncio包，但它没有打补丁uvloop，而且，即使将它与 asyncio 一起使用，第一次使用后服务似乎也会冻结。

你知道我怎么能完成这个吗？

4

1 回答 1

4

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

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在端点代码中或实际上在任何地方为它们提供结果，只要它和 async 函数即可！

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

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

于 2021-03-03T22:14:06.487 回答