FastAPI in Practice: JWT Auth, Request-Scoped Context, and Structured Logs (Hands-On)

FastAPI makes it easy to ship APIs quickly, but production-ish APIs need a few extras: authentication, consistent error handling, and logs you can actually search. In this hands-on guide, you’ll build a small FastAPI service with JWT auth, request-scoped context (request ID + user ID), and structured JSON logs—without turning your project into a framework monster.

You’ll end up with:

A /auth/login endpoint that issues JWTs
A protected /me route using a dependency
Request IDs and user IDs injected into every log line
Consistent JSON error responses

1) Project setup

Create a new folder and install dependencies:

python -m venv .venv # macOS/Linux source .venv/bin/activate # Windows (PowerShell) # .\.venv\Scripts\Activate.ps1 pip install fastapi uvicorn[standard] python-jose[cryptography] passlib[bcrypt] structlog

What these are for:

python-jose to sign/verify JWTs
passlib[bcrypt] for password hashing
structlog for JSON logs

2) A minimal app with structured logging

Create app.py:

import logging import sys import time import uuid from contextvars import ContextVar from typing import Optional import structlog from fastapi import FastAPI, Request from fastapi.responses import JSONResponse # Request-scoped context variables request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-") user_id_ctx: ContextVar[str] = ContextVar("user_id", default="anonymous") def add_request_context(_, __, event_dict): event_dict["request_id"] = request_id_ctx.get() event_dict["user_id"] = user_id_ctx.get() return event_dict def configure_logging(): logging.basicConfig( format="%(message)s", stream=sys.stdout, level=logging.INFO, ) structlog.configure( processors=[ structlog.processors.TimeStamper(fmt="iso"), add_request_context, structlog.processors.add_log_level, structlog.processors.JSONRenderer(), ], wrapper_class=structlog.make_filtering_bound_logger(logging.INFO), cache_logger_on_first_use=True, ) configure_logging() log = structlog.get_logger() app = FastAPI(title="Practical FastAPI Service") @app.middleware("http") async def request_context_middleware(request: Request, call_next): rid = request.headers.get("X-Request-ID") or str(uuid.uuid4()) request_id_ctx.set(rid) start = time.time() try: response = await call_next(request) return response finally: duration_ms = round((time.time() - start) * 1000, 2) log.info( "request_complete", method=request.method, path=request.url.path, status=getattr(locals().get("response"), "status_code", None), duration_ms=duration_ms, ) @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): # In real apps, you might want to hide details in production log.error("unhandled_exception", error=str(exc), path=request.url.path) return JSONResponse( status_code=500, content={"error": {"code": "INTERNAL_ERROR", "message": "Something went wrong."}}, ) @app.get("/health") def health(): log.info("health_check") return {"status": "ok"}

Run it:

uvicorn app:app --reload

Hit /health and you should see JSON logs in your terminal. Notice how every line includes request_id (generated if absent). This is the foundation for tracing requests across services.

3) JWT authentication (login + protect routes)

Now let’s add:

Password hashing
JWT creation
A dependency that reads Authorization: Bearer ...

Append this to app.py (below the existing code), or split into modules later.

from datetime import datetime, timedelta from fastapi import Depends, HTTPException from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel # --- Security config (keep secrets in env vars in real apps) --- JWT_SECRET = "dev-only-change-me" JWT_ALG = "HS256" JWT_EXPIRES_MIN = 60 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") bearer = HTTPBearer(auto_error=False) # --- Fake user store (replace with DB in real apps) --- # Password is "password123" FAKE_USERS = { "alice": { "id": "u_1001", "username": "alice", "hashed_password": pwd_context.hash("password123"), "role": "member", } } class LoginRequest(BaseModel): username: str password: str class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" expires_in: int def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_access_token(*, sub: str, user_id: str, role: str) -> str: exp = datetime.utcnow() + timedelta(minutes=JWT_EXPIRES_MIN) payload = {"sub": sub, "uid": user_id, "role": role, "exp": exp} return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALG) def decode_token(token: str) -> dict: return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG]) def get_current_user(creds: Optional[HTTPAuthorizationCredentials] = Depends(bearer)) -> dict: if creds is None or creds.scheme.lower() != "bearer": raise HTTPException(status_code=401, detail="Missing bearer token") try: payload = decode_token(creds.credentials) username = payload.get("sub") uid = payload.get("uid") role = payload.get("role") if not username or not uid: raise HTTPException(status_code=401, detail="Invalid token payload") # Set request-scoped user id for logging user_id_ctx.set(uid) return {"username": username, "id": uid, "role": role} except JWTError: raise HTTPException(status_code=401, detail="Invalid or expired token") @app.post("/auth/login", response_model=TokenResponse) def login(body: LoginRequest): user = FAKE_USERS.get(body.username) if not user or not verify_password(body.password, user["hashed_password"]): log.info("login_failed", username=body.username) raise HTTPException(status_code=401, detail="Invalid credentials") token = create_access_token(sub=user["username"], user_id=user["id"], role=user["role"]) log.info("login_success", username=user["username"]) return TokenResponse(access_token=token, expires_in=JWT_EXPIRES_MIN * 60) @app.get("/me") def me(current_user: dict = Depends(get_current_user)): log.info("me_called") return {"id": current_user["id"], "username": current_user["username"], "role": current_user["role"]}

