构建一个完整的 REST API 服务

使用 FastAPI + SQLAlchemy + Pydantic 构建一个带认证、分页、错误处理的 RESTful API

你将学到

使用 FastAPI 框架构建高性能 API
使用 SQLAlchemy ORM 操作数据库
使用 Pydantic 进行数据验证和序列化
实现 JWT 认证和授权
设计统一的错误处理和响应格式

前置知识

架构设计

api-project/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI 应用入口
│   ├── config.py         # 配置管理
│   ├── database.py       # 数据库连接
│   ├── models/           # SQLAlchemy 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/          # Pydantic 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/          # API 路由
│   │   ├── __init__.py
│   │   └── users.py
│   ├── services/         # 业务逻辑
│   │   ├── __init__.py
│   │   └── auth.py
│   └── utils/            # 工具函数
│       ├── __init__.py
│       └── security.py
├── requirements.txt
└── run.py

实现步骤

第一步：项目初始化和依赖

mkdir api-project && cd api-project
python -m venv venv
source venv/bin/activate

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

# requirements.txt
fastapi==0.109.0
uvicorn==0.27.0
sqlalchemy==2.0.25
pydantic==2.5.3
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6

第二步：数据库配置和模型

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

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

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

第三步：Pydantic 模型（数据验证）

# app/schemas/user.py
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional

class UserCreate(BaseModel):
    email: EmailStr
    username: str
    password: str

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    username: Optional[str] = None

class UserResponse(BaseModel):
    id: int
    email: str
    username: str
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True

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

class TokenData(BaseModel):
    username: Optional[str] = None

第四步：认证服务

# app/services/auth.py
from datetime import datetime, timedelta
from typing import Optional
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 app.database import get_db
from app.models.user import User
from app.schemas.user import TokenData

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

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

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

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    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)

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> 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
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception

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

第五步：API 路由

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate, UserResponse
from app.services.auth import get_password_hash, get_current_user

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

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    """创建新用户"""
    # 检查邮箱是否已存在
    db_user = db.query(User).filter(User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="邮箱已注册")

    # 检查用户名是否已存在
    db_user = db.query(User).filter(User.username == user.username).first()
    if db_user:
        raise HTTPException(status_code=400, detail="用户名已存在")

    # 创建用户
    hashed_password = get_password_hash(user.password)
    db_user = User(
        email=user.email,
        username=user.username,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@router.get("/", response_model=List[UserResponse])
def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """获取用户列表（分页）"""
    users = db.query(User).offset(skip).limit(limit).all()
    return users

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

@router.put("/{user_id}", response_model=UserResponse)
def update_user(
    user_id: int,
    user_update: UserUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """更新用户信息"""
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")

    update_data = user_update.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(user, field, value)

    db.commit()
    db.refresh(user)
    return user

@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(
    user_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """删除用户"""
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")

    db.delete(user)
    db.commit()

第六步：主应用和登录端点

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from datetime import timedelta
from app.database import engine, get_db
from app.models import user
from app.models.user import User
from app.schemas.user import Token
from app.routers import users
from app.services.auth import (
    verify_password,
    create_access_token,
    ACCESS_TOKEN_EXPIRE_MINUTES
)

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

app = FastAPI(title="用户管理 API", version="1.0.0")

# 注册路由
app.include_router(users.router, prefix="/api/v1")

@app.post("/token", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    """用户登录，获取 JWT Token"""
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/")
async def root():
    return {"message": "用户管理 API v1.0.0"}

@app.get("/health")
async def health():
    return {"status": "healthy"}

第七步：运行和测试

# run.py
import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

# 启动服务
python run.py

# 访问 API 文档
# http://localhost:8000/docs  (Swagger UI)
# http://localhost:8000/redoc (ReDoc)

# 测试 API
curl -X POST http://localhost:8000/api/v1/users/ \
  -H "Content-Type: application/json" \
  -d '{"email": "[email protected]", "username": "testuser", "password": "secret123"}'

curl -X POST http://localhost:8000/token \
  -d "username=testuser&password=secret123"

完整项目结构

api-project/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI 应用入口
│   ├── config.py         # 配置管理
│   ├── database.py       # 数据库连接和会话
│   ├── models/
│   │   ├── __init__.py
│   │   └── user.py       # SQLAlchemy 用户模型
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── user.py       # Pydantic 数据模型
│   ├── routers/
│   │   ├── __init__.py
│   │   └── users.py      # 用户 API 路由
│   └── services/
│       ├── __init__.py
│       └── auth.py       # 认证服务
├── requirements.txt
└── run.py

最佳实践

使用 Pydantic 做数据验证：自动验证请求数据，生成 API 文档
分离关注点：路由、模型、服务、数据库各司其职
使用 JWT 认证：无状态认证，适合分布式系统
分页查询：大数据量接口必须支持分页
统一错误处理：使用 HTTPException 返回标准错误格式

常见问题

Q: FastAPI 和 Flask 选哪个？ A: FastAPI 性能更好，自动生成 API 文档，类型提示支持更完善。新项目推荐 FastAPI。

Q: 如何处理数据库迁移？ A: 使用 Alembic，它是 SQLAlchemy 的官方迁移工具。

Q: 如何部署到生产环境？ A: 使用 Gunicorn + Uvicorn worker，前面加 Nginx 反向代理。

扩展挑战

添加角色权限系统（admin/user）
实现 API 限流（rate limiting）
添加日志记录和监控

课程	相关章节
Python	Web开发入门：Flask 框架简介
Python	使用 Flask 构建 REST API
Python	数据库基础：SQLite 与 Python
Python	异常处理（try-except-finally）

构建一个完整的 REST API 服务

构建一个完整的 REST API 服务

你将学到

前置知识

架构设计

实现步骤

第一步：项目初始化和依赖

第二步：数据库配置和模型

第三步：Pydantic 模型（数据验证）

第四步：认证服务

第五步：API 路由

第六步：主应用和登录端点

第七步：运行和测试

完整项目结构

最佳实践

常见问题

扩展挑战

相关课程

相关课程