Alembic + SQLite 踩坑

Stankle of QuadnucYardOriginal...About 7 min

Alembic + SQLite 踩坑

用 SQLAlchemy 写 ORM 时经常需要修改数据表字段，这时一般方法是手动改数据表，还可能要处理新增的 not null 字段的默认值，比较麻烦。
因此我们选择 alembic 完成数据库迁移操作，快捷更新表结构。

操作

官网有详细教程，这里说明一下我的操作过程，列一下遇到的坑。

创建环境

假设已经用 poetry 创建好了虚拟环境。

poetry add alembic

然后运行 poetry shell 进入虚拟环境，再输入 alembic init <alembic_directory>，如 alembic init alembic。
会自动创建如下文件：

├── alembic
│   └── env.py
└── alembic.ini

如果用的是 async sqlite，需要加额外的参数，不然迁移脚本不会用到 asyncio，导致执行失败：

alembic init -t async <script_directory_here>

编辑文件

在 alembic.ini 主要修改一个参数：

sqlalchemy.url = driver://user:pass@localhost/dbname  # 改成数据库url

如果用了 ruff，取消注释下面内容：

hooks = ruff
ruff.type = exec
ruff.executable = %(here)s/.venv/bin/ruff
ruff.options = --fix REVISION_SCRIPT_FILENAME

然后在 env.py 导入模型：

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
from app.models import Base
target_metadata = Base.metadata

注意导入路径，若有需要可以修改 sys.path。

创建迁移脚本

alembic revision --autogenerate -m "Initial migration"

然后会在 alembic/versions/ 下生成对应的脚本。

打开 xxxx_initial_migration.py（文件名和 hash、message 有关），如果没有建表或者 ORM 跟数据库完全一致，那么 upgrade 和 downgrade 函数将是空的：

revision: str = '2402cd81dacf'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

def upgrade():
    pass

def downgrade():
    pass

但是我已经有数据表了，而且由于之前是手动迁移，存在不一致的地方（null constraint），所以实际内容是：

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('food', 'images',
               existing_type=sa.VARCHAR(),
               nullable=False)
    op.alter_column('food', 'update_time',
               existing_type=sa.DATETIME(),
               nullable=False)
    op.create_foreign_key(None, 'food', 'food', ['parent_id'], ['id'])
    op.alter_column('weigh', 'update_time',
               existing_type=sa.DATETIME(),
               nullable=False)
    # ### end Alembic commands ###

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('weigh', 'update_time',
               existing_type=sa.DATETIME(),
               nullable=True)
    op.drop_constraint(None, 'food', type_='foreignkey')
    op.alter_column('food', 'update_time',
               existing_type=sa.DATETIME(),
               nullable=True)
    op.alter_column('food', 'images',
               existing_type=sa.VARCHAR(),
               nullable=True)
    # ### end Alembic commands ###

虽然内容是自动生成的，但实际上是可以手动改的，而且有些问题必须手动解决。

初次迁移

在对模型进行更改（或者上面在初始化前已经有了更改）后，执行升级：

alembic upgrade head

