FASTAPI / 4. DEPENDENCY INJECTION

Dependency Injection

The most powerful FastAPI feature — clean, reusable, testable logic

EXPLANATION

Dependency Injection (DI) is FastAPI's most elegant feature. Instead of repeating logic (auth checks, DB sessions, pagination params) in every route, you define it once as a dependency and inject it wherever needed.

How it works:
• Define a function (sync or async) that returns something useful
• Declare it as a parameter type-hinted with Depends(your_function)
• FastAPI calls it automatically before your route, passes the result in

Dependencies can themselves have dependencies — you build a dependency tree. FastAPI resolves the whole tree before calling your route. This is exactly like constructor injection in Java/C# but Pythonic.

Common use cases:
① Pagination parameters — extract skip/limit once, reuse everywhere
② Database sessions — open a session, yield it, always close it (generator pattern)
③ Authentication — verify JWT token, return current user
④ Permission checks — "current user must be admin"
⑤ Rate limiting, logging, feature flags

Yield dependencies (yield instead of return) are the pattern for resources that need cleanup. Code before yield = setup, code after yield = teardown. FastAPI handles this like a context manager — teardown always runs, even on exceptions.

Dependencies can be added at route level, router level, or app level. App-level dependencies run on every single request.

Sub-dependencies compose naturally:
get_db → get_current_user → require_admin
Each layer builds on the last. FastAPI caches each dependency's result within a single request so get_db is only called once even if multiple route params need it.

ARCHITECTURE

@app.get("/admin/dashboard")
  def dashboard(admin: User = Depends(require_admin)):
                                         │
            ┌──────────────────────────────┘
            ↓  require_admin calls:
       get_current_user(token)
                │
                ↓ get_current_user calls:
           get_db()  +  oauth2_scheme()
                │
                ↓
           DB Session  +  JWT Token

  FastAPI resolves BOTTOM UP:
  1. get_db()        → db session
  2. oauth2_scheme() → JWT string from header
  3. get_current_user(db, token) → User object
  4. require_admin(user) → User if admin, else 403
  5. dashboard(admin) → response

  All cached per-request. get_db() called only once.

CODE

PYTHON
1from fastapi import FastAPI, Depends, HTTPException, status
2from typing import Annotated
3
4app = FastAPI()
5
6# ── 1. Simple reusable dependency ─────────────────────
7class PaginationParams:
8    def __init__(self, skip: int = 0, limit: int = Query(default=10, le=100)):
9        self.skip = skip
10        self.limit = limit
11
12Pagination = Annotated[PaginationParams, Depends(PaginationParams)]
13
14@app.get("/items/")
15def get_items(pagination: Pagination):
16    return {"skip": pagination.skip, "limit": pagination.limit}
17
18@app.get("/users/")
19def get_users(pagination: Pagination):   # same dependency, zero duplication
20    return {"skip": pagination.skip, "limit": pagination.limit}
21
22# ── 2. Database session (yield pattern) ───────────────
23from sqlmodel import Session, create_engine
24
25engine = create_engine("sqlite:///./app.db")
26
27def get_db():
28    with Session(engine) as session:
29        yield session          # route runs here
30        # session auto-closed after route finishes
31
32DB = Annotated[Session, Depends(get_db)]
33
34@app.get("/db-items/")
35def get_db_items(db: DB):
36    # db is a live SQLModel Session
37    return {"message": "db session injected!"}
38
39# ── 3. Auth dependency chain ──────────────────────────
40from fastapi.security import OAuth2PasswordBearer
41
42oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
43Token = Annotated[str, Depends(oauth2_scheme)]
44
45def get_current_user(token: Token, db: DB):
46    # decode token, fetch user from db
47    # raise 401 if invalid
48    return {"username": "kamran", "role": "admin"}
49
50CurrentUser = Annotated[dict, Depends(get_current_user)]
51
52def require_admin(user: CurrentUser):
53    if user["role"] != "admin":
54        raise HTTPException(status_code=403, detail="Admins only")
55    return user
56
57Admin = Annotated[dict, Depends(require_admin)]
58
59@app.get("/admin/stats")
60def admin_stats(admin: Admin):
61    return {"message": f"Welcome admin {admin['username']}"}
← PREV3. Response Models & Status CodesNEXT →5. JWT Authentication