FastAPI in Production: Auth, Caching, Background Tasks, and Kubernetes Deployment

FastAPI combines Python type hints, automatic validation, and async support into the fastest path to correct APIs.

Configuration with pydantic-settings

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    debug: bool = False
    database_url: str
    redis_url: str = "redis://localhost:6379"
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    model_config = SettingsConfigDict(env_file='.env')

@lru_cache
def get_settings() -> Settings:
    return Settings()

JWT Authentication

from datetime import datetime, timedelta
from jose import JWTError, jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

bearer = HTTPBearer()

def create_access_token(sub: str) -> str:
    payload = {
        "sub": sub,
        "exp": datetime.utcnow() + timedelta(minutes=get_settings().access_token_expire_minutes),
        "type": "access"
    }
    return jwt.encode(payload, get_settings().secret_key, algorithm=get_settings().algorithm)

async def get_current_user(
    creds: HTTPAuthorizationCredentials = Depends(bearer),
    db = Depends(get_db)
) -> User:
    try:
        payload = jwt.decode(creds.credentials, get_settings().secret_key,
                             algorithms=[get_settings().algorithm])
        if payload.get("type") != "access":
            raise HTTPException(status_code=401)
        user = await db.get(User, int(payload["sub"]))
        if not user or not user.is_active:
            raise HTTPException(status_code=401)
        return user
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Redis Caching Decorator

import redis.asyncio as aioredis
import json
from functools import wraps

cache = aioredis.from_url(get_settings().redis_url, decode_responses=True)

def cached(key_prefix: str, ttl: int = 300):
    def decorator(fn):
        @wraps(fn)
        async def wrapper(*args, **kwargs):
            key = f"{key_prefix}:{':'.join(str(a) for a in args)}"
            hit = await cache.get(key)
            if hit:
                return json.loads(hit)
            result = await fn(*args, **kwargs)
            await cache.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

Background Tasks with Celery

from celery import Celery

celery_app = Celery('tasks', broker=get_settings().redis_url)

@celery_app.task(bind=True, max_retries=3)
def send_welcome_email(self, user_id: int, email: str):
    try:
        mailer.send(to=email, template='welcome')
    except Exception as e:
        self.retry(exc=e, countdown=60 * (2 ** self.request.retries))

@router.post("/users/", status_code=201)
async def create_user(data: UserCreate, bg: BackgroundTasks, db=Depends(get_db)):
    user = await user_service.create(db, data)
    bg.add_task(send_welcome_email.delay, user.id, user.email)
    return UserResponse.from_orm(user)

Request Logging Middleware

import time, uuid, structlog
from starlette.middleware.base import BaseHTTPMiddleware

log = structlog.get_logger()

class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        req_id = str(uuid.uuid4())
        start = time.perf_counter()
        resp = await call_next(request)
        log.info("request", id=req_id, method=request.method,
                 path=request.url.path, status=resp.status_code,
                 ms=round((time.perf_counter() - start) * 1000, 2))
        resp.headers["X-Request-ID"] = req_id
        return resp

FastAPI collapses validation, business logic, and serialization into one coherent system.

→ Test your API's JWT tokens with the JWT Parser tool.