Python教程 - 数据库进阶:SQLAlchemy ORM、Alembic 迁移与 PostgreSQL 实战

欢迎来到第十二期！

在第一期的基础教程中，我们用一个简单的字典来存学生成绩；在第七期，我们用 SQLite 给 FastAPI 项目搭了个持久化后端；在第十期，FinanceFlow 项目的数据库虽然能跑，但你仔细一看——全是手动拼 SQL，数据库切换还得改一堆代码……

**这就像盖房子只打了地基，还没砌墙呢。**

今天这一期，我们要完成 Python 后端开发中最核心的基础设施之一——**数据库层的工程化**。

学完这一期，你将能够：

- 用 **SQLAlchemy ORM** 替代手写 SQL，让 Python 对象直接映射到数据库表

- 用 **Alembic** 管理数据库版本迁移，告别"删库重建"的野蛮操作

- 用 **PostgreSQL** 替代 SQLite，让你的应用能扛住真实流量

- 掌握 **多表关联、外键约束、索引优化** 等数据库进阶技能

- 把 FinanceFlow 的数据层从玩具升级为生产级架构

>**前置知识**：需要熟悉本系列前 11 期的内容，特别是 Episode 07（FastAPI）和 Episode 10（FinanceFlow 项目）。

---

## 12.1 为什么需要 ORM？

### 12.1.1 手写 SQL 的痛苦

先看一段典型的「不写 ORM」的代码。假设我们要做一个用户注册功能：

```python

# ❌ 糟糕的做法：到处散落着字符串拼接的 SQL

import sqlite3

defcreate_user(username, email, password_hash):

    conn = sqlite3.connect("app.db")

    cursor = conn.cursor()

    cursor.execute(

f"INSERT INTO users (username, email, password_hash) "

f"VALUES ('{username}', '{email}', '{password_hash}')"

    )

    conn.commit()

    conn.close()

defget_user_by_username(username):

    conn = sqlite3.connect("app.db")

    cursor = conn.cursor()

# ⚠️ SQL 注入攻击风险！用户名里带单引号就挂了

    cursor.execute(f"SELECT * FROM users WHERE username = '{username}'")

    user = cursor.fetchone()

    conn.close()

return user

```

这种写法有几个致命问题：

1.**SQL 注入**：上面的 `f-string` 拼接是经典漏洞。`username = "admin' OR '1'='1"` 就能直接绕过登录。

2.**代码混乱**：SQL 散落在各个函数里，改一个表结构要找到所有引用它的手动查询。

3.**无法复用连接**：每个函数都新建连接，没有连接池概念。

4.**数据库锁定**：SQLite 是文件锁机制，并发写入会直接报错 "database is locked"。

### 12.1.2 ORM 的思路

ORM（Object-Relational Mapping，对象关系映射）的核心思想很简单：**用操作 Python 对象的方式来操作数据库，而不是写 SQL 语句。**

```python

# ✅ 使用 SQLAlchemy ORM

from app import db, User

# 创建用户——注意，这不是 INSERT 语句，而是创建了一个 Python 对象

new_user = User(username="alice", email="alice@example.com", password_hash="hashed_secret")

db.session.add(new_user)      # 标记"准备插入"

db.session.commit()            # 提交事务，实际执行 INSERT

# 查询用户——不是 SELECT，而是 filter_by

user = db.session.query(User).filter_by(username="alice").first()

print(user.email)   # alice@example.com

```

你看不到任何 SQL 字符串，但底层 SQLAlchemy 会自动帮你翻译成 `INSERT` / `SELECT` / `UPDATE` / `DELETE`。

ORM 的三大核心价值：

| 价值 | 说明 |

|------|------|

| **安全性** | 参数化查询自动防止 SQL 注入 |

| **可维护性** | 表结构定义集中管理，改一处全项目生效 |

| **可移植性** | 从 SQLite 切到 PostgreSQL 只需改一行配置 |

---

## 12.2 SQLAlchemy 基础

### 12.2.1 安装与环境准备

```bash

pipinstallsqlalchemy"psycopg2-binary"alembic

```

| 包名 | 用途 |

|------|------|

| `sqlalchemy` | ORM 核心框架 |

| `psycopg2-binary` | PostgreSQL 驱动（开发环境用 binary 版） |

| `alembic` | 数据库迁移工具 |

### 12.2.2 初始化和配置