Test it with curl:

# Login TOKEN=$(curl -s -X POST http://127.0.0.1:8000/auth/login \ -H "Content-Type: application/json" \ -d '{"username":"alice","password":"password123"}' | python -c "import sys, json; print(json.load(sys.stdin)['access_token'])") # Call /me with the token curl -s http://127.0.0.1:8000/me -H "Authorization: Bearer $TOKEN" | python -m json.tool

Check your logs: once authenticated, you should see user_id change from anonymous to the user’s ID for that request.

4) Make error responses consistent (without hiding useful info)

Right now, FastAPI’s HTTPException returns {"detail": ...}, while unhandled errors return {"error": ...}. That inconsistency becomes annoying for frontend clients.

Add a handler for HTTPException to standardize output:

from fastapi.exceptions import HTTPException as FastAPIHTTPException @app.exception_handler(FastAPIHTTPException) async def http_exception_handler(request: Request, exc: FastAPIHTTPException): log.info("http_error", status=exc.status_code, detail=str(exc.detail), path=request.url.path) return JSONResponse( status_code=exc.status_code, content={"error": {"code": "HTTP_ERROR", "message": exc.detail}}, )

Now clients can always expect error.code and error.message.

5) A simple “role check” dependency (common real-world need)

Many APIs have endpoints like “admin-only”. You can express that cleanly with a dependency factory:

def require_role(required: str): def _checker(user: dict = Depends(get_current_user)) -> dict: if user.get("role") != required: raise HTTPException(status_code=403, detail="Forbidden") return user return _checker @app.get("/admin/stats") def admin_stats(admin_user: dict = Depends(require_role("admin"))): return {"secrets": "only admins can see this"}

In a real app, roles/permissions usually live in your DB and your token might include a permissions list. The pattern stays the same: keep auth logic in dependencies so your route code stays boring.

6) Practical checklist for taking this further

Move secrets to environment variables: never hardcode JWT_SECRET outside local dev.
Add refresh tokens if you need long sessions (short access token + longer refresh token).
Persist users in a DB and replace the fake store with queries.
Log sparingly: don’t log passwords, full tokens, or sensitive headers.
Return request IDs to clients: set X-Request-ID on responses so support can ask for it.

If you want the response header too, tweak the middleware:

@app.middleware("http") async def request_context_middleware(request: Request, call_next): rid = request.headers.get("X-Request-ID") or str(uuid.uuid4()) request_id_ctx.set(rid) start = time.time() response = await call_next(request) response.headers["X-Request-ID"] = rid duration_ms = round((time.time() - start) * 1000, 2) log.info( "request_complete", method=request.method, path=request.url.path, status=response.status_code, duration_ms=duration_ms, ) return response

Wrap-up

You now have a compact FastAPI service with JWT auth, request IDs, user-aware logs, and consistent error responses—exactly the kind of “small but real” baseline that helps junior/mid developers ship APIs that are easier to debug and operate.

Next steps if you want to keep leveling it up: add a database layer, migrate settings to pydantic-settings, and add automated tests for auth flows (login success/failure, expired tokens, forbidden role).

FastAPI in Practice: JWT Auth, Request-Scoped Context, and Structured Logs (Hands-On)