Dependencies

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db        # yield = FastAPI calls cleanup after response
    finally:
        db.close()

@app.get("/items")
def list_items(db = Depends(get_db)):
    return db.query(Item).all()

from fastapi import Depends, Query

def common_params(
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=10, le=100),
):
    return {"skip": skip, "limit": limit}

@app.get("/items")
def list_items(params: dict = Depends(common_params)):
    return db[params["skip"] : params["skip"] + params["limit"]]

class PaginationParams:
    def __init__(
        self,
        skip: int = Query(default=0, ge=0),
        limit: int = Query(default=10, le=100),
    ):
        self.skip = skip
        self.limit = limit

@app.get("/items")
def list_items(p: PaginationParams = Depends(PaginationParams)):
    return db[p.skip : p.skip + p.limit]

# Shorthand — FastAPI detects the class:
@app.get("/items")
def list_items(p: PaginationParams = Depends()):
    ...

def get_token(x_token: str = Header(...)):
    return x_token

def get_current_user(token: str = Depends(get_token)):
    user = db.get_user_by_token(token)
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

@app.get("/me")
def me(user = Depends(get_current_user)):
    return user

def get_db():
    db = SessionLocal()
    try:
        yield db      # code below yield runs after response
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# Async version:
async def get_db():
    async with AsyncSession(engine) as session:
        yield session

from fastapi import Depends, HTTPException, Header

def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != settings.api_key:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return x_api_key

@app.get("/secure", dependencies=[Depends(verify_api_key)])
def secure_endpoint():
    return {"ok": True}

from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin)],  # applied to ALL routes
)

@router.get("/users")
def list_users(): ...    # require_admin runs automatically

app = FastAPI(dependencies=[Depends(verify_rate_limit)])
# runs on every single request

# By default, the same dependency called twice in one request returns
# the cached result. Disable with:
def get_timestamp():
    return datetime.now()

@app.get("/test")
def test(
    t1 = Depends(get_timestamp),
    t2 = Depends(get_timestamp, use_cache=False),  # fresh call
):
    return {"t1": t1, "t2": t2}

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    # token extracted from Authorization: Bearer <token>
    return decode_jwt(token)

@app.get("/me")
def me(user = Depends(get_current_user)):
    return user

from functools import lru_cache

@lru_cache          # singleton — created once per process
def get_settings():
    return Settings()

@app.get("/info")
def info(settings = Depends(get_settings)):
    return {"debug": settings.debug}

# When you only need the side effect (auth check, rate limit),
# not the return value:
@app.get("/items", dependencies=[Depends(verify_token), Depends(check_rate)])
def list_items():
    return []