06. 开发后端 JWT 用户认证

POST /api/auth/login
        ↓
验证用户名 + 密码
        ↓
生成 JWT Token
        ↓
返回 Token + 用户信息
        ↓
前端保存 Token
        ↓
请求其他接口时携带 Token
        ↓
后端解析 Token
        ↓
获取当前登录用户

{
  "python.analysis.autoImportCompletions": true,
  "python.languageServer": "Pylance",
  "python.analysis.extraPaths": ["./backend"],
  "editor.quickSuggestions": {
    "other": true,
    "comments": false,
    "strings": false
  },
  "[python]": {
    "editor.codeActionsOnSave": {
      "source.organizeImports": "explicit"
    },
    "editor.defaultFormatter": "ms-python.black-formatter"
  },
}

bcrypt==5.0.0
PyJWT==2.13.0

JWT_SECRET_KEY=sjQ9crUw71awph8ORnvERoUtMQ19qelECFjj06i1urz
JWT_EXPIRE_HOURS=24
JWT_ALGORITHM="HS256"

class Settings(BaseSettings):
    DATABASE_URL: str

    JWT_SECRET_KEY: str
    JWT_ALGORITHM: str = "HS256"
    JWT_EXPIRE_HOURS: int = 24

from datetime import datetime, timedelta
import jwt
from app.config import settings

def create_access_token(user_id: int) -> str:
    """创建JWT token"""
    expire = datetime.now() + timedelta(hours=settings.JWT_EXPIRE_HOURS)

    payload = {"user_id": user_id, "exp": expire}  # exp是固定key，不能写成别的

    return jwt.encode(
        payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM
    )

def decode_access_token(token: str) -> dict:
    """解析jwt token"""
    return jwt.decode(
        token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]
    )

import bcrypt

def hash_password(password: str) -> str:
    """对密码加密"""
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")

def verify_password(password: str, hashed_password: str) -> bool:
    """验证密码的有效性"""
    """  "123" -> "xxsdasdasdasfa"  """
    return bcrypt.checkpw(password.encode("utf-8"), hashed_password.encode("utf-8"))

from fastapi import APIRouter, Depends
from app.schemas.auth import LoginRequest
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.user import User
from app.schemas.user import UserResponse
from app.utils.password import verify_password
from app.utils.jwt import create_access_token

router = APIRouter(prefix="/auth", tags=["权限验证"])

@router.post("/login")
def login(data: LoginRequest, db: Session = Depends(get_db)):
    # 根据用户账号查询数据库
    user = db.query(User).filter(User.username == data.username).first()

    # 判断账号和密码是否正确
    if not user or not verify_password(data.password, user.password):
        return {"code": 400, "message": "账号或密码错误"}
    # 验证账号的状态
    if user.status != 1:
        return {"code": 403, "message": "账号已被禁用"}
    # 创建token
    token = create_access_token(user.id)
    # 返回信息
    return {
        "code": 200,
        "message": "操作成功",
        "data": {"token": token, "user": UserResponse.model_validate(user)},
    }

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.database import get_db
from app.models.user import User
from app.utils.jwt import decode_access_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/autrh/login")

def get_current_user(
    token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)
) -> User:
    """JWT token 鉴权验证合法性"""
    try:
        payload = decode_access_token(token)
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="登录已失效，请重新登录"
        )
    # 获取到用户ID
    user_id = payload.get("user_id")
    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的登录凭证"
        )
    # 从数据库根据用户ID查询用户信息
    user = db.query(User).filter(User.id == user_id).first()
    if user.status != 1:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="用户被禁用"
        )
    return user

from fastapi import APIRouter, Depends
from app.models.user import User
from app.dependencies.auth import get_current_user
from app.schemas.user import UserResponse

router = APIRouter(prefix="/user", tags=["用户信息接口"])

@router.get("/info")
def get_user_info(current_user: User = Depends(get_current_user)):
    """获取当前登录用户信息"""
    return {
        "code": 200,
        "message": "请求成功",
        "data": UserResponse.model_validate(current_user),
    }