User system — registration, login, JWT authentication
In this chapter, you will learn FastAPI's JWT authentication scheme and understand its differences from Django/Flask Session authentication.
JWT vs Session Authentication
Django and Flask useSession authenticationThe server stores user state, and only the Session ID is stored in the cookie.
FastAPI recommendsJWT Authentication: After the user logs in, they obtain a Token. Subsequent requests can carry the Token for authentication.
| Features | Session（Django/Flask） | JWT（FastAPI） |
|---|---|---|
| Storage location | Server-side (memory/database) | Client-side (Cookie/LocalStorage) |
| Extensibility | The server needs to share Session | Stateless, naturally supports distributed systems |
| Applicable scenarios | Server-side rendered websites | API + SPA / Mobile |
| Expiration control | The server can invalidate it at any time | Valid throughout the Token's validity period |
Install dependencies
(venv) $ pip install passlib python-jose python-multipart
- passlib: Password hashing (analogous to werkzeug.security)
- python-jose: JWT generation and verification
- python-multipart: Handle form submission (OAuth2 login form)
Define the User model
Example
# File path: models.py, new additions
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""Hash password"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""Verify password"""
return pwd_context.verify(password, self.hashed_password)
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""Hash password"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""Verify password"""
return pwd_context.verify(password, self.hashed_password)
Generate and execute migrations:
(venv) $ alembic revision --autogenerate -m "新增 User 模型" (venv) $ alembic upgrade head
JWT utility functions
Example
# File path: auth.py (new file)
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# Secret key (read from environment variables in production)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Token validity period: 24 hours
# OAuth2PasswordBearer 告诉 FastAPI: fromRequest headers Authorization: Bearer <token> Mediummention取 JWT
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""Generate JWT Token"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
"""Verify and decode JWT Token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# Secret key (read from environment variables in production)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Token validity period: 24 hours
# OAuth2PasswordBearer 告诉 FastAPI: fromRequest headers Authorization: Bearer <token> Mediummention取 JWT
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""Generate JWT Token"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
"""Verify and decode JWT Token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
Route registration
Example
# File path: routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=["User"])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""User registration"""
# Check if username and email already exist
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail="Username already in use")
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail=Email already registered)
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=["User"])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""User registration"""
# Check if username and email already exist
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail="Username already in use")
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail=Email already registered)
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
Login Route (Issue JWT)
Example
# Append to file: routers/users.py
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
Log in and get Token
Use OAuth2PasswordRequestForm: field names are username and password (form format).
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail="Invalid username or password",
headers={"WWW-Authenticate": "Bearer"}
)
# Issue JWT: contains only user_id (must not contain sensitive information)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
Log in and get Token
Use OAuth2PasswordRequestForm: field names are username and password (form format).
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail="Invalid username or password",
headers={"WWW-Authenticate": "Bearer"}
)
# Issue JWT: contains only user_id (must not contain sensitive information)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
In JWT Token
subIt is the standard subject field, storing the user identifier. The Token itself is not encrypted (just Base64 encoded), soAbsolutely notdo not put sensitive information such as passwords in it. If you want encryption, use JWE instead of JWT.
Use JWT protection in routes
Example
# Dependency function to get current user
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Parse the current user from the JWT Token"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Token invalid or expired")
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail=“User does not exist”)
return user
# Inject into routes that require login
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""Get current user info (login required)"""
return current_user
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Parse the current user from the JWT Token"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Token invalid or expired")
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail=“User does not exist”)
return user
# Inject into routes that require login
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""Get current user info (login required)"""
return current_user
For SSR pages, JWT is usually stored in a Cookie (rather than the Authorization Header), and in template routes the Token is extracted from the Cookie for verification.
Chapter summary
In this chapter, you implemented FastAPI's JWT authentication: passlib for password hashing, python-jose for issuing and verifying JWTs, OAuth2PasswordBearer for extracting Tokens, and Depends for authentication dependency injection.
Unlike Django/Flask's Session authentication, JWT is a stateless solution, more suitable for front-end/back-end separation scenarios.
other extensions