先创建数据库连接配置模块：

```python

# app/database.py

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker, DeclarativeBase

from sqlalchemy.pool import StaticPool

import os

# 数据库 URL，优先读环境变量，默认用 SQLite

DATABASE_URL = os.getenv(

"DATABASE_URL",

"sqlite:///./app.db"# 开发环境默认 SQLite

)

# 创建引擎

# echo=True 会在控制台打印执行的 SQL（开发时很有用）

engine = create_engine(

DATABASE_URL,

echo=False,           # 生产环境关闭 SQL 日志

connect_args={"check_same_thread": False} if"sqlite"inDATABASE_URLelse {},

)

# 会话工厂——每次请求创建一个新会话

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

# Base 类——所有模型都要继承它

classBase(DeclarativeBase):

pass

# 依赖注入：给 FastAPI 用的获取会话函数

defget_db():

    db = SessionLocal()

try:

yield db

finally:

        db.close()

```

关键点：

-**`create_engine`**：创建数据库引擎，它是 ORM 与数据库之间的桥梁。

-**`sessionmaker`**：会话工厂，类比于"数据库连接的管理器"。

-**`DeclarativeBase`**：声明式基类，所有模型通过继承它来声明自己的表结构。

-**`connect_args`**：SQLite 的特有参数，解决多线程下 "database is locked" 的问题。

### 12.2.3 定义第一个模型

```python

# app/models/user.py

from datetime import datetime

from sqlalchemy import String, Integer, DateTime, Boolean

from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base

classUser(Base):

    __tablename__ = "users"# 数据库中的表名

# ===== 字段定义 =====

id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)

    email: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)

    password_hash: Mapped[str] = mapped_column(String(256), nullable=False)

    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

# ===== 关系定义 =====

# 一个用户有多个交易记录（一对多）

    transactions = relationship("Transaction", back_populates="owner", cascade="all, delete-orphan")

def__repr__(self):

returnf"<User {self.username}>"

```

逐行拆解：

-**`Mapped[T]`**：类型提示语法，告诉 SQLAlchemy 这个字段是什么类型。

-**`mapped_column`**：定义列的属性（主键、唯一、非空、索引等）。

-**`index=True`**：在 username 字段上创建索引，加速 `WHERE username = ?` 查询。

-**`relationship`**：定义模型之间的关系。`cascade="all, delete-orphan"` 表示删除用户时自动删除其所有交易记录。

-**`onupdate=datetime.utcnow`**：字段值更新时自动修改时间戳。

### 12.2.4 多表关系建模

以 FinanceFlow 为例，我们有用户、交易、分类三个实体：

```python

# app/models/transaction.py

from sqlalchemy import String, Float, DateTime, ForeignKey, Enum as SAEnum

from sqlalchemy.orm import Mapped, mapped_column, relationship

import enum

import uuid

from app.database import Base

classTransactionType(str, enum.Enum):

INCOME = "income"

EXPENSE = "expense"

classTransaction(Base):

    __tablename__ = "transactions"

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))

    amount: Mapped[float] = mapped_column(Float, nullable=False)

type: Mapped[TransactionType] = mapped_column(SAEnum(TransactionType), nullable=False)

    description: Mapped[str] = mapped_column(String(256), nullable=True)

    date: Mapped[str] = mapped_column(String(10), nullable=False, default=lambda: datetime.utcnow().strftime("%Y-%m-%d"))

# 外键——关联到 users 表

    owner_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False, index=True)

    category_id: Mapped[int] = mapped_column(Integer, ForeignKey("categories.id"), nullable=True)

# 关系

    owner = relationship("User", back_populates="transactions")

    category = relationship("Category", back_populates="transactions")

def__repr__(self):

returnf"<Transaction {self.type.value} ${self.amount}>"

# app/models/category.py

from sqlalchemy import String

from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base

classCategory(Base):

    __tablename__ = "categories"

id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    name: Mapped[str] = mapped_column(String(50), nullable=False)

type: Mapped[TransactionType] = mapped_column(SAEnum(TransactionType), nullable=False)

    icon: Mapped[str] = mapped_column(String(10), default="📁")

# 关系

    transactions = relationship("Transaction", back_populates="category")

def__repr__(self):

returnf"<Category {self.icon}{self.name}>"

```

关系总览：

