FastAPI 用戶認證：OAuth2 + JWT 登錄全解析，一站式入門！
作者：Ss肥魚
本文將帶你一步步實現 用戶認證 + JWT 登錄機制，讓你的 API 擁有完整的登錄保護！
你是否也想在 FastAPI 中實現登錄、認證、權限驗證？
本文將帶你一步步實現 用戶認證 + JWT 登錄機制，讓你的 API 擁有完整的登錄保護！
為什么使用 JWT？
JWT（JSON Web Token）是一種輕量級認證方式，支持：
- 前后端分離（令牌無狀態存儲）
- 認證效率高，適合微服務
- 易于擴展與集成
FastAPI 內置了 OAuth2PasswordBearer 支持，結合 pyjwt 模塊，可以輕松實現。
安裝依賴
pip install python-jose[cryptography] passlib[bcrypt] python-multipart項目結構（新增 auth 模塊）
app/
├── api/
│ └── user.py # 用戶接口（注冊、獲取信息）
├── auth/
│ ├── auth.py # 登錄 & 令牌生成
│ └── deps.py # 獲取當前用戶
├── core/
│ └── security.py # 加密與解密工具
├── models/
│ └── user.py # 用戶模型
├── schemas/
│ └── user.py # 請求 & 響應模型
├── main.py創建用戶模型（models/user.py）
from tortoise.models import Model
from tortoise import fields
class User(Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=50, unique=True)
hashed_password = fields.CharField(max_length=128)
class Meta:
table = "user"密碼加密工具（core/security.py）
from passlib.context import CryptContext
from jose import jwt
from datetime import datetime, timedelta
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
defverify_password(plain_pwd, hashed_pwd):
return pwd_context.verify(plain_pwd, hashed_pwd)
defhash_password(password):
return pwd_context.hash(password)
defcreate_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)用戶注冊 & 登錄（api/user.py）
from fastapi import APIRouter, HTTPException
from app.models.user import User
from app.schemas.user import UserCreate, UserOut
from app.core.security import hash_password
from app.auth.auth import authenticate_user, create_token_response
router = APIRouter(prefix="/users", tags=["用戶"])
@router.post("/register", response_model=UserOut)
asyncdefregister(user: UserCreate):
ifawait User.get_or_none(username=user.username):
raise HTTPException(status_code=400, detail="用戶名已存在")
user_obj = await User.create(
username=user.username,
hashed_password=hash_password(user.password)
)
return user_obj
@router.post("/login")
asyncdeflogin(form_data: OAuth2PasswordRequestForm = Depends()):
user = await authenticate_user(form_data.username, form_data.password)
ifnot user:
raise HTTPException(status_code=401, detail="用戶名或密碼錯誤")
return create_token_response(user.username)登錄邏輯與令牌生成（auth/auth.py）
from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends
from app.models.user import User
from app.core.security import verify_password, create_access_token
from datetime import timedelta
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/login")
asyncdefauthenticate_user(username: str, password: str):
user = await User.get_or_none(username=username)
if user and verify_password(password, user.hashed_password):
return user
returnNone
defcreate_token_response(username: str):
access_token_expires = timedelta(minutes=30)
access_token = create_access_token(
data={"sub": username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}獲取當前用戶（auth/deps.py）
from fastapi import Depends, HTTPException
from jose import JWTError, jwt
from app.auth.auth import oauth2_scheme
from app.core.security import SECRET_KEY, ALGORITHM
from app.models.user import User
asyncdefget_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("sub")
ifnot username:
raise HTTPException(status_code=401, detail="認證失敗")
except JWTError:
raise HTTPException(status_code=401, detail="令牌無效")
user = await User.get_or_none(username=username)
ifnot user:
raise HTTPException(status_code=404, detail="用戶不存在")
return user請求 & 響應模型（schemas/user.py）
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
password: str
class UserOut(BaseModel):
id: int
username: str
class Config:
orm_mode = True路由使用驗證用戶（示例）
from app.auth.deps import get_current_user
@router.get("/me", response_model=UserOut)
async def get_me(current_user=Depends(get_current_user)):
return current_user運行 & 測試登錄流程
uvicorn app.main:app --reload- 訪問 /docs
- 使用 /users/register 創建用戶
- 使用 /users/login 獲取 JWT
- 復制令牌，在右上角 Authorize 中粘貼
- 測試 /users/me 接口獲取當前用戶信息
總結
通過本文你學會了：
- JWT 的核心機制
- 用戶注冊 + 登錄接口設計
- 認證用戶的依賴注入
- 保護接口，防止未授權訪問
責任編輯：趙寧寧
來源：
Ssoul肥魚
