Peewee V3.17的migrator.drop_column方法失效，SQLite删列报错

一、解决drop_column语法错误问题 #

核心原因 #

SQLite 3.35.0以下版本原生不支持直接DROP COLUMN，Peewee会通过「创建新表→复制数据→删除旧表→重命名新表」的方式模拟删除，但如果调用方式错误或环境配置问题，会触发语法报错。

修复步骤 #

1. 检查SQLite版本
   执行以下命令查看版本，若低于3.35.0，需依赖Peewee的模拟删除逻辑：

   sqlite3 --version
   

2. 正确使用migrator API
   确保迁移逻辑通过playhouse.migrate的标准流程执行，示例代码：

   from playhouse.migrate import SqliteMigrator, migrate
   from your_app.models import db  # 导入你的SQLite数据库实例
   
   # 初始化SQLite迁移器
   migrator = SqliteMigrator(db)
   
   def run_drop_column_migration():
       # 执行删除列操作
       migrate(
           migrator.drop_column('target_table', 'column_to_remove'),
       )
   
   # 在原子事务中执行迁移
   with db.atomic():
       run_drop_column_migration()
   

3. 手动模拟删除列（备用方案）
   如果Peewee的自动模拟失效，可手动执行SQL完成删除：

   with db.atomic():
       # 1. 创建不含目标列的新表
       db.execute_sql("CREATE TABLE new_table AS SELECT col1, col2 FROM target_table;")
       # 2. 删除旧表
       db.execute_sql("DROP TABLE target_table;")
       # 3. 重命名新表为原表名
       db.execute_sql("ALTER TABLE new_table RENAME TO target_table;")
   

4. 验证表名与列名
   确保传入drop_column的表名、列名与数据库中实际名称完全一致（SQLite默认大小写不敏感，但需避免拼写错误）。

二、Peewee迁移目录结构最佳实践 #

Peewee没有像Django那样强制的目录规范，推荐以下结构来管理迁移：

推荐目录结构 #

your_project/
├── app/
│   ├── models.py       # Peewee模型定义文件
│   └── main.py         # FastAPI主入口
├── migrations/
│   ├── __init__.py
│   ├── 001_initial.py  # 初始迁移（创建表）
│   ├── 002_add_col.py  # 添加列的迁移
│   └── 003_drop_col.py # 删除列的迁移
├── migrate.py          # 迁移执行脚本
└── requirements.txt

迁移文件规范 #

每个迁移文件需包含一个migrate函数，接收migrator、database等参数，示例003_drop_col.py：

def migrate(migrator, database, fake=False, **kwargs):
    # 执行删除列操作
    migrator.drop_column('user', 'unused_column')

迁移执行脚本migrate.py示例 #

该脚本负责跟踪已执行的迁移、按顺序执行新迁移：

import os
from playhouse.migrate import SqliteMigrator
from app.models import db

# 初始化迁移器
migrator = SqliteMigrator(db)

# 迁移历史表，用于记录已执行的迁移
MIGRATION_HISTORY = 'migration_history'

def init_history_table():
    if not db.table_exists(MIGRATION_HISTORY):
        db.execute_sql(f"""
            CREATE TABLE {MIGRATION_HISTORY} (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                migration_name TEXT UNIQUE NOT NULL,
                executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

def get_executed_migrations():
    cursor = db.execute_sql(f"SELECT migration_name FROM {MIGRATION_HISTORY}")
    return {row[0] for row in cursor.fetchall()}

def run_all_migrations():
    init_history_table()
    executed = get_executed_migrations()

    # 按文件名排序执行迁移
    migration_files = sorted(f for f in os.listdir('migrations') 
                             if f.endswith('.py') and not f.startswith('__'))
    
    for file in migration_files:
        migration_name = file[:-3]
        if migration_name not in executed:
            print(f"Executing migration: {migration_name}")
            # 导入迁移模块并执行
            module = __import__(f"migrations.{migration_name}", fromlist=['migrate'])
            module.migrate(migrator, db)
            # 记录到历史表
            db.execute_sql(f"""
                INSERT INTO {MIGRATION_HISTORY} (migration_name) VALUES (?)
            """, (migration_name,))
            print(f"Migration {migration_name} done")

if __name__ == '__main__':
    run_all_migrations()

执行迁移方式 #

直接运行迁移脚本即可，无需通过FastAPI启动：

python migrate.py

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