```

User (1) ───< Transaction (>1) ──> Category (1)

一个用户可以有多条交易

一条交易属于一个用户和一个分类

一个分类可以包含多条交易

```

这就是典型的 **1:N（一对多）** 关系。

### 12.2.5 创建表与基础 CRUD

```python

# app/crud/users.py

from sqlalchemy import select

from app.database import SessionLocal, Base, engine

from app.models.user import User

# 第一步：创建所有表（相当于 MySQL 的 CREATE TABLE ...）

Base.metadata.create_all(bind=engine)

```

```python

# app/crud/users.py —— CRUD 操作

from sqlalchemy import select, func

from datetime import datetime

from app.database import SessionLocal

from app.models.user import User

defcreate_user(username: str, email: str, password_hash: str) -> User:

"""创建新用户"""

    db = SessionLocal()

try:

        user = User(

username=username,

email=email,

password_hash=password_hash

        )

        db.add(user)

        db.commit()

        db.refresh(user)      # 刷新：获取数据库中生成的 id、created_at 等字段

return user

exceptExceptionas e:

        db.rollback()          # 出错回滚

raise e

finally:

        db.close()

defget_user_by_id(user_id: int) -> User | None:

"""通过 ID 查询用户"""

    db = SessionLocal()

try:

return db.query(User).filter(User.id == user_id).first()

finally:

        db.close()

defget_user_by_username(username: str) -> User | None:

"""通过用户名查询"""

    db = SessionLocal()

try:

return db.query(User).filter(User.username == username).first()

finally:

        db.close()

deflist_users(skip: int = 0, limit: int = 100) -> list[User]:

"""分页列出用户"""

    db = SessionLocal()

try:

return db.query(User).offset(skip).limit(limit).all()

finally:

        db.close()

defupdate_user_email(user_id: int, new_email: str) -> User | None:

"""更新用户邮箱"""

    db = SessionLocal()

try:

        user = db.query(User).filter(User.id == user_id).first()

ifnot user:

returnNone

        user.email = new_email

        user.updated_at = datetime.utcnow()

        db.commit()

        db.refresh(user)

return user

finally:

        db.close()

defdelete_user(user_id: int) -> bool:

"""删除用户"""

    db = SessionLocal()

try:

        user = db.query(User).filter(User.id == user_id).first()

ifnot user:

returnFalse

        db.delete(user)

        db.commit()

returnTrue

finally:

        db.close()

```

```python

# 测试一下

from app.crud.users import create_user, get_user_by_username

# 创建

alice = create_user("alice", "alice@example.com", "hashed_pwd_123")

print(alice)           # <User alice>

print(alice.id)        # 1

print(alice.created_at)  # 2026-07-09 06:00:00

# 查询

user = get_user_by_username("alice")

print(user.email)      # alice@example.com

```

对应的 SQL：

| ORM 操作 | 生成的 SQL |

|----------|-----------|

| `create_user()` | `INSERT INTO users (username, email, password_hash, ...) VALUES (?, ?, ?, ...)` |

| `get_user_by_id(1)` | `SELECT * FROM users WHERE id = ? LIMIT ?` |

| `list_users(skip=0, limit=100)` | `SELECT * FROM users OFFSET 0 LIMIT 100` |

| `update_user_email(1, "new@email")` | `UPDATE users SET email=?, updated_at=? WHERE id=?` |

| `delete_user(1)` | `DELETE FROM users WHERE id=?` |

你不需要写任何 SQL，SQLAlchemy 自动生成安全参数化的查询。

---

## 12.3 SQLAlchemy 高级用法

### 12.3.1 复杂查询

```python

from sqlalchemy import select, func, and_, or_

from app.models.user import User

from app.models.transaction import Transaction, TransactionType

from app.models.category import Category

db = SessionLocal()

# 查询 1：找出所有消费超过 1000 元的用户

stmt = (

    select(User)

    .join(Transaction, User.id == Transaction.owner_id)  # JOIN

    .where(Transaction.type == TransactionType.EXPENSE)   # WHERE

    .group_by(User.id)                                   # GROUP BY

    .having(func.sum(Transaction.amount) > 1000)         # HAVING

)

high_spenders = db.execute(stmt).scalars().all()

# 查询 2：用户列表 + 交易统计（子查询）

subquery = (

    select(

        Transaction.owner_id.label("user_id"),

        func.count(Transaction.id).label("tx_count"),

        func.sum(Transaction.amount).label("total_amount")

    )

    .group_by(Transaction.owner_id)

    .subquery()

)

stmt = (

    select(User.username, User.email, subquery.c.tx_count, subquery.c.total_amount)

    .outerjoin(subquery, User.id == subquery.c.user_id)  # LEFT OUTER JOIN

)

results = db.execute(stmt).all()

for row in results:

print(f"{row.username}: {row.tx_count} 笔交易，合计 ¥{row.total_amount or0:.2f}")

# 查询 3：模糊搜索 + 排序 + 分页

keyword = "%花呗%"

stmt = (

    select(Transaction)

    .where(Transaction.description.like(keyword))

    .order_by(Transaction.date.desc())   # 按日期倒序

    .limit(20)

    .offset(0)

)

search_results = db.execute(stmt).scalars().all()

```

