By day 93 of the 100-day Python Backend Development series, you've likely built routes, connected databases, and handled basic auth. But scaling that auth logic across dozens of endpoints without code duplication is where FastAPI dependency injection for auth and roles explained becomes indispensable. Let's walk through how to build reusable guards that plug into your endpoints and routers with minimal friction.
Why dependency injection for auth matters
Most teams start with auth checks scattered across route handlers: decode the JWT, fetch the user, check permissions, raise an exception if something fails. Within a week, you're copy-pasting the same logic into fifteen endpoints. FastAPI's Depends system lets you encapsulate this as reusable dependencies that compose cleanly and test easily.
The pattern we'll cover here is battle-tested in production systems handling millions of requests. It scales from small internal tools to multi-tenant SaaS platforms, and it keeps your business logic separate from your auth plumbing.
Current user — extracting identity from the token
The foundation is a dependency that decodes the JWT, validates it, and returns the authenticated user. This runs once per request and fails fast if the token is missing or invalid.
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthCredentials
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
import os
security = HTTPBearer()
SECRET_KEY = os.getenv("SECRET_KEY", "dev-key")
ALGORITHM = "HS256"
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
roles = Column(JSON, default=[]) # ["user", "admin", etc]
def get_db() -> Session:
"""Yield a database session for each request."""
db = SessionLocal()
try:
yield db
finally:
db.close()
async def current_user(
credentials: HTTPAuthCredentials = Depends(security),
db: Session = Depends(get_db)
) -> User:
"""
Decode JWT, validate it, and load the user from the database.
Raises 401 if token is invalid or user not found.
"""
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token payload"
)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found"
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User account is inactive"
)
return user
This dependency is the entry point to every protected route. It handles JWT decoding, database lookups, and early rejection. Notice we inject get_db as a dependency too—FastAPI chains them automatically.
Role checker — factory pattern for flexible access control
With the current user in hand, we need to check roles. Rather than write separate functions for admin, moderator, and user, we use a factory that returns a dependency tailored to each role. This keeps code DRY and makes it trivial to add new roles.
def require(role: str):
"""
Factory that returns a dependency checking for a specific role.
Usage: Depends(require('admin'))
"""
async def check_role(user: User = Depends(current_user)) -> User:
if role not in user.roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"User does not have '{role}' role"
)
return user
return check_role
def require_any(roles: list[str]):
"""
Factory that returns a dependency checking if user has ANY of the given roles.
"""
async def check_any_role(user: User = Depends(current_user)) -> User:
if not any(r in user.roles for r in roles):
detail = f"User must have one of: {', '.join(roles)}"
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=detail
)
return user
return check_any_role
def require_all(roles: list[str]):
"""
Factory that returns a dependency checking if user has ALL given roles.
"""
async def check_all_roles(user: User = Depends(current_user)) -> User:
if not all(r in user.roles for r in roles):
detail = f"User must have all of: {', '.join(roles)}"
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=detail
)
return user
return check_all_roles
The factory pattern shines here. Each call to require("admin") returns a fresh dependency function. FastAPI treats it as a new guard, and you can nest and compose them however your business logic demands.
Using guards on individual routes
Once you have your guards, plugging them into routes is straightforward. Declare them in the route's dependencies or as function parameters.
from fastapi import APIRouter, Depends
router = APIRouter()
@router.get("/me")
async def get_current_user_info(user: User = Depends(current_user)):
"""Any authenticated user can see their own profile."""
return {
"id": user.id,
"email": user.email,
"roles": user.roles
}
@router.post("/users")
async def create_user(
payload: CreateUserRequest,
user: User = Depends(require("admin"))
):
"""Only admins can create users."""
# user is guaranteed to have "admin" role here
new_user = User(email=payload.email)
db.add(new_user)
db.commit()
return {"id": new_user.id, "email": new_user.email}
@router.delete("/users/{user_id}")
async def delete_user(
user_id: int,
admin: User = Depends(require("admin")),
db: Session = Depends(get_db)
):
"""Delete a user. Only admins, and they're injected."""
target = db.query(User).filter(User.id == user_id).first()
if not target:
raise HTTPException(status_code=404, detail="User not found")
db.delete(target)
db.commit()
return {"deleted": user_id}
When you declare user: User = Depends(require("admin")), FastAPI runs require("admin") at startup, then calls the returned dependency for each request. If the role check fails, the HTTP 403 exception is raised before your route code runs.
Protecting entire routers with dependencies
For larger feature sets, you can protect all routes in a router at once by passing dependencies at the router level.
admin_router = APIRouter(
prefix="/admin",
tags=["admin"],
dependencies=[Depends(require("admin"))]
)
@admin_router.get("/stats")
async def get_admin_stats(db: Session = Depends(get_db)):
"""All routes in admin_router require 'admin' role."""
# No need to declare require("admin") here; it's enforced by the router
return {"active_users": db.query(User).filter(User.is_active).count()}
@admin_router.post("/broadcast-message")
async def send_broadcast(
payload: BroadcastRequest,
db: Session = Depends(get_db)
):
"""Also protected by the router-level admin dependency."""
# Broadcast logic here
return {"message_sent": True}
# Include the router in your main app
app.include_router(admin_router)
Router-level dependencies reduce boilerplate when you have groups of related endpoints that all require the same guard. They're also easier to refactor later—change the guard in one place and it applies everywhere.
Testing with dependency overrides
The real power of this pattern emerges in tests. FastAPI lets you override dependencies on a per-test basis, so you can inject fake users with different roles without mocking JWT logic.
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
# Fixture for a regular user
@pytest.fixture
def regular_user():
return User(id=1, email="user@example.com", roles=["user"], is_active=True)
# Fixture for an admin
@pytest.fixture
def admin_user():
return User(id=2, email="admin@example.com", roles=["admin"], is_active=True)
def test_create_user_as_admin(admin_user):
"""Admin can create users."""
# Override current_user dependency for this test
app.dependency_overrides[current_user] = lambda: admin_user
response = client.post("/users", json={"email": "newuser@example.com"})
assert response.status_code == 200
assert response.json()["email"] == "newuser@example.com"
# Clean up
app.dependency_overrides.clear()
def test_create_user_as_regular_user(regular_user):
"""Regular users cannot create users."""
app.dependency_overrides[current_user] = lambda: regular_user
response = client.post("/users", json={"email": "newuser@example.com"})
assert response.status_code == 403
app.dependency_overrides.clear()
def test_get_stats_requires_admin(regular_user):
"""The /admin/stats endpoint rejects non-admins."""
app.dependency_overrides[current_user] = lambda: regular_user
response = client.get("/admin/stats")
assert response.status_code == 403
app.dependency_overrides.clear()
@pytest.fixture(autouse=True)
def cleanup():
"""Ensure overrides are cleared after each test."""
yield
app.dependency_overrides.clear()
This approach is far cleaner than mocking the JWT library or faking token generation. You declare exactly which user is making the request, and everything downstream works as if they'd authenticated normally.
Common mistakes
Forgetting to await async dependencies
If your current_user is async (which it often is for database calls), make sure to await it in your route:
@router.get("/me")
async def get_me(user: User = Depends(current_user)): # ✓ Correct
return user
@router.get("/me")
def get_me(user: User = Depends(current_user)): # ✗ Can cause issues with async deps
return user
Nesting dependencies too deeply
Each dependency adds overhead. If you find yourself writing Depends(require("admin", "moderator", "supervisor")), consider collapsing those roles into a single higher-level role or using require_any instead of multiple guards.
Not handling inactive users
Always check is_active (or equivalent) in current_user. A deleted or suspended account should be rejected even if the JWT is valid.
Logging and monitoring gaps
Auth failures are security events. Log them:
import logging
logger = logging.getLogger(__name__)
async def current_user(
credentials: HTTPAuthCredentials = Depends(security),
db: Session = Depends(get_db)
) -> User:
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
logger.warning("Invalid token payload")
raise HTTPException(status_code=401, detail="Invalid token payload")
except JWTError:
logger.warning(f"JWT validation failed for token: {token[:20]}...")
raise HTTPException(status_code=401, detail="Could not validate credentials")
user = db.query(User).filter(User.id == user_id).first()
if not user:
logger.warning(f"User {user_id} not found in database")
raise HTTPException(status_code=401, detail="User not found")
return user
Overusing router-level dependencies
It's tempting to add dependencies at the router level for every concern. But too many can mask what's really being enforced. Use them for broad categories (e.g., all admin routes), not for every nuance.
Key takeaways
- Write
current_useras a dependency that decodes the token, validates it, and loads the user from the database in one place. Reuse it everywhere. - Use a factory pattern with
require(role)to create role-checking dependencies on the fly. Compose them withrequire_anyandrequire_allfor complex rules. - Declare guards either on individual routes with
Depends(require("admin"))or at the router level withdependencies=[Depends(require("admin"))]. - Test guards by overriding the
current_userdependency with fixture users of different roles. No token mocking needed. - Log all auth failures. They're the canary in the coal mine for attacks and misconfigurations.
- Keep dependencies stateless and idempotent. They may run multiple times in a single request chain.
- Start with simple role checks and refactor only when you have real complexity. The pattern scales gracefully from two roles to dozens.
Comments
Loading comments…