Python RESTful API 实战

2025-12-24 约 2807 字预计阅读 6 分钟

学习目标：使用 FastAPI 开发一个完整的 RESTful API 服务，掌握路由设计、数据校验、认证授权与自动文档。

1. RESTful 设计原则

1.1 核心概念

REST = Representational State Transfer（表现层状态转化）

资源（Resource）→ 用 URI 标识，如 /users/1
操作（Action）  → 用 HTTP 方法表达

GET    /users      → 获取用户列表
GET    /users/1    → 获取单个用户
POST   /users      → 创建用户
PUT    /users/1    → 更新用户（全量）
PATCH  /users/1    → 更新用户（部分）
DELETE /users/1    → 删除用户

1.2 HTTP 状态码

状态码	含义	场景
200	OK	请求成功
201	Created	资源创建成功
204	No Content	删除成功
400	Bad Request	参数错误
401	Unauthorized	未认证
403	Forbidden	无权限
404	Not Found	资源不存在
422	Unprocessable Entity	数据校验失败
500	Internal Server Error	服务器错误

2. 项目搭建

2.1 目录结构

api_project/
├── main.py            # 入口
├── config.py          # 配置
├── database.py        # 数据库连接
├── models.py          # 数据模型
├── schemas.py         # 请求/响应模型
├── auth.py            # 认证逻辑
├── routers/
│   ├── __init__.py
│   ├── users.py       # 用户路由
│   └── items.py       # 商品路由
└── requirements.txt

2.2 安装依赖

        
pip install fastapi uvicorn[standard] sqlalchemy pydantic python-jose[cryptography] passlib[bcrypt]

        
# 启动开发服务器
uvicorn main:app --reload --port 8000

3. 数据库与模型

3.1 数据库连接

        
        
        
    
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite 专用
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


def get_db():
    """数据库会话依赖"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3.2 数据模型

        
        
        
    
# models.py
from sqlalchemy import Column, Integer, String, Float, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    hashed_password = Column(String(200), nullable=False)
    is_active = Column(Boolean, default=True)
    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    description = Column(String(500))
    price = Column(Float, nullable=False)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="items")

3.3 Pydantic 模型

        
        
        
    
# schemas.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional


# ---- 请求模型 ----
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=6)


class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = None


class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = None
    price: float = Field(..., gt=0)


class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = Field(None, gt=0)


# ---- 响应模型 ----
class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool

    class Config:
        from_attributes = True  # ORM 模式


class ItemResponse(BaseModel):
    id: int
    name: str
    description: Optional[str]
    price: float
    owner_id: int

    class Config:
        from_attributes = True


class Token(BaseModel):
    access_token: str
    token_type: str

4. 认证授权

        
        
        
    
# auth.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from database import get_db
import models

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

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


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    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),
) -> models.User:
    """获取当前认证用户"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = db.query(models.User).filter(models.User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

5. 路由实现

5.1 认证路由

        
        
        
    
# routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from database import get_db
import models, schemas, auth

router = APIRouter(tags=["认证"])


@router.post("/register", response_model=schemas.UserResponse, status_code=201)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
    """注册用户"""
    if db.query(models.User).filter(models.User.username == user.username).first():
        raise HTTPException(status_code=400, detail="用户名已存在")
    if db.query(models.User).filter(models.User.email == user.email).first():
        raise HTTPException(status_code=400, detail="邮箱已注册")

    db_user = models.User(
        username=user.username,
        email=user.email,
        hashed_password=auth.hash_password(user.password),
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


@router.post("/token", response_model=schemas.Token)
def login(
    form: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db),
):
    """登录获取 Token"""
    user = db.query(models.User).filter(
        models.User.username == form.username
    ).first()
    if not user or not auth.verify_password(form.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )
    access_token = auth.create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

5.2 用户路由

        
        
        
    
# routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from database import get_db
import models, schemas, auth

router = APIRouter(prefix="/users", tags=["用户"])