### 12.3.2 事务处理

```python

from app.models.user import User

from app.models.transaction import Transaction

db = SessionLocal()

try:

# 转账操作：扣 A 的钱，加 B 的钱，要么全部成功，要么全部失败

    sender = db.query(User).filter(User.username == "alice").first()

    receiver = db.query(User).filter(User.username == "bob").first()

# 创建转账记录

    transfer = Transaction(

amount=100.0,

type=TransactionType.EXPENSE,

description="转账给 bob"

    )

    sender.transactions.append(transfer)

    income = Transaction(

amount=100.0,

type=TransactionType.INCOME,

description="收到 alice 的转账"

    )

    receiver.transactions.append(income)

# 提交事务——如果中间任何一步出错，全部回滚

    db.commit()

print("转账成功！")

exceptExceptionas e:

    db.rollback()   # 回滚到转账前的状态

print(f"转账失败：{e}")

finally:

    db.close()

```

### 12.3.3 连接 PostgreSQL

```python

# 将 DATABASE_URL 改为 PostgreSQL：

import os

os.environ["DATABASE_URL"] = "postgresql+psycopg2://user:password@localhost:5432/financeflow"

# 或者在 docker-compose.yml 中传递环境变量：

# environment:

#   - DATABASE_URL=postgresql+psycopg2://postgres:mysecretpassword@db:5432/financeflow

```

只需要改一行配置，从 SQLite 切换到 PostgreSQL，**所有 ORM 代码无需改动**。这就是 ORM 的可移植性优势。

---

## 12.4 Alembic：数据库版本管理

### 12.4.1 为什么需要迁移？

想象这个场景：

```

你上线了 v1.0 版本，数据库表如下：

CREATE TABLE users (

    id INTEGER PRIMARY KEY,

    username VARCHAR(50),

    email VARCHAR(120)

);

后来，你想给用户加上"头像 URL"字段：

ALTER TABLE users ADD COLUMN avatar_url TEXT;

```

手动执行 `ALTER TABLE` 没问题，但当你换了台新服务器、或者有同事拉了最新代码时——**他们本地并没有这张 `avatar_url` 列**。

如果你下次又加了"最后登录时间"，再手动执行 `ALTER`，第三次手动执行 `ALTER`……最终你的团队会有不同的人在不同的数据库上执行不同的 `ALTER` 语句——混乱不可避免。

**这就是 Alembic 解决的问题：像 Git 管代码一样，用版本控制的方式管数据库表结构。**

### 12.4.2 初始化 Alembic

```bash

# 在项目根目录执行

alembicinitalembic

# 项目结构变成这样：

financeflow/

├──app/

│├──database.py# SQLAlchemy 配置

│└──models/# 所有模型

├──alembic/

│├──env.py# Alembic 运行配置

│└──versions/# 迁移脚本存放处

│└──.gitkeep

├──alembic.ini# Alembic 全局配置

└──requirements.txt

```

编辑 `alembic.ini`：

```ini

# 找到 sqlalchemy.url 这行，改为你的数据库 URL

sqlalchemy.url = postgresql+psycopg2://postgres:mysecretpassword@localhost:5432/financeflow

```

编辑 `alembic/env.py`：

