DevNull 知识库

Appearance

ORM 与迁移 ​

上一篇的 SQL 全是裸字符串——字段名改了要满文件找，查询条件复杂了 SQL 又长又难读，换数据库方言还得重写。ORM 用 Python 类描述表结构、用 Python 方法做查询，把这些问题收走了。

Python 最主流的 ORM 是 SQLAlchemy。配合 Alembic 做数据库迁移——表结构变了（加字段、改类型），用迁移脚本管理变更，不用手动 ALTER TABLE。

一、SQLAlchemy 模型 ​

python
from sqlalchemy import String, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime


class Base(DeclarativeBase):
    pass


class Article(Base):
    __tablename__ = "articles"     # 对应数据库里的表名

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    status: Mapped[str] = mapped_column(String(20), default="draft")
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)

Article 类就是数据库表的描述——每个属性对应一列。Mapped[int] 标注类型，mapped_column 配置列的属性（主键、自增、长度、默认值）。这个类就是上一篇 CREATE TABLE 那段 SQL 的等价物，但它是 Python 代码，IDE 有类型提示，改字段名时能全局搜索。

二、创建引擎和会话 ​

数据库连接通过"引擎"管理。引擎管连接池，会话（Session）管一次数据库操作的上下文：

python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 连接 SQLite（换成 MySQL 只改连接字符串）
engine = create_engine("sqlite:///articles.db", echo=False)
SessionLocal = sessionmaker(bind=engine)

# 根据模型创建所有表（开发时用，生产用迁移）
Base.metadata.create_all(engine)

create_engine 的连接字符串决定连什么数据库：

  • SQLite：sqlite:///articles.db
  • MySQL：mysql+pymysql://user:password@localhost:3306/dbname
  • PostgreSQL：postgresql://user:password@localhost:5432/dbname

echo=True 会把生成的 SQL 打到控制台——学 ORM 时开一下，能看到每个操作背后生成了什么 SQL，对照着理解很有帮助。

三、CRUD 操作 ​

python
from sqlalchemy.orm import Session

session = SessionLocal()

1 新增 ​

python
new_article = Article(title="Python 入门", content="基础语法...", status="published")
session.add(new_article)
session.commit()
print(new_article.id)  # commit 后 id 自动填充

session.add() 把对象加入会话，commit() 时生成并执行 INSERT SQL。commit 之后 new_article.id 才有值——自增 ID 是数据库分配的。

2 查询 ​

python
# 查全部
articles = session.query(Article).all()

# 条件查询
published = session.query(Article).filter(Article.status == "published").all()

# 查单条
article = session.query(Article).filter(Article.id == 1).first()

filter(Article.status == "published") 是 SQLAlchemy 的查询语法——Article.status == "published" 在普通 Python 里是比较运算返回 True/False，但在 SQLAlchemy 里它返回一个查询条件对象，被翻译成 WHERE status = 'published'。

3 更新 ​

python
article = session.query(Article).filter(Article.id == 1).first()
article.status = "published"
session.commit()

直接改对象的属性，commit() 时 SQLAlchemy 自动生成 UPDATE SQL。比手写 UPDATE articles SET status = ? WHERE id = ? 简洁得多。

4 删除 ​

python
article = session.query(Article).filter(Article.id == 1).first()
if article:
    session.delete(article)
    session.commit()

5 用完后关闭会话 ​

python
session.close()

四、在 FastAPI 里用会话 ​

跟上一篇的 SQLite 依赖注入一样，用 Depends 管理会话生命周期：

python
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()


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


@app.get("/articles")
def list_articles(db: Session = Depends(get_db)):
    articles = db.query(Article).order_by(Article.id.desc()).all()
    return [{"id": a.id, "title": a.title, "status": a.status} for a in articles]


@app.post("/articles")
def create_article(title: str, content: str, db: Session = Depends(get_db)):
    article = Article(title=title, content=content)
    db.add(article)
    db.commit()
    db.refresh(article)  # 刷新，拿到数据库分配的 id
    return {"id": article.id, "title": article.title}

db.refresh(article) 在 commit 后重新查一次，拿到数据库分配的 id 和默认值。

五、Alembic 迁移 ​

Base.metadata.create_all() 只能建表，不能改已有的表结构。线上数据库已经有数据，直接删表重建会丢数据。加字段、改类型这些表结构变更，用 Alembic 管理。

bash
uv add alembic

初始化：

bash
uv run alembic init alembic

生成的目录结构：

text
alembic/
├── versions/        # 迁移脚本存这里
├── env.py           # 迁移环境配置
└── script.py.mako   # 迁移脚本模板
alembic.ini          # Alembic 配置文件

配置 alembic/env.py，让它知道模型在哪：

python
# alembic/env.py 里，添加模型导入
from main import Base  ① 你的模型定义文件
target_metadata = Base.metadata

配置数据库连接（alembic.ini）：

ini
sqlalchemy.url = sqlite:///articles.db

1 生成迁移脚本 ​

模型加了新字段（比如 summary）后，自动生成迁移脚本：

bash
uv run alembic revision --autogenerate -m "add summary to articles"

Alembic 对比模型和当前数据库的差异，生成一个迁移脚本（versions/xxx_add_summary_to_articles.py）：

python
def upgrade():
    op.add_column("articles", sa.Column("summary", sa.Text(), nullable=True))

def downgrade():
    op.drop_column("articles", "summary")

upgrade 是正向迁移（加字段），downgrade 是回退（删字段）。autogenerate 生成的不一定完全对，复杂变更要人工检查和修改脚本。

2 执行迁移 ​

bash
uv run alembic upgrade head    # 执行所有待执行的迁移
uv run alembic current         # 看当前迁移到哪个版本
uv run alembic downgrade -1    # 回退一个版本

head 表示最新的迁移版本。生产部署时先 alembic upgrade head 再启动应用——这样表结构总是跟代码一致。