fastapi_middleware 10 Q&As

FastAPI Middleware FAQ & Answers

10 expert FastAPI Middleware answers researched from official documentation. Every answer cites authoritative sources you can verify.

unknown

10 questions
A

FastAPI middleware performs exact string matching on request paths, which means trailing slashes are significant. A path '/api/health' will NOT match '/api/health/' in excluded_paths unless both variants are explicitly listed.

Path matching behavior:

  1. Exact Match Only: Middleware compares request.url.path against excluded paths using Python's in operator or explicit equality checks
  2. No Automatic Normalization: FastAPI does not automatically normalize trailing slashes in middleware path matching
  3. Case Sensitive: Path matching is case-sensitive

Example middleware with proper path exclusion:

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, excluded_paths: list):
        super().__init__(app)
        # Include both variants for robustness
        self.excluded_paths = set(excluded_paths)
    
    async def dispatch(self, request: Request, call_next):
        # Exact path matching
        if request.url.path in self.excluded_paths:
            return await call_next(request)
        
        # Add authentication logic here
        return await call_next(request)

app = FastAPI()

# Best practice: include both trailing slash variants
excluded = [
    '/api/health',
    '/api/health/',
    '/api/public',
    '/api/public/'
]

app.add_middleware(AuthMiddleware, excluded_paths=excluded)

Best Practices:

  • Include both trailing slash variants in excluded_paths
  • Use prefix matching for entire path trees: request.url.path.startswith('/api/public')
  • Consider using path normalization in middleware: path = request.url.path.rstrip('/')
  • For production, combine with FastAPI's RedirectSlashesMiddleware to enforce consistency

Version Note: Applies to FastAPI 0.68.0+ and Starlette 0.14.0+

99% confidence
A

FastAPI concatenates the APIRouter prefix with route decorator paths using simple string concatenation, following these rules:

Concatenation Rules:

  1. No Automatic Slash Handling: Prefix + path are concatenated directly
  2. Leading Slash Required: Route paths must start with / (enforced by FastAPI)
  3. No Trailing Slash: Prefixes should NOT end with / to avoid double slashes
  4. Empty String Allowed: Empty prefix "" or no prefix is valid

Path Resolution Examples:

from fastapi import APIRouter, FastAPI

# Example 1: Standard prefix
router1 = APIRouter(prefix="/api/v1")

@router1.get("/users")  # Results in: /api/v1/users
async def get_users():
    return {"users": []}

@router1.get("/users/{id}")  # Results in: /api/v1/users/{id}
async def get_user(id: int):
    return {"user_id": id}

# Example 2: Nested routers
api_router = APIRouter(prefix="/api")
v1_router = APIRouter(prefix="/v1")

@v1_router.get("/items")  # On v1_router: /v1/items
async def get_items():
    return {"items": []}

api_router.include_router(v1_router)  # Results in: /api/v1/items

# Example 3: Empty prefix
router3 = APIRouter(prefix="")

@router3.get("/health")  # Results in: /health
async def health():
    return {"status": "ok"}

# Example 4: Root path on router with prefix
router4 = APIRouter(prefix="/api")

@router4.get("")  # Results in: /api (NOT /api/)
async def api_root():
    return {"message": "API root"}

app = FastAPI()
app.include_router(router1)
app.include_router(api_router)
app.include_router(router3)
app.include_router(router4)

Common Pitfalls:

  • Double Slash: prefix="/api/" + path="/users"/api//users (invalid)
  • Missing Leading Slash: prefix="/api" + path="users" → Error (FastAPI validates)
  • Prefix Without Slash: prefix="api" → Error (must start with /)

Tag Inheritance:
Routers also support tags that apply to all routes:

router = APIRouter(
    prefix="/api/v1",
    tags=["v1"],
    responses={404: {"description": "Not found"}}
)

Version Note: Behavior consistent since FastAPI 0.40.0+

99% confidence
A

No, FastAPI does NOT automatically redirect requests to add or remove trailing slashes by default. Each route is treated as a distinct endpoint.

Default Behavior:

  • /users and /users/ are different routes
  • A request to /users/ when only /users is defined returns 404 Not Found
  • No automatic redirect occurs

Enabling Automatic Redirects:

Use Starlette's RedirectSlashesMiddleware (available since Starlette 0.31.0):

from fastapi import FastAPI
from starlette.middleware import Middleware
from starlette.middleware.redirectslashes import RedirectSlashesMiddleware

# Option 1: Add middleware to existing app
app = FastAPI()
app.add_middleware(RedirectSlashesMiddleware)

# Option 2: Configure during app initialization
app = FastAPI(
    middleware=[
        Middleware(RedirectSlashesMiddleware)
    ]
)

@app.get("/users")
async def get_users():
    return {"users": []}

# Now both /users and /users/ work:
# GET /users → 200 OK
# GET /users/ → 307 Temporary Redirect to /users