```python

# alembic/env.py

import sys

from pathlib import Path

from logging.config import fileConfig

from sqlalchemy import engine_from_config, pool

from alembic import context

# 确保能导入 app 模块

sys.path.insert(0, str(Path(__file__).parent.parent))

config = context.config

if config.config_file_name isnotNone:

    fileConfig(config.config_file_name)

# 关键：导入所有模型，让 Alembic 能检测到表结构

from app.database import Base

from app.models.user import User

from app.models.transaction import Transaction

from app.models.category import Category

target_metadata = Base.metadata

defrun_migrations_offline():

    url = config.get_main_option("sqlalchemy.url")

    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)

with context.begin_transaction():

        context.run_migrations()

defrun_migrations_online():

    connectable = engine_from_config(

        config.get_section(config.config_ini_section),

prefix="sqlalchemy.",

poolclass=pool.NullPool,

    )

with connectable.connect() as connection:

        context.configure(connection=connection, target_metadata=target_metadata)

with context.begin_transaction():

            context.run_migrations()

if context.is_offline_mode():

    run_migrations_offline()

else:

    run_migrations_online()

```

### 12.4.3 自动生成迁移脚本

```bash

# 第一步：检查当前模型有哪些变更

alembiccheck

# 第二步：自动生成迁移脚本

alembicrevision--autogenerate-m"初始迁移，创建所有表"

# 你会看到 alembic/versions/ 目录下多了一个文件：

# 0001_initial_migration.py

# 内容示例：

"""初始迁移，创建所有表

Revision ID: abc123

Revises: 

Create Date: 2026-07-09 10:00:00.000000

"""

fromalembicimportop

importsqlalchemyassa

defupgrade():

op.create_table('users',

sa.Column('id',sa.Integer(),primary_key=True,autoincrement=True),

sa.Column('username',sa.String(length=50),nullable=False),

sa.Column('email',sa.String(length=120),nullable=False),

sa.Column('password_hash',sa.String(length=256),nullable=False),

sa.Column('is_active',sa.Boolean(),server_default=sa.text('true')),

sa.Column('created_at',sa.DateTime(),server_default=sa.func.now()),

sa.UniqueConstraint('username'),

sa.UniqueConstraint('email'),

    )

# ... 其他表的创建语句

defdowngrade():

op.drop_table('users')

# ... 其他表的删除语句（反向操作）

```

### 12.4.4 执行迁移

```bash

# 执行最新的迁移

alembicupgradehead

# 查看当前版本

alembiccurrent

# 查看迁移历史

alembichistory--verbose

# 回滚到上一个版本

alembicdowngrade-1

# 回滚到指定版本

alembicdowngradeabc123

# 回滚到初始状态（删库级别，慎用！）

alembicdowngradebase

```

### 12.4.5 实际工作流

```

日常工作流示意：

第 1 天：创建 models → 生成迁移 → 执行迁移 → 上线 v1.0 ✅

第 5 天：给用户表加 "phone" 字段 → 重新生成迁移 → 执行迁移 → 上线 v1.1 ✅

第 10 天：新同事拉代码 → 运行 "alembic upgrade head" → 自动拿到最新版表结构 ✅

```

---

## 12.5 PostgreSQL 实战

### 12.5.1 为什么选 PostgreSQL？

| 特性 | SQLite | PostgreSQL |

|------|--------|------------|

| 并发写入 | 差（文件锁） | 优秀（MVCC） |

| 数据量 | 适合小项目 |  TB 级没压力 |

| 高级类型 | JSON、数组 | ✅ JSONB、数组、几何、GIS |

| 全文检索 | 无 | ✅ tsvector |

| 权限控制 | 粗粒度 | ✅ 行级安全（RLS） |

| 适用场景 | 开发/小型个人项目 | 生产环境 |

对于个人项目和原型验证，SQLite 完全够用。但一旦你要面对**多用户同时写入**、**日活上千**、**数据量过万**，PostgreSQL 是最佳选择。

### 12.5.2 本地启动 PostgreSQL

```bash

# 方式 1：用 Docker 启动（推荐）

dockerrun-d\

  --name pg-financeflow \

-ePOSTGRES_PASSWORD=mysecretpassword\

  -e POSTGRES_DB=financeflow \

-p5432:5432\

  postgres:16-alpine

# 方式 2：macOS Homebrew

brewinstallpostgresql

brewservicesstartpostgresql

# 方式 3：Windows - 下载官方安装包

# https://www.postgresql.org/download/windows/

```

连接测试：

```bash

# 命令行连接

psql-Upostgres-dfinanceflow

# 输出：

# psql (16.0)

# Type "help" for help.

#

# financeflow=# SELECT version();

#  PostgreSQL 16.x on ...

# financeflow=# \q

```

