Skip to content

Instantly share code, notes, and snippets.

@euri10
Created March 10, 2020 12:21
Show Gist options
  • Save euri10/f88e0b677caefac2c74cd058acc68327 to your computer and use it in GitHub Desktop.
Save euri10/f88e0b677caefac2c74cd058acc68327 to your computer and use it in GitHub Desktop.
async cache bug
import asyncio
import logging
import time
from functools import lru_cache
import uvicorn
from async_lru import alru_cache
from fastapi import FastAPI, APIRouter, Depends
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class AsyncGetQueries:
def __init__(self, sqlfile: str):
self.sqlfile = sqlfile
@lru_cache
def loaded(self) -> str:
logger.debug("loaded")
time.sleep(2)
logger.debug("out loaded")
response = self.sqlfile + "_sync"
return response
@alru_cache
async def async_load(self) -> str:
logger.debug("async loaded")
await asyncio.sleep(2)
logger.debug("out async loaded")
response = self.sqlfile + "_async"
return response
app = FastAPI()
users_router = APIRouter()
get_users_queries = AsyncGetQueries("users.sql")
@users_router.get("/asyncload")
async def asyncload(users_queries=Depends(get_users_queries.async_load)):
return users_queries
@users_router.get("/syncload")
def syncload(users_queries=Depends(get_users_queries.loaded)):
return users_queries
app.include_router(users_router)
if __name__ == "__main__":
uvicorn.run("toto:app", reload=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment