29.2 后端接口：RESTful API、权限认证、数据库 CRUD

这是 Python 在后端开发中最常见的组合需求：搭建一套符合 REST 规范的数据接口，实现增删改查，并保护接口不被随意调用。

下面以一个典型的图书管理接口为例，用 FastAPI 快速说明核心实现思路（Django/Flask 原理相通，只是写法不同）。

RESTful API 设计原则

资源导向：把业务对象映射为 URL 路径，如 /books 表示图书集合，/books/{id} 表示单本书。
HTTP 方法对应操作：
GET —— 查询（安全、幂等）
POST —— 创建（非幂等）
PUT 或 PATCH —— 全量更新 / 部分更新
DELETE —— 删除
状态码规范：200 成功、201 创建成功、400 请求错误、404 不存在、500 服务器错误。
数据格式：请求体和响应体通常使用 JSON。

最小化示例（FastAPI + SQLAlchemy）：

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base

# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./books.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

# 数据库模型
class Book(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    author = Column(String)

# Pydantic 模型（请求/响应）
class BookCreate(BaseModel):
    title: str
    author: str

class BookResponse(BookCreate):
    id: int
    class Config:
        orm_mode = True

app = FastAPI()

# 数据库依赖注入
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# CRUD 接口
@app.post("/books", response_model=BookResponse, status_code=201)
async def create_book(book: BookCreate, db: Session = Depends(get_db)):
    db_book = Book(**book.dict())
    db.add(db_book)
    db.commit()
    db.refresh(db_book)
    return db_book

@app.get("/books", response_model=list[BookResponse])
async def list_books(db: Session = Depends(get_db)):
    return db.query(Book).all()

@app.get("/books/{book_id}", response_model=BookResponse)
async def get_book(book_id: int, db: Session = Depends(get_db)):
    book = db.query(Book).filter(Book.id == book_id).first()
    if not book:
        raise HTTPException(status_code=404, detail="Book not found")
    return book

@app.put("/books/{book_id}", response_model=BookResponse)
async def update_book(book_id: int, book: BookCreate, db: Session = Depends(get_db)):
    db_book = db.query(Book).filter(Book.id == book_id).first()
    if not db_book:
        raise HTTPException(status_code=404, detail="Book not found")
    db_book.title = book.title
    db_book.author = book.author
    db.commit()
    db.refresh(db_book)
    return db_book

@app.delete("/books/{book_id}", status_code=204)
async def delete_book(book_id: int, db: Session = Depends(get_db)):
    db_book = db.query(Book).filter(Book.id == book_id).first()
    if not db_book:
        raise HTTPException(status_code=404, detail="Book not found")
    db.delete(db_book)
    db.commit()
    return

权限认证

真实业务中，大部分接口需要验证用户身份，常见方案是 JWT（JSON Web Token）。

核心流程：

用户登录时，服务器验证账号密码，生成一个签名过的 JWT（包含用户 ID、过期时间）。
客户端后续请求时，在 Authorization 头中携带 Bearer <token>。
服务端通过依赖注入解析 Token，获取用户信息，再校验权限。

FastAPI 示例（基于 python-jose 和 passlib）：

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# 生成 Token
def create_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)

# 获取当前用户依赖
def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401)
    except JWTError:
        raise HTTPException(status_code=401)
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=401)
    return user

在需要保护的接口中，添加 current_user: User = Depends(get_current_user) 即可。
对于不同角色（管理员、普通用户），可在依赖中继续判断权限，或使用装饰器简化。

数据库 CRUD 实战要点

ORM 选型：小型项目用 SQLAlchemy 或 Peewee，大型项目用 SQLAlchemy（配合 Alembic 管理数据库迁移）。
避免 SQL 注入：始终使用 ORM 的参数绑定，绝不拼接原始 SQL 字符串。
连接池：生产环境务必配置连接池（SQLAlchemy 默认内置），避免频繁创建/销毁连接。
异步支持：FastAPI 原生异步，若数据库操作需异步，可使用 databases + sqlalchemy 异步驱动，或直接使用 Tortoise-ORM。
分页与过滤：不要一次性返回全表数据，实现 skip/limit 分页，并提供过滤参数（如 ?author=Martin）。

生产环境补充

接口文档：FastAPI 自动生成交互式文档（/docs），极大方便前后端联调。
跨域配置：前端分离部署时，需要通过 CORS 中间件允许前端域名访问。
限流与监控：使用 slowapi 等实现接口限流，避免恶意调用。

这套组合已经能撑起大部分中小型项目的后端接口需求，思路清晰后，你可以随时查阅具体框架的文档来补全细节。