### 12.5.3 集成到 FastAPI

```python

# app/main.py

from fastapi import FastAPI, Depends, HTTPException

from sqlalchemy.orm import Session

from pydantic import BaseModel, EmailStr

from app.database import engine, Base, get_db

from app.models.user import User

from app.models.transaction import Transaction, TransactionType

from app.models.category import Category

from app.crud.users import create_user, get_user_by_username

# 创建表（开发环境）

Base.metadata.create_all(bind=engine)

app = FastAPI(title="FinanceFlow API", version="2.0.0")

# ===== Pydantic 模型（用于 API 请求/响应验证）=====

classUserCreate(BaseModel):

    username: str

    email: EmailStr

    password: str# 实际应在服务端哈希后存入 password_hash

classTransactionCreate(BaseModel):

    amount: float

type: TransactionType

    description: str | None = None

    category_id: int | None = None

@app.post("/api/users", response_model=UserCreate)

defregister_user(user_data: UserCreate, db: Session = Depends(get_db)):

"""注册新用户"""

    existing = get_user_by_username(user_data.username)

if existing:

raise HTTPException(status_code=409, detail="用户名已存在")

# 密码哈希（生产环境用 passlib + bcrypt）

from passlib.context import CryptContext

    pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")

    hashed = pwd_ctx.hash(user_data.password)

    user = create_user(user_data.username, user_data.email, hashed)

return user

@app.get("/api/users/{user_id}/stats")

defget_user_stats(user_id: int, db: Session = Depends(get_db)):

"""获取用户的财务统计"""

from sqlalchemy import func

    user = db.query(User).filter(User.id == user_id).first()

ifnot user:

raise HTTPException(status_code=404, detail="用户不存在")

    stats = db.query(

        func.sum(Transaction.amount).label("total_expense"),

        func.count(Transaction.id).label("tx_count")

    ).filter(

        Transaction.owner_id == user_id,

        Transaction.type == TransactionType.EXPENSE

    ).first()

return {

"user_id": user_id,

"username": user.username,

"total_expenses": float(stats.total_expense or0),

"transaction_count": stats.tx_count or0,

    }

```

```bash

# 启动服务

uvicornapp.main:app--reload--host0.0.0.0--port8000

# 测试注册

curl-XPOSThttp://localhost:8000/api/users\

  -H "Content-Type: application/json" \

-d'{"username": "alice", "email": "alice@example.com", "password": "secure123"}'

# 返回：

# {"username": "alice", "email": "alice@example.com"}

```

---

## 12.6 实操练习

### 练习 1：从零搭建一个简单的图书管理系统

**题目**：设计 `Book` 和 `Author` 两个模型，实现多对多关系（一个作者可以写多本书，一本书也可以有多个合著者）。

**任务**：

1. 定义 `Book`、`Author` 表和中间关联表（`book_authors`）

2. 编写 `add_author_to_book(book_id, author_id)` 和 `get_books_by_author(author_id)` 函数

3. 编写 Alembic 迁移脚本并执行

<details>

<summary>参考答案</summary>

```python

from sqlalchemy import Table, Column, Integer, String, ForeignKey

from sqlalchemy.orm import Mapped, mapped_column, relationship

# 多对多中间表（不需要继承 Base，只是纯关联）

book_authors = Table(

"book_authors",

    Base.metadata,

    Column("book_id", Integer, ForeignKey("books.id"), primary_key=True),

    Column("author_id", Integer, ForeignKey("authors.id"), primary_key=True),

)

classAuthor(Base):

    __tablename__ = "authors"

id: Mapped[int] = mapped_column(Integer, primary_key=True)

    name: Mapped[str] = mapped_column(String(100), nullable=False)

    books: Mapped[list["Book"]] = relationship("Book", secondary=book_authors, back_populates="authors")

classBook(Base):

    __tablename__ = "books"

id: Mapped[int] = mapped_column(Integer, primary_key=True)

    title: Mapped[str] = mapped_column(String(200), nullable=False)

    isbn: Mapped[str] = mapped_column(String(13), unique=True)

    authors: Mapped[list["Author"]] = relationship("Author", secondary=book_authors, back_populates="books")

```