结果没那么顺利，遇到了报错：

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "ALTER": syntax error
[SQL: ALTER TABLE food ALTER COLUMN update_time SET NOT NULL]
(Background on this error at: https://sqlalche.me/e/20/e3q8)

为什么呢？因为 SQLite 天生不支持修改列属性，要不然 DBMS 为啥不让改……

注释掉改 NOT NULL 的操作，重新执行，得到新的报错：

NotImplementedError: No support for ALTER of constraints in SQLite dialect. Please refer to the batch mode feature which
allows for SQLite migrations using a copy-and-move strategy.

你就说啥都做不了呗。

根据提示，修改 env.py 内容：

def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
    # 加上 render_as_batch=True

但是还没有用。

最后还是参考文档 https://alembic.sqlalchemy.org/en/latest/batch.html，手动使用 batch_op。本质上就是根据操作新建一张临时表，然后替换掉原表。

修改后代码如下（不需要再设 render_as_batch）：

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table("food") as batch_op:
        batch_op.alter_column("images", existing_type=sa.VARCHAR(), nullable=False)
        batch_op.alter_column("update_time", existing_type=sa.DATETIME(), nullable=False)
        batch_op.create_foreign_key("fk_food_parent", "food", ["parent_id"], ["id"])

    with op.batch_alter_table("weigh") as batch_op:
        batch_op.alter_column("update_time", existing_type=sa.DATETIME(), nullable=False)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table("weigh") as batch_op:
        batch_op.alter_column("update_time", existing_type=sa.DATETIME(), nullable=True)

    with op.batch_alter_table("food") as batch_op:
        batch_op.drop_constraint("fk_food_parent", type_="foreignkey")
        batch_op.alter_column("update_time", existing_type=sa.DATETIME(), nullable=True)
        batch_op.alter_column("images", existing_type=sa.VARCHAR(), nullable=True)
    # ### end Alembic commands ###

注意，自动生成的代码没有给 FK constraint 命名，这是不允许的（类型提示会爆红），需要手动设一个。

为了保证健壮性，需要检验版本回退正常：

alembic downgrade -1

可能的输出：

(meal-tools-backend-py3.11) $ alembic downgrade -1
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running downgrade 2402cd81dacf -> , Initial migration
(meal-tools-backend-py3.11) $ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 2402cd81dacf, Initial migration

更新数据表

$ alembic revision --autogenerate -m "Add tags"
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'tag'
INFO  [alembic.autogenerate.compare] Detected added table 'food_tag_link'
Generating E:\Project\meal-tools-backend\alembic\versions\afe80fda900d_add_tags.py ...  done

如果不加 --autogenerate 就不会自动生成 upgrade 和 downgrade 函数中的内容。

自动生成代码如下。

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('tag',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(), nullable=False),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('food_tag_link',
    sa.Column('food_id', sa.Integer(), nullable=False),
    sa.Column('tag_id', sa.Integer(), nullable=False),
    sa.ForeignKeyConstraint(['food_id'], ['food.id'], ),
    sa.ForeignKeyConstraint(['tag_id'], ['tag.id'], ),
    sa.PrimaryKeyConstraint('food_id', 'tag_id')
    )
    # ### end Alembic commands ###

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('food_tag_link')
    op.drop_table('tag')
    # ### end Alembic commands ###

可以尝试运行。因为不涉及列和约束操作，不需要 batch mode。

$ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 2402cd81dacf -> afe80fda900d, Add tags

降级也是没问题的。

$ alembic downgrade -1
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running downgrade afe80fda900d -> 2402cd81dacf, Add tags

再次迁移

再做一次迁移，增加时间字段，并且给 name 加上 unique 和 index：

class Tag(TimeMixin, Base):
    id: Mapped[int_pk]
    name: Mapped[str] = mapped_column(unique=True, index=True)

    foods: Mapped[list["Food"]] = relationship(secondary=FoodTagLink, back_populates="tags", lazy="raise")

其中 TimeMixin 和 int_pk 的定义是：

int_pk = Annotated[int, mapped_column(primary_key=True)]

class TimeMixin:
    create_time: Mapped[datetime] = mapped_column(default=func.now())
    update_time: Mapped[datetime] = mapped_column(default=func.now(), onupdate=func.now())

执行结果如下：

$ alembic revision --autogenerate -m "Update tag"
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added column 'tag.create_time'
INFO  [alembic.autogenerate.compare] Detected added column 'tag.update_time'
INFO  [alembic.autogenerate.compare] Detected added index ''ix_tag_name'' on '('name',)'
Generating E:\Project\meal-tools-backend\alembic\versions\1dae51f22bb2_update_tag.py ...  done

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('tag', sa.Column('create_time', sa.DateTime(), nullable=False))
    op.add_column('tag', sa.Column('update_time', sa.DateTime(), nullable=False))
    op.create_index(op.f('ix_tag_name'), 'tag', ['name'], unique=True)
    # ### end Alembic commands ###

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_tag_name'), table_name='tag')
    op.drop_column('tag', 'update_time')
    op.drop_column('tag', 'create_time')
    # ### end Alembic commands ###

此处 upgrade 当然是有问题的，因为新增的非空字段缺少默认值。一个解决方法是，先用 null 填上，再批量赋一个初值，最后改成 not null。这时又要用 batch mode 了。代码如下，downgrade 不需要更改。

需要注意，SQLite 获取当前时间用的是 datetime('now')（GMT+0）或 datetime('now', 'localtime')（GMT+8）。例如当前是 16:00，则前者存的时间为 8:00 而后者为 16:00。

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column("tag", sa.Column("create_time", sa.DateTime(), nullable=True))
    op.add_column("tag", sa.Column("update_time", sa.DateTime(), nullable=True))
    op.execute("UPDATE tag SET create_time = datetime('now'), update_time = datetime('now')")
    op.create_index(op.f("ix_tag_name"), "tag", ["name"], unique=True)
    with op.batch_alter_table("tag") as batch_op:
        batch_op.alter_column("create_time", nullable=False)
        batch_op.alter_column("update_time", nullable=False)
    # ### end Alembic commands ###

升级降级结果：

$ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade afe80fda900d -> 1dae51f22bb2, Update tag
$ alembic downgrade -1
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running downgrade 1dae51f22bb2 -> afe80fda900d, Update tag

又一次迁移

这回增加 category 和 color 两个字段，前者可空，后者不可空但有默认值。

class Tag(TimeMixin, Base):
    id: Mapped[int_pk]
    name: Mapped[str] = mapped_column(unique=True, index=True)
    category: Mapped[str | None] = mapped_column(default=None)
    color: Mapped[str] = mapped_column(default="grey")

    foods: Mapped[list["Food"]] = relationship(secondary=FoodTagLink, back_populates="tags", lazy="raise")

生成的代码如下，相比之前的简单不少。

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('tag', sa.Column('category', sa.String(), nullable=True))
    op.add_column('tag', sa.Column('color', sa.String(), nullable=False))
    # ### end Alembic commands ###

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('tag', 'color')
    op.drop_column('tag', 'category')
    # ### end Alembic commands ###

看样子没有问题，对吧？但是运行会报错如下。再仔细看看上面的 upgrade，并没有设置模型里的默认值。

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) Cannot add a NOT NULL column with default value NULL
[SQL: ALTER TABLE tag ADD COLUMN color VARCHAR NOT NULL]
(Background on this error at: https://sqlalche.me/e/20/e3q8)

用 default 是不可的，要用 server_default。因为在调用 op.add_column 时，SQLAlchemy 并不会设置用户希望的默认值，需要让数据库自己来设置 server_default 里的默认值。上面的数据表也需要改成。

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('tag', sa.Column('category', sa.String(), nullable=True))
    op.add_column('tag', sa.Column('color', sa.String(), nullable=False, server_default="grey"))
    # ### end Alembic commands ###