Redirect Behavior with Middleware:

  • Removes trailing slash: /users/ → 307 redirect to /users
  • HTTP Status: 307 Temporary Redirect (preserves HTTP method)
  • Preserves Query Parameters: /users/?page=2/users?page=2
  • POST/PUT/DELETE: Method preserved due to 307 status code

Alternative: Manual Route Duplication

@app.get("/users")
@app.get("/users/")
async def get_users():
    return {"users": []}

Best Practices:

  1. Choose a Convention: Either always use trailing slashes or never use them
  2. Document Clearly: Specify in API documentation whether trailing slashes are significant
  3. Use Middleware for Flexibility: RedirectSlashesMiddleware allows both variants
  4. Consider SEO: For public APIs, redirects help with URL variations
  5. OpenAPI Documentation: Only the defined route appears in /docs, redirects are runtime behavior

Middleware Ordering:
RedirectSlashesMiddleware should be added early in the middleware stack:

app.add_middleware(RedirectSlashesMiddleware)  # First
app.add_middleware(CORSMiddleware, ...)        # Then other middleware

Version Note: RedirectSlashesMiddleware available in Starlette 0.31.0+, FastAPI 0.100.0+

99% confidence
A

FastAPI middleware and dependency injection are separate layers with different error handling mechanisms. Dependencies used in routes raise HTTPException which FastAPI catches and returns as HTTP responses, but middleware errors require explicit exception handling.

Error Handling in Route Dependencies:

from fastapi import Depends, FastAPI, HTTPException
from typing import Annotated

def verify_token(token: str = Header()):
    if token != "valid-token":
        raise HTTPException(
            status_code=401,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"}
        )
    return token

@app.get("/protected")
async def protected_route(token: Annotated[str, Depends(verify_token)]):
    return {"message": "Access granted"}

# HTTPException automatically converted to 401 response

Error Handling in Middleware:

Middleware must catch and handle exceptions explicitly:

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import logging

logger = logging.getLogger(__name__)

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            # Dependency-like logic in middleware
            token = request.headers.get("Authorization")
            
            if not token:
                return JSONResponse(
                    status_code=401,
                    content={"detail": "Missing authorization header"},
                    headers={"WWW-Authenticate": "Bearer"}
                )
            
            if not self.verify_token(token):
                return JSONResponse(
                    status_code=401,
                    content={"detail": "Invalid token"}
                )
            
            # Add user context to request state
            request.state.user = self.get_user(token)
            
            response = await call_next(request)
            return response
            
        except HTTPException as e:
            # HTTPException in middleware must be manually converted
            return JSONResponse(
                status_code=e.status_code,
                content={"detail": e.detail},
                headers=e.headers
            )
        except Exception as e:
            # Catch unexpected errors
            logger.error(f"Middleware error: {e}", exc_info=True)
            return JSONResponse(
                status_code=500,
                content={"detail": "Internal server error"}
            )
    
    def verify_token(self, token: str) -> bool:
        # Token validation logic
        return token.startswith("Bearer ")
    
    def get_user(self, token: str):
        # User lookup logic
        return {"id": 1, "name": "User"}

Key Differences:

Aspect Route Dependencies Middleware
HTTPException Auto-converted to response Must manually handle
Error Propagation FastAPI handles Must catch explicitly
Access to call_next No Yes
Execution Order After middleware Before route
Request Modification Limited Full control

Hybrid Approach: Middleware with Dependencies

FastAPI 0.95.0+ supports dependencies in middleware via request.state:

from fastapi import Depends, FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

async def get_current_user(request: Request):
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return {"id": 1, "name": "User"}

class UserMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            # Manually resolve dependency
            user = await get_current_user(request)
            request.state.user = user
        except HTTPException as e:
            return JSONResponse(
                status_code=e.status_code,
                content={"detail": e.detail}
            )
        
        response = await call_next(request)
        return response

# Then in routes, access via request.state
@app.get("/profile")
async def profile(request: Request):
    return {"user": request.state.user}

Best Practices:

  1. Use dependencies for business logic (auth, validation)
  2. Use middleware for cross-cutting concerns (logging, timing)
  3. Always wrap middleware logic in try/except
  4. Log errors in middleware for debugging
  5. Return appropriate HTTP status codes
  6. Use request.state to pass data from middleware to routes

Version Note: Applies to FastAPI 0.68.0+, improved dependency resolution in 0.95.0+

99% confidence
A

FastAPI processes requests through a specific execution pipeline: middleware runs first (outermost to innermost), then dependencies (in declaration order), then the route handler. The response flows back through the same layers in reverse.

Request/Response Flow:

INBOUND REQUEST:
1. Middleware Layer 1 (first added) - before call_next()
2. Middleware Layer 2 (second added) - before call_next()
3. Middleware Layer N (last added) - before call_next()
4. Route Dependencies (in declaration order)
5. Path Operation Function (route handler)

OUTBOUND RESPONSE:
6. Middleware Layer N (last added) - after call_next()
7. Middleware Layer 2 (second added) - after call_next()
8. Middleware Layer 1 (first added) - after call_next()
9. Return to client