```python

# CRUD 操作

from sqlalchemy import select

defadd_author_to_book(db: Session, book_id: int, author_id: int):

    book = db.query(Book).filter(Book.id == book_id).first()

    author = db.query(Author).filter(Author.id == author_id).first()

if book and author:

        book.authors.append(author)

        db.commit()

defget_books_by_author(db: Session, author_id: int) -> list[Book]:

    author = db.query(Author).filter(Author.id == author_id).first()

ifnot author:

return []

return author.books

```

```bash

# 生成并执行迁移

alembicrevision--autogenerate-m"创建书籍和作者表"

alembicupgradehead

```

</details>

---

### 练习 2：性能优化——给慢查询加索引

**题目**：FinanceFlow 的 `/api/transactions` 接口在数据量大时变得很慢。分析原因并优化。

```python

# 原始查询（慢）

@app.get("/api/transactions")

deflist_transactions(

user_id: int,

start_date: str,

end_date: str,

db: Session = Depends(get_db)

):

return db.query(Transaction).filter(

        Transaction.owner_id == user_id,

        Transaction.date >= start_date,

        Transaction.date <= end_date

    ).order_by(Transaction.date.desc()).all()

```

**任务**：

1. 指出为什么这个查询在万级数据时会慢

2. 在模型中添加复合索引

3. 用 `EXPLAIN ANALYZE` 验证优化效果

<details>

<summary>参考答案</summary>

```python

# 解决方案 1：在模型中给常用查询字段加索引

classTransaction(Base):

    __tablename__ = "transactions"

# 已有 owner_id 索引

    owner_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), index=True)

# 新增：给 date 也加索引（用于范围查询）

    date: Mapped[str] = mapped_column(String(10), index=True)

# 更优方案：加复合索引（需要原生 SQL）

# 可以用 event listener 或手动写 migration：

```

```python

# 在 Alembic 迁移脚本中手动添加复合索引：

defupgrade():

    op.create_index(

"ix_transaction_owner_date",

"transactions",

        ["owner_id", "date"],

unique=False,

    )

defdowngrade():

    op.drop_index("ix_transaction_owner_date", table_name="transactions")

```

```bash

# 验证：在 psql 中执行 EXPLAIN ANALYZE

EXPLAINANALYZESELECT*FROMtransactions

WHEREowner_id=1ANDdate >='2026-01-01'ANDdate <='2026-12-31';

--优化前：SeqScan（全表扫描），耗时~50ms（1万条数据）

--优化后：IndexScan（索引扫描），耗时~1ms

```

</details>

---

### 练习 3：写一个复杂的聚合查询

**题目**：为每个用户生成分月的支出报告。

**任务**：

1. 用 SQLAlchemy 写出按月分组 + 按类别分组的支出汇总

2. 返回格式如：

```json

{

"month": "2026-06",

"categories": [

    {"name": "餐饮", "amount": 1500.00},

    {"name": "交通", "amount": 300.00},

    {"name": "购物", "amount": 800.00}

  ],

"total": 2600.00

}

```

<details>

<summary>参考答案</summary>

```python

from sqlalchemy import extract, func

from datetime import datetime

defget_monthly_report(db: Session, user_id: int, year: int, month: int):

# 筛选当月数据

    monthly_txns = (

        db.query(Transaction, Category.name.label("cat_name"))

        .outerjoin(Category, Transaction.category_id == Category.id)

        .join(User, Transaction.owner_id == User.id)

        .filter(

            User.id == user_id,

            extract("year", Transaction.date) == year,

            extract("month", Transaction.date) == month,

            Transaction.type == TransactionType.EXPENSE,

        )

        .all()

    )

# 按类别聚合

    cat_totals = {}

    total = 0.0

for txn, cat_name in monthly_txns:

        key = cat_name or"未分类"

        cat_totals[key] = cat_totals.get(key, 0.0) + txn.amount

        total += txn.amount

return {

"month": f"{year}-{month:02d}",

"categories": [

            {"name": k, "amount": round(v, 2)}

for k, v insorted(cat_totals.items(), key=lambdax: -x[1])

        ],

"total": round(total, 2),

    }

```

</details>

---

### 练习 4：从 SQLite 迁移到 PostgreSQL

**题目**：你有一个用 SQLite 开发的 FinanceFlow 项目，现在要迁移到 PostgreSQL。

**任务**：

1. 列出完整迁移步骤

2. 如果迁移过程中发现 Alembic 迁移丢失了，该怎么办？

<details>

<summary>参考答案</summary>