FastAPI Cheatsheet 04 — Dependency Injection

from fastapi import Depends

def get_settings():
    return Settings()

@app.get("/info")
async def info(s: Settings = Depends(get_settings)):
    return {"db": s.database_url}

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session            # cleanup runs after handler

@app.get("/users")
async def list_(db: AsyncSession = Depends(get_db)):
    return await db.execute(select(User))

async def get_settings(): ...

async def get_db(s: Settings = Depends(get_settings)) -> AsyncSession:
    ...

# Both deps below resolve get_settings ONCE (cached per request)
async def get_user(s: Settings = Depends(get_settings)): ...

class Pagination:
    def __init__(self, page: int = 1, limit: int = Query(20, le=100)):
        self.page, self.limit = page, limit

@app.get("/posts")
async def posts(p: Pagination = Depends()):    # () or =Depends(Pagination)
    ...

from fastapi.security import OAuth2PasswordBearer, HTTPBearer, APIKeyHeader

oauth2 = OAuth2PasswordBearer(tokenUrl="token")
bearer = HTTPBearer()
api_key = APIKeyHeader(name="X-API-Key")

async def current_user(token: str = Depends(oauth2)) -> User:
    return verify(token)

@app.get("/me")
async def me(u: User = Depends(current_user)): ...

@app.get("/items", dependencies=[Depends(verify_api_key)])     # path-level
async def items(): return [...]

router = APIRouter(dependencies=[Depends(verify_api_key)])     # router
app.include_router(router, dependencies=[Depends(rate_limit)]) # include
app = FastAPI(dependencies=[Depends(global_logger)])           # app

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.engine = create_async_engine(URL)
    app.state.sm = async_sessionmaker(app.state.engine, expire_on_commit=False)
    app.state.http = httpx.AsyncClient(timeout=10)
    yield
    await app.state.http.aclose()
    await app.state.engine.dispose()

app = FastAPI(lifespan=lifespan)

# Access via Request.app.state
async def get_db(request: Request) -> AsyncSession:
    async with request.app.state.sm() as session:
        yield session

from functools import lru_cache

@lru_cache
def get_settings() -> Settings:
    return Settings()

class UserService:
    def __init__(self, db: AsyncSession, mailer: Mailer):
        self.db, self.mailer = db, mailer
    async def create(self, data): ...

async def get_user_service(
    db: AsyncSession = Depends(get_db),
    mailer: Mailer = Depends(get_mailer),
) -> UserService:
    return UserService(db, mailer)

@app.post("/users")
async def create_(d: UserCreate, svc: UserService = Depends(get_user_service)):
    return await svc.create(d)

def override_db(): return TestDB()

app.dependency_overrides[get_db] = override_db
# ...run tests...
app.dependency_overrides = {}        # reset

async def get_tenant(req: Request) -> Tenant:
    tid = req.headers.get("x-tenant-id")
    return await load_tenant(tid)

async def get_tenant_db(t: Tenant = Depends(get_tenant)) -> AsyncSession:
    async with sm() as s:
        await s.execute(text("SET LOCAL app.tenant_id = :t"), {"t": t.id})
        yield s

from fastapi import BackgroundTasks

@app.post("/signup")
async def signup(u: UserIn, bg: BackgroundTasks, db = Depends(get_db)):
    user = await create_user(db, u)
    bg.add_task(send_welcome, user.email)   # fire-and-forget
    return {"id": user.id}