AuthX provides middleware that can intercept and process requests at the application level, enabling features like implicit token refresh, global authentication checks, and request logging.
Available middleware
AuthX includes built-in middleware for common authentication patterns:
| Middleware | Purpose | Use case |
|---|
implicit_refresh_middleware | Auto-refresh expiring tokens | Cookie-based web apps |
| Custom middleware | Authentication checks, logging | Application-specific needs |
Implicit refresh middleware
The implicit refresh middleware automatically renews access tokens before they expire:
from fastapi import FastAPI
from authx import AuthX, AuthXConfig
from datetime import timedelta
app = FastAPI()
config = AuthXConfig(
JWT_SECRET_KEY="your-secret-key",
JWT_TOKEN_LOCATION=["cookies"],
JWT_ACCESS_TOKEN_EXPIRES=timedelta(minutes=30),
JWT_IMPLICIT_REFRESH_DELTATIME=timedelta(minutes=10),
)
auth = AuthX(config=config)
# Add middleware to application
app.middleware("http")(auth.implicit_refresh_middleware)
Custom authentication middleware
Create custom middleware for application-wide authentication logic:
from fastapi import FastAPI, Request, Response
from authx import AuthX, AuthXConfig
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable
app = FastAPI()
auth = AuthX(config=AuthXConfig(JWT_SECRET_KEY="your-secret-key"))
class AuthenticationMiddleware(BaseHTTPMiddleware):
"""Middleware to attach user information to all requests."""
async def dispatch(
self, request: Request, call_next: Callable
) -> Response:
# Skip authentication for public routes
if request.url.path in ["/", "/login", "/register", "/docs"]:
return await call_next(request)
try:
# Get and verify token
token = await auth.get_access_token_from_request(
request,
optional=True
)
if token:
payload = auth.verify_token(token, verify_csrf=False)
# Attach user info to request state
request.state.user_id = payload.sub
request.state.authenticated = True
else:
request.state.authenticated = False
except Exception:
request.state.authenticated = False
# Continue processing request
response = await call_next(request)
return response
# Add middleware
app.add_middleware(AuthenticationMiddleware)
# Use in routes
@app.get("/profile")
async def get_profile(request: Request):
if not request.state.authenticated:
raise HTTPException(status_code=401, detail="Not authenticated")
user_id = request.state.user_id
return {"user_id": user_id, "message": "Profile data"}
Request logging middleware
Log all authenticated requests:
import logging
from fastapi import Request, Response
from authx import AuthX, AuthXConfig
from datetime import datetime
from typing import Callable
logger = logging.getLogger(__name__)
auth = AuthX(config=AuthXConfig(JWT_SECRET_KEY="your-secret-key"))
async def request_logging_middleware(
request: Request,
call_next: Callable[[Request], Response]
) -> Response:
"""Log all requests with authentication info."""
start_time = datetime.now()
# Extract user info if available
user_id = "anonymous"
try:
token = await auth.get_access_token_from_request(request, optional=True)
if token:
payload = auth.verify_token(token, verify_csrf=False)
user_id = payload.sub
except Exception:
pass
# Process request
response = await call_next(request)
# Log request details
duration = (datetime.now() - start_time).total_seconds()
logger.info(
f"User: {user_id} | "
f"Method: {request.method} | "
f"Path: {request.url.path} | "
f"Status: {response.status_code} | "
f"Duration: {duration:.2f}s"
)
return response
# Add to app
app.middleware("http")(request_logging_middleware)
Token refresh with custom logic
Implement custom token refresh logic:
from fastapi import Request, Response
from authx import AuthX, AuthXConfig
from datetime import datetime, timedelta
from typing import Callable
auth = AuthX(config=AuthXConfig(
JWT_SECRET_KEY="your-secret-key",
JWT_TOKEN_LOCATION=["cookies"],
))
async def smart_refresh_middleware(
request: Request,
call_next: Callable[[Request], Response]
) -> Response:
"""Refresh tokens based on custom business logic."""
response = await call_next(request)
# Only refresh on successful requests
if response.status_code != 200:
return response
# Skip for non-authenticated routes
if request.url.path.startswith("/public"):
return response
try:
# Get current token
token = await auth.get_access_token_from_request(
request,
locations=["cookies"],
optional=True
)
if not token:
return response
payload = auth.verify_token(token, verify_csrf=False)
# Custom refresh logic
# Refresh if token expires within 15 minutes
time_until_expiry = payload.exp - datetime.now()
if time_until_expiry < timedelta(minutes=15):
# Check if user is still active in database
# (add your custom logic here)
# Create new token with extended data
new_token = auth.create_access_token(
uid=payload.sub,
fresh=False,
data={
**payload.extra_dict,
"refreshed_at": datetime.now().isoformat(),
}
)
auth.set_access_cookies(new_token, response)
except Exception as e:
# Log error but don't fail the request
logger.error(f"Token refresh failed: {e}")
return response
app.middleware("http")(smart_refresh_middleware)
CORS with authentication
Combine CORS middleware with authentication:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from authx import AuthX, AuthXConfig
app = FastAPI()
auth = AuthX(config=AuthXConfig(
JWT_SECRET_KEY="your-secret-key",
JWT_TOKEN_LOCATION=["headers", "cookies"],
JWT_COOKIE_SECURE=True,
JWT_COOKIE_SAMESITE="none", # Required for cross-origin cookies
))
# Add CORS middleware first
app.add_middleware(
CORSMiddleware,
allow_origins=["https://yourdomain.com"],
allow_credentials=True, # Required for cookies
allow_methods=["*"],
allow_headers=["*"],
)
# Then add authentication middleware
app.middleware("http")(auth.implicit_refresh_middleware)
Middleware order matters! CORS middleware should be added before authentication middleware to ensure preflight requests are handled correctly.
Middleware execution order
Request arrives
FastAPI receives the incoming HTTP request.
Middleware chain (top-down)
Middleware executes in the order they were added (first added = first executed).
Route handler
Your endpoint function executes after all middleware.
Middleware chain (bottom-up)
Response passes back through middleware in reverse order.
Response sent
Final response is sent to the client.
Router-specific middleware
Apply middleware to specific routers only:
from fastapi import APIRouter, Request, Response
from authx import AuthX, AuthXConfig
from typing import Callable
auth = AuthX(config=AuthXConfig(JWT_SECRET_KEY="your-secret-key"))
# Create router
api_router = APIRouter(prefix="/api")
async def api_auth_middleware(
request: Request,
call_next: Callable[[Request], Response]
) -> Response:
"""Middleware for API routes only."""
# Require authentication for all /api routes
token = await auth.get_access_token_from_request(request)
payload = auth.verify_token(token, verify_csrf=False)
# Add to request state
request.state.user_id = payload.sub
return await call_next(request)
# Add middleware to router
api_router.middleware("http")(api_auth_middleware)
# Define routes
@api_router.get("/users")
def get_users(request: Request):
user_id = request.state.user_id
return {"message": "Users list", "requested_by": user_id}
@api_router.post("/posts")
def create_post(request: Request):
user_id = request.state.user_id
return {"message": "Post created", "author": user_id}
# Add router to app
app.include_router(api_router)
Error handling in middleware
Handle authentication errors gracefully:
from fastapi import Request, Response
from fastapi.responses import JSONResponse
from authx import AuthX, AuthXConfig
from authx.exceptions import AuthXException, MissingTokenError, RevokedTokenError
from typing import Callable
auth = AuthX(config=AuthXConfig(JWT_SECRET_KEY="your-secret-key"))
async def auth_middleware_with_error_handling(
request: Request,
call_next: Callable[[Request], Response]
) -> Response:
"""Middleware with comprehensive error handling."""
# Public routes
if request.url.path in ["/", "/login", "/health"]:
return await call_next(request)
try:
# Try to get and verify token
token = await auth.get_access_token_from_request(request)
payload = auth.verify_token(token, verify_csrf=False)
# Check if token is in blocklist
if await auth.is_token_in_blocklist(token.token):
raise RevokedTokenError("Token has been revoked")
# Attach user to request
request.state.user_id = payload.sub
request.state.is_admin = payload.extra_dict.get("is_admin", False)
return await call_next(request)
except MissingTokenError:
return JSONResponse(
status_code=401,
content={"detail": "Authentication required"},
)
except RevokedTokenError:
return JSONResponse(
status_code=401,
content={"detail": "Token has been revoked"},
)
except AuthXException as e:
return JSONResponse(
status_code=401,
content={"detail": str(e)},
)
except Exception as e:
# Log unexpected errors
logger.error(f"Unexpected error in auth middleware: {e}")
return JSONResponse(
status_code=500,
content={"detail": "Internal server error"},
)
app.middleware("http")(auth_middleware_with_error_handling)
Middleware runs on every request. Keep authentication logic efficient by:
- Caching user lookups
- Using connection pooling for databases
- Avoiding heavy computations
- Setting appropriate token expiry times
Best practices
Order matters
Add middleware in the correct order - CORS first, then logging, then authentication.
Skip public routes
Check the request path early and skip authentication for public endpoints.
Handle errors gracefully
Don’t let authentication errors crash your application - return proper error responses.
Use request state
Store authenticated user info in request.state for easy access in route handlers.
Keep it lightweight
Middleware runs on every request - avoid expensive operations like complex database queries.
For route-specific authentication, consider using dependencies instead of middleware. Middleware is best for global concerns that apply to most or all routes.