Detailed Example:

from fastapi import Depends, FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# MIDDLEWARE LAYER 1 (added first, executes OUTERMOST)
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print("1. TimingMiddleware - BEFORE call_next")
        start = time.time()
        
        response = await call_next(request)  # Passes to next layer
        
        duration = time.time() - start
        print(f"8. TimingMiddleware - AFTER call_next ({duration:.3f}s)")
        response.headers["X-Process-Time"] = str(duration)
        return response

# MIDDLEWARE LAYER 2 (added second, executes INSIDE Layer 1)
class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print("2. AuthMiddleware - BEFORE call_next")
        token = request.headers.get("Authorization")
        request.state.token = token
        
        response = await call_next(request)  # Passes to next layer
        
        print("7. AuthMiddleware - AFTER call_next")
        return response

# MIDDLEWARE LAYER 3 (added third, executes INSIDE Layer 2)
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print(f"3. LoggingMiddleware - BEFORE call_next - {request.url.path}")
        
        response = await call_next(request)  # Passes to dependencies
        
        print(f"6. LoggingMiddleware - AFTER call_next - {response.status_code}")
        return response

app.add_middleware(TimingMiddleware)    # Outermost
app.add_middleware(AuthMiddleware)      # Middle
app.add_middleware(LoggingMiddleware)   # Innermost

# DEPENDENCY LAYER (executes after all middleware)
async def verify_token(request: Request):
    print("4. Dependency: verify_token")
    token = request.state.token
    if not token:
        raise HTTPException(status_code=401, detail="No token")
    return {"user_id": 123}

async def get_db():
    print("4. Dependency: get_db")
    db = {"connection": "active"}
    try:
        yield db
    finally:
        print("5. Dependency cleanup: get_db")

# ROUTE HANDLER (executes last)
@app.get("/users")
async def get_users(
    user: dict = Depends(verify_token),  # Executes first
    db: dict = Depends(get_db)           # Executes second
):
    print("5. Route Handler: get_users")
    return {"users": [], "auth": user}

Console Output for GET /users:

1. TimingMiddleware - BEFORE call_next
2. AuthMiddleware - BEFORE call_next
3. LoggingMiddleware - BEFORE call_next - /users
4. Dependency: verify_token
4. Dependency: get_db
5. Route Handler: get_users
5. Dependency cleanup: get_db
6. LoggingMiddleware - AFTER call_next - 200
7. AuthMiddleware - AFTER call_next
8. TimingMiddleware - AFTER call_next (0.003s)

Key Principles:

  1. Middleware Stack (LIFO for response): Last middleware added executes closest to dependencies
  2. Dependencies Execute in Order: Listed left-to-right in function signature
  3. Cleanup in Reverse: Generator dependencies cleanup after route handler
  4. Early Exit: Middleware can return response before call_next(), skipping remaining pipeline
  5. State Passing: Middleware can set request.state.* for dependencies to access

Early Exit Example:

class CacheMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        cached = get_from_cache(request.url.path)
        if cached:
            # Skip dependencies and route handler entirely
            return JSONResponse(content=cached)
        
        response = await call_next(request)
        save_to_cache(request.url.path, response)
        return response

Middleware vs Dependencies Decision Guide:

  • Use Middleware For: Request/response transformation, timing, logging, early exits, global behavior
  • Use Dependencies For: Authentication (route-specific), database connections, business logic, dependency injection, testability

Version Note: Execution order consistent since FastAPI 0.40.0+, Starlette 0.13.0+

99% confidence
A

FastAPI's RedirectResponse uses 307 Temporary Redirect as the default status code (not 302 or 303).

Default Behavior:

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/old-path")
async def old_path():
    # Default: 307 Temporary Redirect
    return RedirectResponse(url="/new-path")
    
@app.get("/explicit-307")
async def explicit_307():
    # Explicitly specify 307 (same as default)
    return RedirectResponse(url="/new-path", status_code=307)

HTTP Redirect Status Codes Comparison:

Code Name Method Preserved Use Case
302 Found ❌ Changes to GET Legacy temporary redirect, browsers may change POST to GET
303 See Other ❌ Always GET After POST/PUT/DELETE, redirect to GET resource
307 Temporary Redirect ✅ Yes FastAPI default, preserves HTTP method
308 Permanent Redirect ✅ Yes Permanent redirect preserving method
301 Moved Permanently ❌ Changes to GET SEO-friendly permanent redirect

Why 307 is the Default:

  1. Method Preservation: A POST to /old-path redirects as POST to /new-path
  2. RFC 7231 Compliance: Modern HTTP/1.1 standard
  3. Predictable Behavior: No automatic method transformation
  4. Form Safety: POST form submissions remain POST after redirect

Common Redirect Patterns:

from fastapi import FastAPI, status
from fastapi.responses import RedirectResponse

app = FastAPI()

