FastAPI 用戶認(rèn)證:OAuth2 + JWT 登錄全解析,一站式入門!
作者:Ss肥魚
本文將帶你一步步實現(xiàn) 用戶認(rèn)證 + JWT 登錄機(jī)制,讓你的 API 擁有完整的登錄保護(hù)!
你是否也想在 FastAPI 中實現(xiàn)登錄、認(rèn)證、權(quán)限驗證?
本文將帶你一步步實現(xiàn) 用戶認(rèn)證 + JWT 登錄機(jī)制,讓你的 API 擁有完整的登錄保護(hù)!
為什么使用 JWT?
JWT(JSON Web Token)是一種輕量級認(rèn)證方式,支持:
- 前后端分離(令牌無狀態(tài)存儲)
- 認(rèn)證效率高,適合微服務(wù)
- 易于擴(kuò)展與集成
FastAPI 內(nèi)置了 OAuth2PasswordBearer 支持,結(jié)合 pyjwt 模塊,可以輕松實現(xiàn)。
安裝依賴
pip install python-jose[cryptography] passlib[bcrypt] python-multipart
項目結(jié)構(gòu)(新增 auth 模塊)
app/
├── api/
│ └── user.py # 用戶接口(注冊、獲取信息)
├── auth/
│ ├── auth.py # 登錄 & 令牌生成
│ └── deps.py # 獲取當(dāng)前用戶
├── core/
│ └── security.py # 加密與解密工具
├── models/
│ └── user.py # 用戶模型
├── schemas/
│ └── user.py # 請求 & 響應(yīng)模型
├── main.py
創(chuàng)建用戶模型(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"}
獲取當(dāng)前用戶(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="認(rèn)證失敗")
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
請求 & 響應(yīng)模型(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 創(chuàng)建用戶
- 使用 /users/login 獲取 JWT
- 復(fù)制令牌,在右上角 Authorize 中粘貼
- 測試 /users/me 接口獲取當(dāng)前用戶信息
總結(jié)
通過本文你學(xué)會了:
- JWT 的核心機(jī)制
- 用戶注冊 + 登錄接口設(shè)計
- 認(rèn)證用戶的依賴注入
- 保護(hù)接口,防止未授權(quán)訪問
責(zé)任編輯:趙寧寧
來源:
Ssoul肥魚