53｜后端接口实现（上）

模型和迁移就绪后，开始写路由层。上半部分实现用户认证和资产管理的 CRUD 接口，覆盖登录、Token 颁发、资产列表查询、创建、详情、更新、删除。

一、用户认证接口

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from passlib.context import CryptContext
import jwt
from datetime import datetime, timedelta

from app.database import get_db
from app.models import User
from app.config import settings

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

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/login")

def create_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=settings.token_expire_minutes)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.secret_key, algorithm="HS256")

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
        user_id = payload.get("sub")
        if not user_id:
            raise HTTPException(status_code=401, detail="无效凭证")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="凭证无效或已过期")

    user = db.get(User, int(user_id))
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db),
):
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not pwd_context.verify(form_data.password, user.password_hash):
        raise HTTPException(status_code=401, detail="用户名或密码错误")

    token = create_token({"sub": str(user.id)})
    return {"access_token": token, "token_type": "bearer"}

@router.get("/me")
async def read_me(user: User = Depends(get_current_user)):
    return {
        "id": user.id,
        "username": user.username,
        "role": user.role,
    }

二、资产管理接口

# routers/assets.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from sqlalchemy import select, func
from typing import List

from app.database import get_db
from app.models import Asset
from app.schemas import AssetCreate, AssetUpdate, AssetOut
from app.routers.users import get_current_user, require_admin

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

@router.get("/", response_model=List[AssetOut])
async def list_assets(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    status: str | None = None,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
):
    stmt = select(Asset)

    if user.role != "admin":
        stmt = stmt.where(Asset.owner_id == user.id)
    if status:
        stmt = stmt.where(Asset.status == status)

    total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar()
    items = db.execute(stmt.offset(skip).limit(limit)).scalars().all()
    return items

@router.post("/", response_model=AssetOut, status_code=201)
async def create_asset(
    data: AssetCreate,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
):
    asset = Asset(**data.dict(), owner_id=user.id)
    db.add(asset)
    db.commit()
    db.refresh(asset)
    return asset

@router.get("/{asset_id}", response_model=AssetOut)
async def get_asset(
    asset_id: int,
    db: Session = Depends(get_db),
    user: User = Depends(get_current_user),
):
    asset = db.get(Asset, asset_id)
    if not asset:
        raise HTTPException(status_code=404, detail="资产不存在")
    if user.role != "admin" and asset.owner_id != user.id:
        raise HTTPException(status_code=403, detail="无权限")
    return asset

三、权限依赖

# routers/users.py

def require_admin(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

四、注册路由

# app/main.py
from fastapi import FastAPI
from app.routers import users, assets

app = FastAPI()
app.include_router(users.router)
app.include_router(assets.router)

五、常见错误

Token 过期时间不设置

# 错误：永久有效 Token
jwt.encode({"sub": "1"}, SECRET_KEY, algorithm="HS256")

# 正确：设置 exp
jwt.encode({"sub": "1", "exp": datetime.utcnow() + timedelta(minutes=30)}, ...)

查询资产时没做权限过滤

# 错误：普通用户能看到所有资产
stmt = select(Asset)

# 正确：非管理员只能看自己的
if user.role != "admin":
    stmt = stmt.where(Asset.owner_id == user.id)

创建资产后没 refresh

# 错误：asset.id 是 None
db.add(asset)
db.commit()
return asset   # id 还没从数据库获取

# 正确
db.commit()
db.refresh(asset)
return asset