# Pattern 1: Post-Redirect-Get (PRG) - Use 303
@app.post("/submit-form")
async def submit_form():
    # Process form data...
    # Redirect to GET endpoint to prevent form resubmission
    return RedirectResponse(
        url="/success",
        status_code=status.HTTP_303_SEE_OTHER  # Forces GET
    )

# Pattern 2: Temporary redirect preserving method - Use 307 (default)
@app.post("/api/v1/users")
async def create_user_v1():
    # Temporarily redirect to v2 API
    return RedirectResponse(
        url="/api/v2/users",
        status_code=307  # Preserves POST method
    )

# Pattern 3: Permanent redirect - Use 308 or 301
@app.get("/old-blog-post")
async def old_blog():
    return RedirectResponse(
        url="/new-blog-post",
        status_code=status.HTTP_308_PERMANENT_REDIRECT  # Preserves method
        # or status.HTTP_301_MOVED_PERMANENTLY for GET-only redirects
    )

# Pattern 4: Login redirect - Use 302 or 303
@app.get("/protected")
async def protected_resource(authenticated: bool = False):
    if not authenticated:
        return RedirectResponse(
            url="/login",
            status_code=status.HTTP_303_SEE_OTHER  # Always GET
        )
    return {"data": "protected content"}

Browser Behavior:

  • 307/308: Browser shows confirmation dialog for POST/PUT/DELETE redirects (security feature)
  • 302/303: Browser automatically follows without confirmation
  • 301: Browsers may cache permanently (use carefully)

Security Considerations:

from urllib.parse import urlparse

@app.get("/redirect")
async def safe_redirect(next_url: str):
    # Validate redirect URL to prevent open redirect attacks
    parsed = urlparse(next_url)
    
    # Only allow relative paths or same domain
    if parsed.netloc and parsed.netloc != "yourdomain.com":
        return {"error": "Invalid redirect URL"}
    
    return RedirectResponse(url=next_url)

OpenAPI Documentation:

Redirects can be documented in OpenAPI schema:

@app.get(
    "/old-endpoint",
    status_code=307,
    responses={
        307: {
            "description": "Redirect to new endpoint",
            "headers": {
                "Location": {"schema": {"type": "string"}}
            }
        }
    }
)
async def old_endpoint():
    return RedirectResponse(url="/new-endpoint")

Version Note: Default 307 behavior since FastAPI 0.1.0, follows Starlette's RedirectResponse

99% confidence
A

In FastAPI, the final endpoint URL is formed by concatenating the APIRouter prefix with the individual route path. The prefix applies to all routes within that router, while the route path is specific to each endpoint.

URL Formation Formula:

Final URL = Router Prefix + Route Path

Basic Example:

from fastapi import APIRouter, FastAPI

app = FastAPI()

# Router with prefix
users_router = APIRouter(prefix="/api/v1")

@users_router.get("/users")           # Final URL: /api/v1/users
async def get_users():
    return {"users": []}

@users_router.get("/users/{id}")      # Final URL: /api/v1/users/{id}
async def get_user(id: int):
    return {"user_id": id}

@users_router.post("/users")          # Final URL: /api/v1/users
async def create_user():
    return {"created": True}

app.include_router(users_router)

Key Differences:

Aspect Router Prefix Route Path
Scope Applies to ALL routes in router Specific to ONE endpoint
Reusability Shared across router Unique per route
Nesting Can be nested (router in router) Cannot be nested
Empty Value prefix="" is valid Path must start with /
Modification Set once at router creation Set per route decorator

Advanced Patterns:

from fastapi import APIRouter, FastAPI

app = FastAPI()

# Pattern 1: Nested Routers
api_router = APIRouter(prefix="/api")
v1_router = APIRouter(prefix="/v1")
v2_router = APIRouter(prefix="/v2")

@v1_router.get("/items")              # Final: /api/v1/items
async def get_items_v1():
    return {"version": "v1", "items": []}

@v2_router.get("/items")              # Final: /api/v2/items
async def get_items_v2():
    return {"version": "v2", "items": []}

api_router.include_router(v1_router)
api_router.include_router(v2_router)
app.include_router(api_router)

# Pattern 2: Empty Prefix
root_router = APIRouter(prefix="")     # No prefix

@root_router.get("/health")           # Final: /health
async def health():
    return {"status": "ok"}

app.include_router(root_router)

# Pattern 3: Root Path on Prefixed Router
api_root_router = APIRouter(prefix="/api")

@api_root_router.get("")              # Final: /api (not /api/)
async def api_root():
    return {"message": "API root"}

@api_root_router.get("/")             # ERROR: Results in /api/ which is different from /api
async def api_root_slash():           # FastAPI treats /api and /api/ as different routes
    return {"message": "API root with slash"}

app.include_router(api_root_router)

# Pattern 4: Multiple Routers with Same Prefix
users_router = APIRouter(prefix="/api/v1", tags=["users"])
products_router = APIRouter(prefix="/api/v1", tags=["products"])

@users_router.get("/users")           # Final: /api/v1/users
async def get_users():
    return {"users": []}

