FastAPI leverages the OAuth2PasswordBearer security scheme to enforce bearer token authentication following the OAuth2 password grant flow. This approach requires configuring a dependency that extracts tokens from the Authorization header, validating credentials at a dedicated login route, and issuing JSON Web Tokens (JWTs) via third-party libraries like python-jose and cryptography.

Token Extraction Setup

Initialize the security scheme by specfiying the endpoint responsible for token issuance:

from fastapi.security import OAuth2PasswordBearer

token_handler = OAuth2PasswordBearer(tokenUrl="/auth/session")

Inject this scheme into protected routes using FastAPI's dependency injection system. The framework will automatically parse the Bearer <token> header and pass the extracted string to your handler:

@app.get("/secure-resource")
async def fetch_data(credentials: str = Depends(token_handler)):
    return {"status": "success", "received_token_prefix": credentials[:8] + "..."}

Credential Verification & Token Issuance

The authentication endpoint must accept form data containing username and password. FastAPI provides OAuth2PasswordRequestForm to parse these fields compliant with the OAuth2 specification. Always install python-multipart to handle multipart/form-data requests:

pip install python-multipart passlib[bcrypt] python-jose[cryptography]

Password hashing should occur securely. The passlib library paired with bcrypt handles salting and verification efficiently. During registration or setup, hash passwords before storage. At login, verify plaintext inputs against stored hashes:

from passlib.context import CryptContext

pwd_cipher = CryptContext(schemes=["bcrypt"], deprecated="auto")

def validate_credentials(plaintext: str, ciphertext: str) -> bool:
    return pwd_cipher.verify(plaintext, ciphertext)

Successful authentication triggers JWT creation. Utilize jwt.encode() from python-jose to sign payloads with a secret key and expiration timestamp:

import jwt as jose_encode
from datetime import datetime, timedelta, timezone

SECRET_KEY = "a3f8b9c2d1e0f7a6b5c4d3e2f1a0b9c8d7e6f5a4b3c2d1e0"
ALGORITHM = "HS256"
ACCESS_TOKEN_TTL = timedelta(minutes=45)

def construct_jws(user_identifier: str) -> str:
    payload = {"uid": user_identifier, "iat": datetime.now(timezone.utc)}
    payload["exp"] = datetime.now(timezone.utc) + ACCESS_TOKEN_TTL
    return jose_encode.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

Combine verification and issuance into the login route:

from fastapi import HTTPException, status

class UserProfile(BaseModel):
    username: str
    hashed_secret: str

# Placeholder for actual database logic
def fetch_account(username: str) -> UserProfile | None:
    pass

@app.post("/auth/session")
async def authenticate(form: OAuth2PasswordRequestForm = Depends()):
    acct = fetch_account(form.username)
    
    if not acct or not validate_credentials(form.password, acct.hashed_secret):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication failed",
            headers={"WWW-Authenticate": "Bearer"},
        )
        
    jws_token = construct_jws(acct.username)
    return {"access_token": jws_token, "token_type": "bearer"}

Decoding & Route Protection

Extracting the username from an incoming token requires decoding and validation. Wrap this logic in a reusable dependency to maintain clean route handlers:

from jose import JWTError, jwt as jose_decode

def resolve_identity(jwt_string: str = Depends(token_handler)) -> UserProfile:
    auth_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or malformed signature",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        claims = jose_decode.decode(jwt_string, SECRET_KEY, algorithms=[ALGORITHM])
        subject = claims.get("uid")
        if not subject:
            raise auth_exception
    except JWTError:
        raise auth_exception
        
    current_acct = fetch_account(subject)
    if not current_acct or not getattr(current_acct, "active", True):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access restricted")
        
    return current_acct

Apply this dependency to restrict access to authorized accounts only:

@app.get("/dashboard/metrics")
async def view_metrics(current: UserProfile = Depends(resolve_identity)):
    return {"owner": current.username, "view": "allowed"}

Request Processing Middleware

Interceptors execute around every request lifecycle phase. Decorate async functions with @app.middleware("http") to hook into pre-processing and post-processing stages:

import time

@app.middleware("http")
async def track_latency(request: Request, next_handler):
    begin_timestamp = time.monotonic()
    response = await next_handler(request)
    elapsed_ms = (time.monotonic() - begin_timestamp) * 1000
    response.headers["X-Duration-MS"] = f"{elapsed_ms:.2f}"
    return response

Place timing calculations before and after invoking next_handler() to measure total latency without bolcking internal logic execution.