如何在FastAPI正确的使用依赖注入

from fastapi import FastAPI, Depends

app = FastAPI()


###########
#  第一块  #
###########
class FakerDB(object):
    def __init__(self) -> None:
        self._initialized = False
    async def create(self) -> None:
        self._initialized = True

    async def get_user_info(self) -> dict:
        if not self._initialized:
            raise RuntimeError("FakerDB not initialized")
        return {
            "id": 1,
            "name": "so1n",
            "age": 18,
        }


###########
#  第二块  #
###########
_db = None

def get_user_db() -> FakerDB:
    global _db
    if _db is None:
        _db = FakerDB()
    return _db


###########
#  第三块  #
###########
@app.on_event("startup")
async def startup_event(user_db: FakerDB = Depends(get_user_db)):
    await user_db.create()


@app.get("/api/user/info")
async def login(user_db: FakerDB = Depends(get_user_db)):
    return await user_db.get_user_info()


import uvicorn

uvicorn.run(app)

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

import time
import uvicorn
from fastapi import FastAPI, Depends, Query
from asyncio import Queue, sleep

app = FastAPI()


###########
#  第一块  #
###########
class FakerDB(object):
    async def get_user_info(self) -> dict:
        return {
            "id": 1,
            "name": "so1n",
            "age": 18,
        }


###########
#  第二块  #
###########
_db_pool = Queue()


async def get_user_db() -> FakerDB:
    db = await _db_pool.get()
    try:
        yield db
    finally:
        _db_pool.put_nowait(db)



###########
#  第三块  #
###########
@app.on_event("startup")
async def start_up() -> None:
    # 由于uvicorn的特性，需要在这里重新创建Queue，不然它们将不会被同一个event loop调用
    global _db_pool
    _db_pool = Queue()

    for _ in range(3):
        _db_pool.put_nowait(FakerDB())


@app.get("/api/user/info")
async def login(uid: int = Query(...), user_db: FakerDB = Depends(get_user_db)):
    print(f"Hello! {uid}, access time {time.time()}")
    await sleep(3)  # 模拟其它io
    return await user_db.get_user_info()


uvicorn.run(app)

import httpx
import asyncio


async def main():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"http://127.0.0.1:8000/api/user/info?uid={i}") for i in range(5)]
        await asyncio.gather(*tasks)


asyncio.run(main())

import json
import time
from redis.asyncio import Redis
from httpx import AsyncClient
from typing import Optional

_namespace = "example:info"

class InfoService(object):
    def __init__(self, *, redis: Redis, client: AsyncClient) -> None:
        self.redis = redis
        self.client = client

    async def get(self, url: str, cache_timeout: int = 9) -> dict:
        redis_result: Optional[str] = await self.redis.get(f"{_namespace}:{url}")
        if not redis_result:
            async with self.client as client:
                response = await client.get(url)
                status_code = response.status_code
                timestamp = int(time.time())
            result = {"timestamp": timestamp, "status_code": status_code}
            await self.redis.set(f"{_namespace}:{url}", json.dumps(result), ex=cache_timeout)
        else:
            result = json.loads(redis_result)
        return result

from dependency_injector import containers, providers
from redis.asyncio import Redis
from httpx import AsyncClient
from services import InfoService


class Container(containers.DeclarativeContainer):
    config = providers.Configuration(yaml_files=["config.yaml"])

    redis = providers.Singleton(
        Redis,
        host=config.redis.host,
        port=config.redis.port,
    )
    client = providers.Singleton(
        AsyncClient,
        timeout=config.http.timeout,
    )
    info_service = providers.Factory(
        InfoService,
        redis=redis,
        client=client,
    )

from fastapi import APIRouter, Depends, Body
from dependency_injector.wiring import inject, Provide
from services import InfoService
from containers import Container


router = APIRouter()

@router.post("/")
@inject
async def demo(
    url: str = Body(..., embed=True),
    cache_timeout: int = Depends(Provide[Container.config.default.cache_timeout]),
    info_service: InfoService = Depends(Provide[Container.info_service]),
):
    return await info_service.get(url, cache_timeout)

from fastapi import FastAPI
from dependency_injector.wiring import inject, Provide

from containers import Container
import endpoints


def create_app() -> FastAPI:
    container = Container()

    app = FastAPI()
    app.container = container
    app.include_router(endpoints.router)

    @app.on_event("startup")
    async def start_up_1() -> None:
        result = await app.container.redis().info()
        if result:
            print(f"Redis: {id(app.container.redis())} connect success 1")

    @app.on_event("startup")
    @inject
    async def start_up_2(_container: Container = Provide[Container]) -> None:
        result = await _container.redis().info()
        if result:
            print(f"Redis: {id(_container.redis())} connect success 2")

    container.wire(modules=[__name__, endpoints])
    return app


app = create_app()

from dependency_injector.wiring import Provide
from typing import TypeVar, Type
_T = TypeVar("_T")

def MyProvide(a: Type[_T]) -> _T:
    return Provide[a]

@app.on_event("startup")
@inject
async def start_up_2(_container = MyProvide(Container)) -> None:
    result = await _container.redis().info()
    if result:
        print(f"Redis: {id(_container.redis())} connect success 2")