@products_router.get("/products")     # Final: /api/v1/products
async def get_products():
    return {"products": []}

app.include_router(users_router)
app.include_router(products_router)

Common Pitfalls:

# ❌ WRONG: Double slash
router = APIRouter(prefix="/api/")  # Trailing slash

@router.get("/users")  # Results in /api//users (invalid)
async def get_users():
    pass

# ✅ CORRECT: No trailing slash on prefix
router = APIRouter(prefix="/api")

@router.get("/users")  # Results in /api/users
async def get_users():
    pass

# ❌ WRONG: Missing leading slash on route
router = APIRouter(prefix="/api")

@router.get("users")  # ERROR: Path must start with /
async def get_users():
    pass

# ✅ CORRECT: Leading slash on route path
@router.get("/users")  # Results in /api/users
async def get_users():
    pass

Prefix Inheritance with Tags and Dependencies:

from fastapi import APIRouter, Depends

async def api_key_auth(api_key: str = Header()):
    if api_key != "secret":
        raise HTTPException(status_code=403)
    return api_key

# Prefix + tags + dependencies all inherited
api_router = APIRouter(
    prefix="/api/v1",
    tags=["api", "v1"],
    dependencies=[Depends(api_key_auth)],  # Applied to ALL routes
    responses={404: {"description": "Not found"}}
)

@api_router.get("/users")  # Inherits prefix, tags, and dependencies
async def get_users():
    return {"users": []}

URL Visibility:

  • OpenAPI Docs: Shows final concatenated URL (/api/v1/users)
  • Route Listing: app.routes shows complete paths
  • Client Requests: Must use complete URL including prefix

Best Practices:

  1. Use prefixes for versioning: /api/v1, /api/v2
  2. Group related routes in routers with shared prefix
  3. Never end prefix with /
  4. Always start route path with /
  5. Use nested routers for complex API structures
  6. Apply common dependencies/tags at router level

Version Note: Behavior consistent since FastAPI 0.40.0+

99% confidence
A

FastAPI processes middleware in reverse order of registration: the first middleware added runs outermost (first on request, last on response), and the last middleware added runs innermost (last on request, first on response). SessionMiddleware must be positioned carefully to ensure session data is available when needed.

Middleware Execution Order:

from fastapi import FastAPI
from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(TimingMiddleware)      # Executes 1st (outermost)
app.add_middleware(CORSMiddleware, ...)   # Executes 2nd
app.add_middleware(SessionMiddleware, ...) # Executes 3rd
app.add_middleware(AuthMiddleware)        # Executes 4th (innermost)

# Request flow: Timing → CORS → Session → Auth → Route
# Response flow: Auth → Session → CORS → Timing → Client

SessionMiddleware Positioning Rules:

  1. After CORS Middleware: CORS needs to run before session cookies are set
  2. Before Authentication Middleware: Auth often needs to read/write session data
  3. Before Application Middleware: Any middleware that uses request.session must run after SessionMiddleware

Correct Middleware Stack:

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import secrets

app = FastAPI()

# Layer 1: CORS (outermost) - handles preflight, sets CORS headers
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Layer 2: Session - provides request.session to inner middleware
app.add_middleware(
    SessionMiddleware,
    secret_key=secrets.token_hex(32),  # Use secure secret in production
    session_cookie="session_id",
    max_age=1800,  # 30 minutes
    same_site="lax",
    https_only=True  # Production only
)

# Layer 3: Custom Auth - can now use request.session
class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # ✅ Session available here
        user_id = request.session.get("user_id")
        
        if user_id:
            request.state.user = await get_user(user_id)
        
        response = await call_next(request)
        return response

app.add_middleware(AuthMiddleware)

# Layer 4: Other application middleware (innermost)
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # ✅ Session and user available here
        user_id = request.session.get("user_id")
        print(f"Request by user: {user_id}")
        
        response = await call_next(request)
        return response

app.add_middleware(LoggingMiddleware)

Why Order Matters for SessionMiddleware:

# ❌ WRONG ORDER - SessionMiddleware after Auth
app.add_middleware(AuthMiddleware)        # Tries to access request.session
app.add_middleware(SessionMiddleware, ...) # But session not yet available!

# Result: AttributeError: 'Request' object has no attribute 'session'

# ✅ CORRECT ORDER - SessionMiddleware before Auth
app.add_middleware(SessionMiddleware, ...)
app.add_middleware(AuthMiddleware)        # Now request.session is available

Common Middleware Stack Pattern:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.sessions import SessionMiddleware
import secrets

app = FastAPI()

# Recommended order (outermost to innermost):

# 1. Trusted Host (security) - outermost
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

# 2. CORS (cross-origin)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 3. GZip (compression)
app.add_middleware(GZipMiddleware, minimum_size=1000)

# 4. Session (state management)
app.add_middleware(
    SessionMiddleware,
    secret_key=secrets.token_hex(32)
)

# 5. Custom timing middleware
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

app.add_middleware(TimingMiddleware)

