JWT 授权 FastAPI：逐步代码

FastAPI 中实现 JWT 认证：完整开发者指南

JWT（JSON Web Token）是一种用于在各方之间传输声明的紧凑令牌。令牌由三个 Base64 编码部分组成：头部、载荷和签名，用句点分隔。

头部包含签名算法（例如 HS256）和令牌类型（JWT）：

{"alg": "HS256", "typ": "JWT"}

载荷存储声明：sub（用户标识符）、iat（签发时间）、exp（过期时间）：

{"sub": "alex", "name": "Alex Tomchok", "iat": 1516239022, "exp": 1516242622}

签名计算为 HMAC-SHA256（头部.载荷 + 密钥）。服务器在验证期间检查签名：对载荷的任何更改都会使令牌失效。数据可读但受保护不被修改。

访问令牌寿命短（15–30 分钟），而刷新令牌寿命长（数天/数月）。本示例仅使用访问令牌。

项目设置与依赖

创建结构：

fastapi-jwt-auth/
├── app/
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── schemas/
│   │   ├── token.py
│   │   └── user.py
│   ├── api/
│   │   └── v1/
│   │       ├── endpoints/
│   │       │   ├── auth.py
│   │       │   └── users.py
│   │       └── dependencies.py
│   └── services/
│       └── user_service.py
└── requirements.txt

requirements.txt：

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
pydantic[email]==2.5.0

安装：pip install -r requirements.txt。

配置与 Pydantic 模式

在 core/config.py 中，使用 Pydantic 定义设置：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "your-super-secret-key-change-this-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

schemas/user.py 中的模式：

from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool = True

    class Config:
        from_attributes = True

class UserInDB(User):
    hashed_password: str

schemas/token.py：

from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: str | None = None

安全：哈希与 JWT 操作

在 core/security.py 中，实现密码哈希（bcrypt）和 JWT 函数：

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def decode_access_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

用户服务与模拟数据库

services/user_service.py 使用内存存储：

from app.schemas.user import UserInDB
from app.core.security import verify_password, get_password_hash

fake_users_db = {
    "alex": {
        "id": 1,
        "username": "alex",
        "email": "[email protected]",
        "hashed_password": get_password_hash("secret123"),
        "is_active": True,
    }
}

def get_user_by_username(username: str) -> UserInDB | None:
    if username in fake_users_db:
        return UserInDB(**fake_users_db[username])
    return None

def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user_by_username(username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

授权依赖

在 api/v1/dependencies.py 中，使用 OAuth2PasswordBearer：

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.core.security import decode_access_token
from app.services.user_service import get_user_by_username
from app.schemas.token import TokenData
from app.schemas.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    user = get_user_by_username(username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户未激活")
    return current_user

端点：登录与受保护路由

api/v1/endpoints/auth.py：

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.schemas.token import Token
from app.services.user_service import authenticate_user
from app.core.security import create_access_token

router = APIRouter(prefix="/auth", tags=["authentication"])

@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token, token_type="bearer")

api/v1/endpoints/users.py：

from fastapi import APIRouter, Depends
from app.schemas.user import User
from app.api.v1.dependencies import get_current_active_user

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

在 main.py 中连接路由器并运行 uvicorn app.main:app。

关键要点

• JWT 载荷未加密，仅签名——仅使用非敏感数据。
• 将 SECRET_KEY 存储在环境变量中，生成 32 个以上字符。
• 在生产环境中，添加刷新令牌和数据库（SQLAlchemy + Alembic）。
• FastAPI 依赖允许重用授权逻辑，无需重复代码。
• 通过 Swagger UI 测试端点：/docs。

— Editorial Team