第一篇我们把FastAPI的工程底盘搭稳了：
- 单体CRUD跑通
- 分层职责清晰
- 统一异常和响应结构固定

第二篇我们进入真正会影响项目寿命的一步：数据库接入与迁移策略
这一步如果做得随意，后期会出现本地能跑，线上全是坑的典型问题

本文目标：
1. 用SQLAlchemy2+Alembic完成FastAPI的数据库接入
2. 先用SQLite快速起步，再迁移到MySQL/PostgreSQL
3. 给出一套可维护的数据库工程实践清单

一、技术选型建议
 

推荐组合：
- ORM：SQLAlchemy2.x
- 迁移：Alembic
- 驱动：
  - SQLite：内置驱动
  - MySQL：pymysql或 syncmy
  - PostgreSQL：psycopg或  syncpg

为什么不建议先裸SQL再补ORM：
- 初期快，后期模型治理成本高
- 迁移脚本更难维护
- 团队协作时缺少统一抽象

二、推荐目录结构
 

app/
  db/
    base.py
    session.py
    models/
      user.py
      room.py
  repositories/
    user_repo.py
    room_repo.py
  services/
  routers/
alembic/
alembic.ini
 

角色说明：
- models：数据库模型定义
- session.py：数据库连接、会话生命周期
- epositories：数据访问细节，关联增删改查SQL相关逻辑
- services：业务规则，不直接拼SQL

三、从SQLite起步：最小可用接入
 

3.1 安装依赖

python -m pip install sqlalchemy alembic pydantic-settings

3.2 配置数据库URL

pp/config.py：

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str = "sqlite:///./app.db"


settings = Settings()

3.3 创建Engine与Session

pp/db/session.py：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.config import settings

engine = create_engine(settings.database_url, echo=False, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


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

3.4 定义Base和模型

pp/db/base.py：

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

 pp/db/models/user.py：

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(255), nullable=False, unique=True, index=True)

四、用Alembic管理结构变更
 

4.1 初始化

alembic init alembic

4.2 配置lembic/env.py

- 导入Base.metadata
- 使用项目配置中的database_url

4.3 生成迁移

alembic revision --autogenerate -m "create users and rooms"

4.4 执行迁移

alembic upgrade head

关键建议：
- 结构变更一律通过迁移脚本，不直接改数据库
- 每次发布都执行迁移，保证环境一致

五、Repository层：把数据访问细节收口

epositories/user_repo.py：

from sqlalchemy.orm import Session
from app.db.models.user import User


class UserRepository:
    def get_by_email(self, db: Session, email: str) -> User | None:
        return db.query(User).filter(User.email == email).first()

    def list_all(self, db: Session) -> list[User]:
        return db.query(User).all()

    def create(self, db: Session, name: str, email: str) -> User:
        user = User(name=name, email=email)
        db.add(user)
        db.commit()
        db.refresh(user)
        return user

为什么要repo层：
- ORM细节不污染service
- 后续换数据库或优化查询更可控

六、SQLite->MySQL/PostgreSQL迁移步骤
 

迁移步骤建议：
1. 保持ORM模型不变，先保证抽象层稳定
2. 修改 DATABASE_URL 指向新数据库
3. 新库执行  lembic upgrade head 建结构
4. 做数据迁移
5. 回归测试关键链路
6. 切流并观察日志与慢查询

常见连接串：
- MySQL：
  - mysql+pymysql://user:password@host:3306/dbname
- PostgreSQL：
  - postgresql+psycopg://user:password@host:5432/dbname

七、MySQL与PostgreSQL差异
 

1. 字符集与排序规则：
- MySQL需确认utf8mb4
- PostgreSQL默认UTF-8友好

2. 时间字段：
- 统一使用UTC存储
- 应用层做时区展示转换

3. JSON字段能力：
- PostgreSQL JSONB查询能力更强
- MySQL 也支持JSON，但表达能力略弱

4. 自增策略：
- MySQL常见AUTO_INCREMENT
- PostgreSQL常见SERIAL/IDENTITY

5. Upsert 语法不同：
- MySQL：ON DUPLICATE KEY UPDATE
- PostgreSQL：ON CONFLICT ... DO UPDATE

八、事务与一致性：别只会commit()
 

推荐原则：
- 一个业务动作一个事务边界
- service层明确事务语义
- 对关键写操作补唯一约束，避免并发写穿

示例思路：
- 先查重再写入并不绝对安全
- 最终要靠数据库唯一索引兜底
- 捕获唯一冲突异常并映射为400/409

九、性能与索引
 

索引建议：
- 高频筛选字段加索引，如email、room_name
- 高频排序字段评估联合索引
- 避免全表扫描+分页深翻页组合

分页建议：
- 初期可用offset/limit
- 数据量上来后考虑游标分页

十、数据库接入后的异常映射建议
 

建议增加这类映射：
- 唯一冲突 -> 409或400，团队统一即可
- 外键冲突 -> 400
- 连接失败/超时 -> 503
- 未知数据库错误 -> 500

十一、容易踩的坑
 

1. 开发环境和生产环境数据库行为不一致
- 解决：本地就尽量用接近生产的数据库镜像

2. 只靠应用层校验唯一性
- 解决：数据库层必须有唯一约束

3. 手动改线上表，不走Alembic
- 解决：所有结构改动都必须迁移脚本化

4. Session生命周期管理混乱
- 解决：统一依赖注入get_db()，请求级关闭

十二、本篇小结
 

这一步的核心是建立一套可持续演进的数据库工程习惯：- 模型统一定义
- 迁移脚本治理
- 会话边界清晰
- 异常语义可控

下一篇我们进入第三阶段：
FastAPI鉴权实战，完整落地JWT与OAuth2，并给出角色权限的工程写法