搞定这个问题的核心是利用SQLite的事务支持，把所有列转换操作包裹在一个原子事务里——这样任何步骤出错都会回滚，绝不会留下半完成的表（既有decimal列又有integer列的尴尬状态）。结合你已经知道的SQLite删列方法，下面是完整的迁移方案：

核心思路 #

因为SQLite对ALTER TABLE的限制，我们不能直接修改列类型，得用「添加临时列→迁移数据→切换列名→清理旧列」的流程，而且所有步骤必须在同一个事务中执行。

完整的升级脚本示例 #

假设你的表叫products，要转换的列是price（原类型DECIMAL）：

from alembic import op
import sqlalchemy as sa

def upgrade():
    # 用autocommit_block确保所有操作在一个事务中，出错自动回滚
    with op.get_context().autocommit_block():
        # 1. 新增临时integer列
        op.add_column('products', sa.Column('temp_price', sa.Integer(), nullable=True))
        
        # 2. 迁移数据：把decimal转成integer（根据需求选截断/四舍五入，这里用CAST截断）
        op.execute("""
            UPDATE products 
            SET temp_price = CAST(price AS INTEGER)
            WHERE price IS NOT NULL
        """)
        
        # 3. 切换列名（SQLite 3.25.0+支持RENAME COLUMN，旧版本看下面的替代方案）
        op.execute("ALTER TABLE products RENAME COLUMN price TO old_price")
        op.execute("ALTER TABLE products RENAME COLUMN temp_price TO price")
        
        # 4. 恢复原列的NOT NULL约束（如果原列是NOT NULL的话）
        op.alter_column('products', 'price', nullable=False)
        
        # 5. 安全删除旧列（用创建新表的方式，兼容所有SQLite版本）
        # 注意：要复制原表的所有其他列和约束！
        op.execute("""
            CREATE TABLE products_new (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                price INTEGER NOT NULL,
                name TEXT NOT NULL,
                created_at DATETIME NOT NULL
                -- 这里把原表的其他列都列出来
            )
        """)
        # 复制数据
        op.execute("""
            INSERT INTO products_new (id, price, name, created_at)
            SELECT id, price, name, created_at FROM products
        """)
        # 替换旧表
        op.execute("DROP TABLE products")
        op.execute("ALTER TABLE products_new RENAME TO products")

def downgrade():
    # 降级脚本也要保持事务安全，确保能回滚到原状态
    with op.get_context().autocommit_block():
        op.add_column('products', sa.Column('temp_price', sa.Numeric(), nullable=True))
        
        op.execute("""
            UPDATE products 
            SET temp_price = CAST(price AS NUMERIC)
            WHERE price IS NOT NULL
        """)
        
        op.execute("ALTER TABLE products RENAME COLUMN price TO new_price")
        op.execute("ALTER TABLE products RENAME COLUMN temp_price TO price")
        
        op.alter_column('products', 'price', nullable=False)
        
        # 同样用新表替换的方式清理临时列
        op.execute("""
            CREATE TABLE products_new (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                price NUMERIC NOT NULL,
                name TEXT NOT NULL,
                created_at DATETIME NOT NULL
            )
        """)
        op.execute("""
            INSERT INTO products_new (id, price, name, created_at)
            SELECT id, price, name, created_at FROM products
        """)
        op.execute("DROP TABLE products")
        op.execute("ALTER TABLE products_new RENAME TO products")

关键注意事项 #

1. 事务原子性：autocommit_block()会把所有操作包裹在一个事务里，任何一步失败都会自动回滚，彻底避免半完成状态。
2. SQLite版本兼容：
   • 如果你的SQLite版本≥3.25.0，可以直接用RENAME COLUMN简化步骤；
   • 如果是旧版本，跳过重命名步骤，直接在创建新表时把临时列作为最终列，一次性替换旧表。
3. 数据转换逻辑：CAST(price AS INTEGER)会直接截断小数部分，如果你需要四舍五入，可以改成ROUND(price)再转成整数。
4. 备份优先：执行迁移前一定要备份数据库，哪怕是测试环境——数据安全永远是第一位的。
5. 约束同步：创建新表时一定要复制原表的所有约束（主键、外键、唯一约束等），不然会丢失表结构信息。

简化版（适用于SQLite≥3.35.0） #

如果你的SQLite版本≥3.35.0，官方已经支持DROP COLUMN，那删除旧列的步骤可以简化成直接调用op.drop_column，不用折腾新表替换：

def upgrade():
    with op.get_context().autocommit_block():
        op.add_column('products', sa.Column('temp_price', sa.Integer(), nullable=True))
        op.execute("UPDATE products SET temp_price = CAST(price AS INTEGER) WHERE price IS NOT NULL")
        op.execute("ALTER TABLE products RENAME COLUMN price TO old_price")
        op.execute("ALTER TABLE products RENAME COLUMN temp_price TO price")
        op.alter_column('products', 'price', nullable=False)
        op.drop_column('products', 'old_price')  # 直接删列，SQLite 3.35.0+支持

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