Dependency Injection in FastAPI

Depends — FastAPI’s dependency injection system for reusing code across endpoints.

Basic Example

from fastapi import FastAPI, Depends

app = FastAPI()

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

@app.get("/tasks/")
def list_tasks(db: Session = Depends(get_db)):
    return db.query(Task).all()

Request Parameters as a Dependency

class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@app.get("/tasks/")
def list_tasks(pagination: Pagination = Depends()):
    return {
        "page": pagination.page,
        "offset": pagination.offset,
    }

Authentication via Depends

from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordBearer

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

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    user = verify_token(token, db)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

@app.get("/tasks/")
def list_tasks(current_user: User = Depends(get_current_user)):
    return db.query(Task).filter(Task.owner_id == current_user.id).all()

Dependency Chaining

def get_db():
    ...  # dependency 1

def get_current_user(db = Depends(get_db)):
    ...  # dependency 2, uses dependency 1

def get_admin_user(user = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(403)
    return user  # dependency 3, uses dependency 2

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, admin: User = Depends(get_admin_user)):
    ...

yield Dependencies (with Cleanup)

def get_db():
    db = SessionLocal()
    try:
        yield db        # code before yield runs before the handler
    finally:
        db.close()      # code after yield runs after the handler (cleanup)

Router-Level Depends

router = APIRouter(
    prefix="/tasks",
    dependencies=[Depends(get_current_user)],  # applied to all endpoints in this router
)