Database migrations using Alembic and connection pooling are essential practices for managing schema changes and maintaining performance in modern applications.
Alembic is a database migration tool commonly used with SQLAlchemy that helps track and apply incremental changes to the database schema over time. Connection pooling, on the other hand, manages a set of reusable database connections, reducing the overhead of creating and closing connections for each request and improving application efficiency.
Understanding Authentication and Authorization Basics
Before implementing advanced protocols, grasp the foundational difference between authentication (verifying who you are) and authorization (determining what you can do). FastAPI abstracts these via fastapi.security, making secure APIs intuitive without boilerplate.
In full-stack development, stateless auth like JWT shines for microservices and SPAs, as servers don't store sessions—tokens carry all needed info. This reduces database load and scales effortlessly.
Key Concepts: JWT vs. Traditional Sessions
JWT is a compact, self-contained token encoding user claims (e.g., ID, roles) in a signed JSON structure: header.payload.signature. Unlike cookies or sessions, it's tamper-proof via digital signatures like HS256 or RS256.
1. Pros of JWT: Stateless, scalable, works across domains (CORS-friendly).
2. Cons: Larger payloads; revocation requires blacklisting (mitigated with short expiry + refresh tokens).
3. When to use: APIs with mobile/SPA clients.
JWT VS Sessions
Setting Up FastAPI with Security Dependencies
FastAPI's OAuth2PasswordBearer and Depends make security declarative. Start by installing essentials: pip install python-jose[cryptography] passlib[bcrypt] python-multipart.
Create a secure app skeleton with Pydantic models for users/tokens. This setup uses bcrypt for password hashing, aligning with OWASP best practices.
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "your-secret-key" # Use env vars in prod
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30Implementing JWT-Based Authentication Flow
JWT auth follows a standard flow: login → token issuance → protected endpoint validation. FastAPI handles token extraction via Depends(oauth2_scheme).
This pattern secures your full-stack API, where frontend sends tokens in Authorization: Bearer <token> headers.
Step 1: User Model and Password Utilities
Define models first
1. Create User and Token Pydantic schemas.
2. Implement verify_password and get_password_hash using bcrypt.
3. Fake a user DB for demo (replace with SQLAlchemy/Postgres in production).
class User(BaseModel):
username: str
email: str | None = None
full_name: str | None = None
disabled: bool = False
class Token(BaseModel):
access_token: str
token_type: str
fake_users_db = {"user": {"username": "user", "full_name": "User", "email": "user@example.com", "hashed_password": pwd_context.hash("secret"), "disabled": False}}Step 2: Token Creation and Verification
Generate tokens with claims:
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(username) # Fetch from DB
if user is None:
raise credentials_exception
return userStep 3: Login Endpoint
Handle login with OAuth2PasswordRequestForm:
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = get_user(form_data.username)
if not user or not verify_password(form_data.password, user.get("hashed_password")):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect credentials")
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)
return {"access_token": access_token, "token_type": "bearer"}Integrating OAuth2 for Advanced Authorization
OAuth2 extends JWT with scopes and grant types (e.g., password, client_credentials). FastAPI's OAuth2PasswordBearer supports scopes for fine-grained access, per RFC 6749.
Use this for APIs needing role-based access, like admin vs. user endpoints in full-stack dashboards.
Defining Scopes and Protected Routes
Scopes are space-separated strings (e.g., "user:read profile:write")
from fastapi.security.utils import get_authorization_scheme_param
class TokenData(BaseModel):
username: str | None = None
scopes: list[str] = []
def verify_token_scopes(token_scopes: list[str], required_scopes: list[str]):
if set(required_scopes).issubset(set(token_scopes)):
return True
raise HTTPException(status_code=403, detail="Insufficient scopes")
# Enhanced get_current_user with scopes
async def get_current_active_user(token: str = Depends(oauth2_scheme), scopes: list[str] = ["user:read"]):
user = await get_current_user(token)
token_data = TokenData(scopes=token.split(" ")[1].split(".")[1]) # Simplified; parse properly
verify_token_scopes(token_data.scopes, scopes)
if user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return user
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
@app.get("/admin/users/", response_model=list[User])
async def read_users(current_user: User = Depends(get_current_active_user(scopes=["admin:read"]))):
return list(fake_users_db.values())Best Practices and Common Pitfalls
Secure implementations follow OWASP API Security Top 10. Always use HTTPS, short-lived tokens (15-60 min), and refresh tokens stored securely (HttpOnly cookies).
1. Do: Rotate SECRET_KEY regularly; use JWK for RS256 in distributed systems.
2. Don't: Store sensitive claims in JWT (use opaque tokens + introspection).
3. Pitfalls: Clock skew (add 5s leeway in jwt.decode); no token revocation (use Redis blacklist).
Example refresh endpoint
1. Client sends refresh token.
2. Validate and issue new access token.
3. Rotate refresh token to prevent replay.
For production, integrate Auth0 or Keycloak for managed OAuth2/JWT.
Hands-On: Complete Example and Testing
Combine everything in a runnable app. Test with curl or Postman
# Login
curl -X POST "http://localhost:8000/token" -d "username=user&password=secret"
# Protected
curl -H "Authorization: Bearer <token>" http://localhost:8000/users/meDeploy with Uvicorn + Gunicorn; add CORS middleware for frontend integration.
