Authentication Dependencies — Current User and Role Guards

Authentication dependencies are the most critical in a FastAPI application — they protect every route that requires a logged-in user. The dependency chain for a typical authenticated request is: get_current_user → reads the Authorization: Bearer <token> header → validates the JWT → looks up the user in the database → returns the User ORM object. This chapter covers JWT authentication mechanics in detail in Chapter 29; here we focus on how the dependency is structured, how it composes with role-based access control guards, and how optional authentication works for endpoints that serve both anonymous and logged-in users.

Authentication Dependency Architecture #

# app/dependencies/auth.py
from fastapi import Depends, Header, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from app.dependencies.db import get_db
from app.models.user import User
import jwt
import os

security = HTTPBearer()   # Extracts Bearer token from Authorization header

SECRET_KEY = os.environ["SECRET_KEY"]
ALGORITHM  = "HS256"

# ── Step 1: extract and validate the JWT ─────────────────────────────────────
def get_token_data(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    """Validate the Bearer token and return its payload."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM],
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code = status.HTTP_401_UNAUTHORIZED,
            detail      = "Token has expired",
            headers     = {"WWW-Authenticate": "Bearer"},
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code = status.HTTP_401_UNAUTHORIZED,
            detail      = "Invalid token",
            headers     = {"WWW-Authenticate": "Bearer"},
        )

# ── Step 2: load user from database using token's sub claim ──────────────────
def get_current_user(
    token_data: dict    = Depends(get_token_data),
    db:         Session = Depends(get_db),
) -> User:
    """Load and return the currently authenticated user."""
    user_id = token_data.get("sub")
    if not user_id:
        raise HTTPException(401, "Token missing subject claim")

    user = db.get(User, int(user_id))
    if not user or not user.is_active:
        raise HTTPException(
            status_code = status.HTTP_401_UNAUTHORIZED,
            detail      = "User not found or inactive",
        )
    return user

# ── Annotated aliases ─────────────────────────────────────────────────────────
from typing import Annotated
CurrentUser = Annotated[User, Depends(get_current_user)]

Optional Authentication #

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, Request

# auto_error=False: returns None instead of raising 403 when no token
optional_security = HTTPBearer(auto_error=False)

def get_optional_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(optional_security),
    db: Session = Depends(get_db),
) -> User | None:
    """
    Returns the current user if authenticated, or None for anonymous requests.
    Used for endpoints that work for both logged-in and anonymous users.
    """
    if credentials is None:
        return None   # anonymous user
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if not user_id:
            return None
        return db.get(User, int(user_id))
    except jwt.InvalidTokenError:
        return None   # invalid token = treat as anonymous

OptionalUser = Annotated[User | None, Depends(get_optional_user)]

# Usage: post list endpoint that shows like status for authenticated users
@router.get("/posts", response_model=list[PostResponse])
def list_posts(
    db:   Session      = Depends(get_db),
    user: OptionalUser = None,
):
    posts = load_published_posts(db)
    if user:
        liked_ids = load_liked_post_ids(db, user.id)
    else:
        liked_ids = set()
    ...

Role-Based Access Control Guards #

from fastapi import Depends, HTTPException

def require_role(*roles: str):
    """
    Factory function that returns a dependency requiring specific roles.
    Usage: Depends(require_role('admin', 'editor'))
    """
    def role_checker(current_user: CurrentUser) -> User:
        if current_user.role not in roles:
            raise HTTPException(
                status_code = 403,
                detail      = f"Requires role: {', '.join(roles)}",
            )
        return current_user
    return role_checker

# Convenience wrappers
def require_admin(user: CurrentUser) -> User:
    if user.role != "admin":
        raise HTTPException(403, "Admin access required")
    return user

def require_editor(user: CurrentUser) -> User:
    if user.role not in ("admin", "editor"):
        raise HTTPException(403, "Editor access required")
    return user

AdminUser  = Annotated[User, Depends(require_admin)]
EditorUser = Annotated[User, Depends(require_editor)]

# Usage
@router.delete("/posts/{post_id}/hard-delete", status_code=204)
def hard_delete_post(post_id: int, admin: AdminUser, db: DB):
    post = db.get(Post, post_id)
    if not post:
        raise HTTPException(404, "Post not found")
    db.delete(post)
    db.flush()

# Router-level guard: all routes require editor or higher
admin_router = APIRouter(
    prefix       = "/admin",
    tags         = ["Admin"],
    dependencies = [Depends(require_admin)],
)

Common Mistakes #

Mistake 1 — Not setting WWW-Authenticate header on 401 responses #

❌ Wrong — OAuth2 spec requires this header on 401:

raise HTTPException(401, "Not authenticated")   # missing WWW-Authenticate header

✅ Correct:

raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})   # ✓

Mistake 2 — Not checking is_active after JWT validation #

❌ Wrong — deactivated user can still access API with a valid token:

user = db.get(User, user_id)
if not user:
    raise HTTPException(401, "User not found")
return user   # user.is_active might be False!

✅ Correct:

if not user or not user.is_active:
    raise HTTPException(401, "User not found or inactive")   # ✓

Mistake 3 — Using optional_security on endpoints that must require auth #

❌ Wrong — endpoint silently accepts anonymous users when it should reject them:

@router.post("/posts")
def create_post(user: OptionalUser):
    if not user: return {"error": "Please log in"}   # 200 instead of 401!

✅ Correct — use CurrentUser (not OptionalUser) for endpoints requiring authentication.

Quick Reference #