执行flask db migrate遇问题：NullType报错与旧版本文件被修改

问题1：AttributeError: module 'sqlalchemy' has no attribute 'NullType' #

错误信息 #

File "/Users/user1/Development/projects/zen/zen_api/zen_api/migrations/versions/ab4404d743ac_.py", line 24, in upgrade
    sa.Column("expires_at", sa.NullType(), nullable=False),
                            ^^^^^^^^^^^
AttributeError: module 'sqlalchemy' has no attribute 'NullType'. Did you mean: 'TupleType'?

原因 #

SQLAlchemy 2.0及以上版本已移除NullType，迁移文件中出现该类型，是因为对应的模型字段类型未正确定义（如字段类型缺失或错误），导致Alembic自动生成了无效的字段类型。

解决步骤 #

1. 修正模型字段类型：检查RefreshTokenBlocklist模型，确保expires_at和refresh_token使用正确类型：
   • expires_at 对应db.DateTime（数据库datetime类型）
   • refresh_token 对应db.String（数据库varchar/text类型）
2. 手动修复迁移文件：打开ab4404d743ac_.py，替换错误的字段类型：

   # 替换前
   sa.Column("expires_at", sa.NullType(), nullable=False),
   sa.Column("refresh_token", sa.NullType(), nullable=False),
   
   # 替换后
   sa.Column("expires_at", sa.DateTime(), nullable=False),
   sa.Column("refresh_token", sa.String(length=512), nullable=False),
   

3. 重新执行迁移：修复后运行flask db upgrade即可正常执行。

问题2：执行 flask db migrate 时修改旧迁移文件而非生成新文件 #

原因 #

核心是迁移历史与数据库实际状态不一致，常见场景：

• 已应用的迁移文件被手动修改过
• 数据库未同步到最新的迁移版本
• 迁移版本的依赖关系（down_revision）出错

解决步骤 #

1. 确认当前数据库迁移版本：执行命令查看已应用的最新版本：

   flask db current
   

2. 根据状态处理：
   • 若数据库未应用出错的旧迁移（ab4404d743ac）：直接删除该文件，重新执行flask db migrate生成新迁移。
   • 若数据库已应用该迁移：先标记数据库为正确版本，再重新生成迁移：

     # 将数据库标记为初始版本 fbaa8bda9a8b
     flask db stamp fbaa8bda9a8b
     # 生成新迁移
     flask db migrate -m "Add refresh_token_blocklist table"
     

3. 规范操作：后续禁止手动修改已应用的迁移文件，保持迁移历史一致性。

相关文件与依赖版本 #

出错的迁移文件（ab4404d743ac）

"""empty message

Revision ID: ab4404d743ac
Revises: fbaa8bda9a8b
Create Date: 2023-04-26 00:00:29.089871

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = "ab4404d743ac"
down_revision = "fbaa8bda9a8b"
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "refresh_token_blocklist",
        sa.Column("jti", sa.UUID(), nullable=False),
        sa.Column("expires_at", sa.NullType(), nullable=False),
        sa.Column("refresh_token", sa.NullType(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint("jti"),
    )
    with op.batch_alter_table("refresh_token_blocklist", schema=None) as batch_op:
        batch_op.create_index(
            batch_op.f("ix_refresh_token_blocklist_jti"), ["jti"], unique=False
        )

    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table("refresh_token_blocklist", schema=None) as batch_op:
        batch_op.drop_index(batch_op.f("ix_refresh_token_blocklist_jti"))

    op.drop_table("refresh_token_blocklist")
    # ### end Alembic commands ###

初始迁移文件（fbaa8bda9a8b）

"""empty message

Revision ID: fbaa8bda9a8b
Revises: 
Create Date: 2023-04-23 01:07:46.538734

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = "fbaa8bda9a8b"
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "property",
        sa.Column("id", sa.UUID(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_table(
        "user",
        sa.Column("id", sa.UUID(), nullable=False),
        sa.Column("first_name", sa.String(length=128), nullable=False),
        sa.Column("last_name", sa.String(length=128), nullable=False),
        sa.Column("email", sa.String(length=128), nullable=False),
        sa.Column("password_hash", sa.String(length=128), nullable=True),
        sa.Column("phone_number", sa.String(length=10), nullable=True),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    with op.batch_alter_table("user", schema=None) as batch_op:
        batch_op.create_index(batch_op.f("ix_user_email"), ["email"], unique=True)

    op.create_table(
        "review",
        sa.Column("id", sa.UUID(), nullable=False),
        sa.Column("user_id", sa.UUID(), nullable=True),
        sa.Column("rating", sa.Integer(), nullable=True),
        sa.Column("comment", sa.Text(), nullable=True),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(
            ["user_id"],
            ["user.id"],
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table("review")
    with op.batch_alter_table("user", schema=None) as batch_op:
        batch_op.drop_index(batch_op.f("ix_user_email"))

    op.drop_table("user")
    op.drop_table("property")
    # ### end Alembic commands ###

使用的依赖版本

flask = "^2.2.3"
flask-sqlalchemy = "^3.0.3"
flask-migrate = "^4.0.4"

内容的提问来源于stack exchange，提问作者Liondancer