@router.get("/", response_model=List[schemas.UserResponse])
def list_users(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    """获取用户列表"""
    return db.query(models.User).offset(skip).limit(limit).all()


@router.get("/{user_id}", response_model=schemas.UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    """获取单个用户"""
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user


@router.put("/me", response_model=schemas.UserResponse)
def update_profile(
    user_update: schemas.UserUpdate,
    current_user: models.User = Depends(auth.get_current_user),
    db: Session = Depends(get_db),
):
    """更新个人信息"""
    if user_update.email:
        current_user.email = user_update.email
    if user_update.is_active is not None:
        current_user.is_active = user_update.is_active
    db.commit()
    db.refresh(current_user)
    return current_user


@router.delete("/me", status_code=204)
def delete_account(
    current_user: models.User = Depends(auth.get_current_user),
    db: Session = Depends(get_db),
):
    """注销账户"""
    db.delete(current_user)
    db.commit()

5.3 商品路由（含权限控制）

        
        
        
    
# routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from database import get_db
import models, schemas, auth

router = APIRouter(prefix="/items", tags=["商品"])


@router.get("/", response_model=List[schemas.ItemResponse])
def list_items(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    """获取商品列表（公开）"""
    return db.query(models.Item).offset(skip).limit(limit).all()


@router.post("/", response_model=schemas.ItemResponse, status_code=201)
def create_item(
    item: schemas.ItemCreate,
    current_user: models.User = Depends(auth.get_current_user),
    db: Session = Depends(get_db),
):
    """创建商品（需登录）"""
    db_item = models.Item(**item.model_dump(), owner_id=current_user.id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


@router.put("/{item_id}", response_model=schemas.ItemResponse)
def update_item(
    item_id: int,
    item_update: schemas.ItemUpdate,
    current_user: models.User = Depends(auth.get_current_user),
    db: Session = Depends(get_db),
):
    """更新商品（仅拥有者）"""
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="商品不存在")
    if item.owner_id != current_user.id:
        raise HTTPException(status_code=403, detail="无权修改他人商品")

    update_data = item_update.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(item, field, value)
    db.commit()
    db.refresh(item)
    return item


@router.delete("/{item_id}", status_code=204)
def delete_item(
    item_id: int,
    current_user: models.User = Depends(auth.get_current_user),
    db: Session = Depends(get_db),
):
    """删除商品（仅拥有者）"""
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="商品不存在")
    if item.owner_id != current_user.id:
        raise HTTPException(status_code=403, detail="无权删除他人商品")
    db.delete(item)
    db.commit()

6. 主应用入口

        
        
        
    
# main.py
from fastapi import FastAPI
from database import Base, engine
from routers import auth, users, items

# 创建数据表
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="商品管理 API",
    description="FastAPI + SQLAlchemy + JWT 认证的 RESTful API 示例",
    version="1.0.0",
)

# 注册路由
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(items.router)


@app.get("/", tags=["根"])
def root():
    return {"message": "欢迎使用商品管理 API", "docs": "/docs"}


@app.get("/health", tags=["根"])
def health_check():
    return {"status": "healthy"}

7. 中间件与异常处理

7.1 自定义异常

        
        
        
    
# main.py 追加
from fastapi import Request
from fastapi.responses import JSONResponse


class BizError(Exception):
    """业务异常"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message


@app.exception_handler(BizError)
async def biz_error_handler(request: Request, exc: BizError):
    return JSONResponse(
        status_code=200,
        content={"code": exc.code, "message": exc.message, "data": None},
    )


# 全局异常兜底
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误", "data": None},
    )

7.2 请求日志中间件

        
        
        
    
import time
import logging

logger = logging.getLogger("api")


@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = (time.time() - start) * 1000

    logger.info(
        f"{request.method} {request.url.path} "
        f"→ {response.status_code} ({duration:.1f}ms)"
    )
    return response

7.3 统一响应格式

        
        
        
    
# schemas.py 追加
from typing import TypeVar, Generic, Optional
from pydantic import BaseModel

T = TypeVar("T")


class ApiResponse(BaseModel, Generic[T]):
    """统一响应模型"""
    code: int = 200
    message: str = "success"
    data: Optional[T] = None


# 使用示例
@router.get("/{item_id}", response_model=ApiResponse[schemas.ItemResponse])
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if not item:
        raise BizError(code=404, message="商品不存在")
    return ApiResponse(data=item)

8. 自动文档与测试

8.1 访问交互式文档

启动服务后访问：
- Swagger UI:  http://localhost:8000/docs
- ReDoc:       http://localhost:8000/redoc

8.2 API 测试

        
        
        
    
# test_api.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)


def test_register_and_login():
    """测试注册和登录"""
    # 注册
    resp = client.post("/register", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "123456",
    })
    assert resp.status_code == 201

    # 登录
    resp = client.post("/token", data={
        "username": "testuser",
        "password": "123456",
    })
    assert resp.status_code == 200
    token = resp.json()["access_token"]
    return token


def test_create_item():
    """测试创建商品"""
    token = test_register_and_login()
    resp = client.post(
        "/items/",
        json={"name": "笔记本电脑", "price": 5999.0},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 201
    assert resp.json()["name"] == "笔记本电脑"

        
# 运行测试
pytest test_api.py -v

9. 小结

模块	技术选型
Web 框架	FastAPI（异步、高性能）
数据库 ORM	SQLAlchemy
数据校验	Pydantic
认证授权	JWT（python-jose）+ OAuth2
密码加密	passlib + bcrypt
自动文档	Swagger UI / ReDoc
测试	TestClient + pytest

10. 练习题

为商品 API 添加分类（Category）功能，实现商品与分类的一对多关系。
实现分页查询，返回 total、page、items 的分页结构。
添加管理员角色（is_superuser），管理员可删除任何用户的商品。
编写完整的 pytest 测试用例，覆盖所有 API 端点。

下节预告：我们将总结 Python 项目的最佳实践，涵盖目录结构、配置管理与部署。