fastapi_depends_singleton.py

import asyncio
import logging
import time
from functools import lru_cache

import uvicorn
from aiocache import cached
from fastapi import FastAPI, APIRouter, Depends

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


class GetCachedSomething:
    def __init__(self, value: str):
        self.value = value

    @lru_cache
    def sync_load(self) -> str:
        logger.debug("loaded")
        response = self.value + "_sync"
        time.sleep(1)
        logger.debug("out loaded")
        return response

    @cached()
    async def async_load(self) -> str:
        logger.debug("async loaded")
        response = self.value + "_async"
        await asyncio.sleep(1)
        logger.debug("out async loaded")
        return response


app = FastAPI()

users_router = APIRouter()

get_users_something = GetCachedSomething("users.sql")


@users_router.get("/asyncload")
async def asyncload(users_queries=Depends(get_users_something.async_load)):
    return users_queries


@users_router.get("/syncload")
def syncload(users_queries=Depends(get_users_something.sync_load)):
    return users_queries


app.include_router(users_router)


if __name__ == "__main__":
    uvicorn.run("fastapi_depends_singleton:app", reload=True)