# 6. Authentication (business logic) - innermost
app.add_middleware(AuthMiddleware)

OAuth Flow Example:

from authlib.integrations.starlette_client import OAuth

app = FastAPI()

# SessionMiddleware required BEFORE OAuth setup
app.add_middleware(
    SessionMiddleware,
    secret_key=secrets.token_hex(32)
)

oauth = OAuth()
oauth.register(
    name='google',
    client_id='...',
    client_secret='...',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

@app.get("/login")
async def login(request: Request):
    # ✅ request.session available
    redirect_uri = request.url_for('auth')
    return await oauth.google.authorize_redirect(request, redirect_uri)

@app.get("/auth")
async def auth(request: Request):
    # ✅ request.session available for OAuth state verification
    token = await oauth.google.authorize_access_token(request)
    user = token.get('userinfo')
    request.session['user'] = dict(user)
    return {"user": user}

SessionMiddleware Configuration:

app.add_middleware(
    SessionMiddleware,
    secret_key="your-secret-key-min-32-chars",  # Required
    session_cookie="session",      # Cookie name (default: "session")
    max_age=1800,                  # Seconds (default: 2 weeks)
    same_site="lax",               # "lax", "strict", or "none"
    https_only=True,               # Production: True, Dev: False
    path="/"                       # Cookie path
)

Debugging Middleware Order:

# Print middleware order
for middleware in app.user_middleware:
    print(middleware.cls.__name__)

# Output (innermost to outermost):
# AuthMiddleware
# TimingMiddleware
# SessionMiddleware
# GZipMiddleware
# CORSMiddleware
# TrustedHostMiddleware

Version Note: Middleware ordering behavior consistent since FastAPI 0.40.0+, Starlette 0.13.0+

99% confidence
A

FastAPI supports both synchronous (def) and asynchronous (async def) route handlers. For OAuth callbacks, the choice affects performance, I/O handling, and compatibility with OAuth libraries. Modern OAuth libraries like Authlib support async operations, making async def the recommended approach.

Key Differences:

Aspect async def def
Execution Runs in event loop (non-blocking) Runs in thread pool (blocking)
Performance Better for I/O operations Adds thread overhead
await keyword Required for async calls Not supported
OAuth Library Requires async-compatible library Works with sync libraries
Database Calls Use async drivers (asyncpg, motor) Use sync drivers (psycopg2, pymongo)
Recommended ✅ Yes (FastAPI 0.68.0+) Legacy/simple use cases

Async OAuth Callback (Recommended):

from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from authlib.integrations.starlette_client import OAuth
from starlette.middleware.sessions import SessionMiddleware
import secrets

app = FastAPI()

app.add_middleware(
    SessionMiddleware,
    secret_key=secrets.token_hex(32)
)

oauth = OAuth()
oauth.register(
    name='google',
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

# ✅ ASYNC - Recommended
@app.get("/login/google")
async def login_google(request: Request):
    redirect_uri = request.url_for('auth_google')
    # await required for async operations
    return await oauth.google.authorize_redirect(request, redirect_uri)

@app.get("/auth/google")
async def auth_google(request: Request):
    try:
        # await required for token exchange
        token = await oauth.google.authorize_access_token(request)
        
        # await required for userinfo fetch
        user = token.get('userinfo')
        if not user:
            user = await oauth.google.userinfo(token=token)
        
        # Async database operation
        user_id = await save_user_to_db(user)
        
        # Session storage (sync operation)
        request.session['user_id'] = user_id
        request.session['email'] = user['email']
        
        return RedirectResponse(url='/dashboard')
        
    except Exception as e:
        # Async logging
        await log_error(str(e))
        return RedirectResponse(url='/login?error=auth_failed')

async def save_user_to_db(user: dict) -> int:
    # Async database operation with asyncpg
    async with db_pool.acquire() as conn:
        user_id = await conn.fetchval(
            "INSERT INTO users (email, name) VALUES ($1, $2) "
            "ON CONFLICT (email) DO UPDATE SET name = $2 "
            "RETURNING id",
            user['email'],
            user['name']
        )
    return user_id

async def log_error(error: str):
    # Async logging to external service
    async with aiohttp.ClientSession() as session:
        await session.post('https://logging-service.com/log', json={'error': error})

Sync OAuth Callback (Legacy):

# ❌ SYNC - Not recommended for modern FastAPI apps
@app.get("/login/google")
def login_google(request: Request):  # No async
    redirect_uri = request.url_for('auth_google')
    # Authlib 1.0+ requires async, so this would need sync version
    # OR run in thread pool (performance penalty)
    return oauth.google.authorize_redirect(request, redirect_uri)

@app.get("/auth/google")
def auth_google(request: Request):  # No async
    try:
        # Blocking I/O - runs in thread pool
        token = oauth.google.authorize_access_token(request)
        user = token.get('userinfo')
        
        # Blocking database operation
        user_id = save_user_to_db_sync(user)
        
        request.session['user_id'] = user_id
        return RedirectResponse(url='/dashboard')
        
    except Exception as e:
        log_error_sync(str(e))
        return RedirectResponse(url='/login?error=auth_failed')

def save_user_to_db_sync(user: dict) -> int:
    # Sync database operation with psycopg2
    with db_connection.cursor() as cursor:
        cursor.execute(
            "INSERT INTO users (email, name) VALUES (%s, %s) "
            "ON CONFLICT (email) DO UPDATE SET name = %s "
            "RETURNING id",
            (user['email'], user['name'], user['name'])
        )
        user_id = cursor.fetchone()[0]
    return user_id

How FastAPI Handles Sync Functions:

When you define a route with def instead of async def, FastAPI automatically runs it in a thread pool:

# Behind the scenes FastAPI does:
# result = await run_in_threadpool(your_sync_function, *args)

This adds overhead:

  • Thread creation/management
  • Context switching
  • Reduced concurrency under high load

Authlib Async Support:

Authlib 1.0+ (2021+) provides native async support for Starlette/FastAPI:

from authlib.integrations.starlette_client import OAuth

# Async methods available:
await oauth.google.authorize_redirect(request, redirect_uri)
await oauth.google.authorize_access_token(request)
await oauth.google.userinfo(token=token)
await oauth.google.parse_id_token(request, token)

Mixed Async/Sync Example:

@app.get("/auth/callback")
async def callback(request: Request):
    # Async OAuth call
    token = await oauth.google.authorize_access_token(request)
    
    # Call sync function (FastAPI handles with thread pool)
    user_data = process_user_sync(token['userinfo'])
    
    # Async database save
    await save_to_db_async(user_data)
    
    return {"status": "success"}

def process_user_sync(userinfo: dict) -> dict:
    # Sync data processing (CPU-bound)
    return {
        "email": userinfo['email'].lower(),
        "name": userinfo['name'].title()
    }

Performance Comparison:

import asyncio
import time

# Async: Handles 1000 concurrent OAuth callbacks efficiently
async def async_callback():
    token = await oauth.google.authorize_access_token(request)  # Non-blocking
    user_id = await save_to_db_async(token['userinfo'])        # Non-blocking
    return user_id

# Sync: Limited by thread pool size (default: 40 threads)
def sync_callback():
    token = oauth.google.authorize_access_token(request)  # Blocks thread
    user_id = save_to_db_sync(token['userinfo'])         # Blocks thread
    return user_id

# Under high load:
# Async: ~1000 requests/sec
# Sync: ~40 requests/sec (limited by thread pool)

When to Use Sync:

  1. Legacy OAuth Library: Library doesn't support async (migrate if possible)
  2. CPU-Bound Processing: Heavy computation without I/O
  3. Quick Prototype: Testing without async setup
  4. Simple Application: Low traffic, simplicity preferred

Best Practices:

  1. ✅ Use async def for all I/O operations (database, HTTP, file access)
  2. ✅ Use async OAuth libraries (Authlib 1.0+)
  3. ✅ Use async database drivers (asyncpg, motor, databases)
  4. ✅ Use await for all async function calls
  5. ❌ Don't mix await with def (syntax error)
  6. ❌ Don't use blocking I/O in async def (blocks event loop)

Version Note: Async OAuth support in Authlib 1.0+ (2021), FastAPI async/await since 0.1.0

99% confidence
A

When both FastAPI's CORSMiddleware and nginx CORS headers are configured, they both add Access-Control-* headers to responses, resulting in duplicate headers. Modern browsers reject responses with duplicate Access-Control-Allow-Origin headers, causing CORS failures even though both layers intend to allow the request.

The Problem: Duplicate Headers

# Response with BOTH FastAPI and nginx CORS:
HTTP/1.1 200 OK
Access-Control-Allow-Origin: https://example.com  ← From FastAPI
Access-Control-Allow-Origin: https://example.com  ← From nginx
Access-Control-Allow-Methods: GET, POST, PUT, DELETE
Access-Control-Allow-Methods: GET, POST, PUT, DELETE

# Browser console error:
"The 'Access-Control-Allow-Origin' header contains multiple values 'https://example.com, https://example.com', but only one is allowed."

Browser Behavior:

  • Chrome/Edge: Rejects with CORS error
  • Firefox: Rejects with CORS error
  • Safari: Rejects with CORS error
  • Result: Frontend cannot access API even though CORS is "configured"

Solution: Choose ONE Layer

Option 1: nginx CORS Only (Recommended for Production)

# /etc/nginx/sites-available/your-site
server {
    listen 443 ssl;
    server_name api.example.com;

    location / {
        # CORS headers in nginx
        add_header 'Access-Control-Allow-Origin' 'https://example.com' always;
        add_header 'Access-Control-Allow-Methods' 'GET, POST, PUT, DELETE, OPTIONS' always;
        add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always;
        add_header 'Access-Control-Allow-Credentials' 'true' always;
        add_header 'Access-Control-Max-Age' '3600' always;

        # Handle preflight requests
        if ($request_method = 'OPTIONS') {
            return 204;
        }

        # Proxy to FastAPI
        proxy_pass http://localhost:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}
# FastAPI app - DISABLE CORSMiddleware
from fastapi import FastAPI

app = FastAPI()

# ❌ REMOVE or COMMENT OUT CORSMiddleware
# from fastapi.middleware.cors import CORSMiddleware
# app.add_middleware(
#     CORSMiddleware,
#     allow_origins=["https://example.com"],
#     allow_credentials=True,
#     allow_methods=["*"],
#     allow_headers=["*"],
# )

# ✅ Let nginx handle CORS

@app.get("/api/data")
async def get_data():
    return {"data": "value"}

Option 2: FastAPI CORS Only (Development/Simple Deployments)

# nginx - NO CORS headers
server {
    listen 443 ssl;
    server_name api.example.com;

    location / {
        # ❌ NO CORS headers in nginx
        
        # Just proxy to FastAPI
        proxy_pass http://localhost:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}
# FastAPI app - ENABLE CORSMiddleware
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# ✅ Handle CORS in FastAPI
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # Specific origin
    allow_credentials=True,
    allow_methods=["*"],  # Or specific: ["GET", "POST", "PUT", "DELETE"]
    allow_headers=["*"],  # Or specific: ["Content-Type", "Authorization"]
)

@app.get("/api/data")
async def get_data():
    return {"data": "value"}

Why nginx CORS is Recommended for Production:

  1. Performance: nginx handles preflight OPTIONS requests without hitting FastAPI
  2. Centralized Configuration: All infrastructure config in one place
  3. Security: Can apply rate limiting, IP filtering at nginx layer
  4. Flexibility: Can route to multiple backends with different CORS policies
  5. Caching: nginx can cache preflight responses

Debugging Duplicate Headers:

# Check response headers
curl -I -X OPTIONS https://api.example.com/api/data \
  -H "Origin: https://example.com" \
  -H "Access-Control-Request-Method: POST"

# Look for duplicate headers:
# Access-Control-Allow-Origin: https://example.com
# Access-Control-Allow-Origin: https://example.com  ← DUPLICATE!

# Count occurrences
curl -I -X OPTIONS https://api.example.com/api/data \
  -H "Origin: https://example.com" \
  -H "Access-Control-Request-Method: POST" 2>&1 | \
  grep -c "Access-Control-Allow-Origin"

# Output: 2 (means duplicate, should be 1)

Common Mistake: Environment-Specific Config

# ❌ BAD: Enables CORS in all environments
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(CORSMiddleware, ...)  # Always enabled

# ✅ GOOD: Only enable in development
import os

if os.getenv("ENVIRONMENT") == "development":
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["http://localhost:3000"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
# Production: nginx handles CORS

Testing CORS Configuration:

// Frontend test
fetch('https://api.example.com/api/data', {
  method: 'GET',
  headers: {
    'Content-Type': 'application/json',
  },
  credentials: 'include', // For cookies/auth
})
  .then(response => response.json())
  .then(data => console.log('Success:', data))
  .catch(error => console.error('CORS Error:', error));

// Check browser console:
// ✅ Success: No CORS errors
// ❌ Failure: "Access-Control-Allow-Origin header contains multiple values"

Complete nginx CORS Configuration:

server {
    listen 443 ssl;
    server_name api.example.com;

    location / {
        # CORS configuration
        set $cors '';
        if ($http_origin = 'https://example.com') {
            set $cors 'true';
        }

        if ($cors = 'true') {
            add_header 'Access-Control-Allow-Origin' '$http_origin' always;
            add_header 'Access-Control-Allow-Credentials' 'true' always;
            add_header 'Access-Control-Allow-Methods' 'GET, POST, PUT, DELETE, OPTIONS' always;
            add_header 'Access-Control-Allow-Headers' 'Accept,Authorization,Cache-Control,Content-Type,DNT,If-Modified-Since,Keep-Alive,Origin,User-Agent,X-Requested-With' always;
        }

        # Preflight requests
        if ($request_method = 'OPTIONS') {
            add_header 'Access-Control-Allow-Origin' '$http_origin' always;
            add_header 'Access-Control-Allow-Credentials' 'true' always;
            add_header 'Access-Control-Allow-Methods' 'GET, POST, PUT, DELETE, OPTIONS' always;
            add_header 'Access-Control-Allow-Headers' 'Accept,Authorization,Cache-Control,Content-Type,DNT,If-Modified-Since,Keep-Alive,Origin,User-Agent,X-Requested-With' always;
            add_header 'Access-Control-Max-Age' 3600;
            add_header 'Content-Type' 'text/plain; charset=utf-8';
            add_header 'Content-Length' 0;
            return 204;
        }

        proxy_pass http://localhost:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Version Note: Applies to FastAPI 0.68.0+, nginx 1.18.0+. CORS duplicate header rejection is standard browser behavior (CORS spec)

99% confidence