数据库与 ORM：SQLAlchemy 集成

核心理念

生产环境的 API 需要持久化数据。FastAPI 与 SQLAlchemy 的集成提供了类型安全的 ORM，配合依赖注入实现数据库会话管理。

SQLAlchemy 核心组件

1. Engine - 数据库连接池

from sqlalchemy import create_engine

DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(
    DATABASE_URL,
    pool_size=5,          # 连接池大小
    max_overflow=10,      # 超出pool_size的最大连接数
    pool_pre_ping=True,   # 连接前测试
    echo=True            # 打印 SQL 语句（开发环境）
)

2. SessionLocal - 会话工厂

from sqlalchemy.orm import sessionmaker

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

3. Base - 声明式基类

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

定义模型

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime

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)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 关系
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"))
    
    # 关系
    author = relationship("User", back_populates="posts")

数据库依赖注入

def get_db():
    db = SessionLocal()
    try:
        yield db  # 提供会话
    finally:
        db.close()  # 请求结束后关闭

执行流程：

1. 请求到来
2. 创建数据库会话
3. yield db - 暂停，注入到路径操作
4. 路径操作执行
5. finally 块执行，关闭会话

CRUD 操作

Create

from sqlalchemy.orm import Session

@app.post("/users/", response_model=UserResponse)
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="Email already registered")
    
    # 创建用户
    db_user = User(
        email=user.email,
        username=user.username,
        hashed_password=hash_password(user.password)
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)  # 刷新以获取数据库生成的字段
    return db_user

Read

@app.get("/users/{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 not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/", response_model=List[UserResponse])
def list_users(
    skip: int = 0,
    limit: int = 10,
    db: Session = Depends(get_db)
):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

Update

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(
    user_id: int,
    user_update: UserUpdate,
    db: Session = Depends(get_db)
):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    # 更新字段
    for key, value in user_update.dict(exclude_unset=True).items():
        setattr(db_user, key, value)
    
    db.commit()
    db.refresh(db_user)
    return db_user

Delete

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    db.delete(db_user)
    db.commit()
    return {"message": "User deleted successfully"}

查询优化

预加载关系（避免 N+1）

from sqlalchemy.orm import joinedload

@app.get("/users/{user_id}/with-posts")
def get_user_with_posts(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User)\
        .options(joinedload(User.posts))\
        .filter(User.id == user_id)\
        .first()
    
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

条件查询

@app.get("/users/search")
def search_users(
    username: str = None,
    email: str = None,
    db: Session = Depends(get_db)
):
    query = db.query(User)
    
    if username:
        query = query.filter(User.username.contains(username))
    if email:
        query = query.filter(User.email == email)
    
    return query.all()

与 Django ORM 对比

特性	SQLAlchemy	Django ORM
查询风格	显式查询	QuerySet 链式
类型安全	✅ (Pydantic)	部分
迁移工具	Alembic	内置
会话管理	手动	自动
灵活性	高	中

返回第三章概览