如何用原生SQL批量更新旧数据库表结构？

核心思路 #

• 定义基准表结构模板：用结构化方式存储每种表类型的最新结构（字段、约束、索引、默认值）
• 提取现有表结构：通过查询数据库系统表获取旧表的当前结构
• 对比结构差异：自动识别新增/修改/删除的字段、约束、索引，生成对应的ALTER TABLE语句
• 批量执行变更：分批处理大量表，带事务执行，记录操作日志

步骤1：定义基准表结构模板 #

把每种表类型（比如table1）的最新结构用Python字典存储，方便后续对比：

# 存储所有表类型的最新结构
TABLE_TEMPLATES = {
    "table1": {
        "columns": [
            {"name": "id", "type": "varchar(30)", "constraints": ["primary key"]},
            {"name": "name", "type": "varchar(50)", "constraints": ["not null"]},
            {"name": "date", "type": "timestamp with time zone", "constraints": [], "default": "now()"},
            # 新增字段直接添加到这里
            {"name": "status", "type": "int", "constraints": ["not null"], "default": "0"}
        ],
        "indexes": [
            # 示例：新增索引（注意替换实际表名的占位符）
            {"name": "idx_table1_status", "definition": "CREATE INDEX idx_table1_status ON table1_instace1(status)"}
        ]
    }
}

步骤2：提取现有表结构 #

通过查询数据库系统表（以下以PostgreSQL为例），获取指定表的字段、约束、索引信息：

import psycopg2

def get_existing_table_structure(conn, table_name):
    cursor = conn.cursor()
    # 查询字段信息
    cursor.execute("""
        SELECT 
            column_name, 
            data_type, 
            is_nullable, 
            column_default,
            (SELECT string_agg(constraint_type, ', ')
             FROM information_schema.table_constraints tc
             JOIN information_schema.constraint_column_usage ccu
             ON tc.constraint_name = ccu.constraint_name
             WHERE tc.table_name = %s AND ccu.column_name = columns.column_name) AS constraints
        FROM information_schema.columns
        WHERE table_name = %s AND table_schema = 'public'
    """, (table_name, table_name))
    
    columns = []
    for row in cursor.fetchall():
        col_constraints = []
        if row[2] == 'NO':
            col_constraints.append('not null')
        if row[4] and 'PRIMARY KEY' in row[4]:
            col_constraints.append('primary key')
        columns.append({
            "name": row[0],
            "type": row[1],
            "constraints": col_constraints,
            "default": row[3]
        })
    
    # 查询索引信息
    cursor.execute("""
        SELECT indexname, indexdef
        FROM pg_indexes
        WHERE tablename = %s AND schemaname = 'public'
    """, (table_name,))
    indexes = [{"name": row[0], "definition": row[1]} for row in cursor.fetchall()]
    
    cursor.close()
    return {"columns": columns, "indexes": indexes}

步骤3：对比结构生成ALTER语句 #

编写逻辑对比基准模板和现有表的差异，生成对应的变更语句：

def generate_alter_commands(table_name, existing_structure, template_structure):
    alter_commands = []
    existing_cols = {col["name"]: col for col in existing_structure["columns"]}
    template_cols = {col["name"]: col for col in template_structure["columns"]}

    # 1. 新增字段
    for col_name, template_col in template_cols.items():
        if col_name not in existing_cols:
            col_def = f"{col_name} {template_col['type']}"
            if template_col["constraints"]:
                col_def += " " + " ".join(template_col["constraints"])
            if template_col.get("default"):
                col_def += f" default {template_col['default']}"
            alter_commands.append(f"ALTER TABLE {table_name} ADD COLUMN {col_def};")
    
    # 2. 修改字段（类型、约束、默认值）
    for col_name, existing_col in existing_cols.items():
        if col_name in template_cols:
            template_col = template_cols[col_name]
            # 检查字段类型
            if existing_col["type"] != template_col["type"]:
                alter_commands.append(f"ALTER TABLE {table_name} ALTER COLUMN {col_name} TYPE {template_col['type']};")
            # 检查NOT NULL约束
            existing_has_not_null = 'not null' in existing_col["constraints"]
            template_has_not_null = 'not null' in template_col["constraints"]
            if template_has_not_null and not existing_has_not_null:
                alter_commands.append(f"ALTER TABLE {table_name} ALTER COLUMN {col_name} SET NOT NULL;")
            elif not template_has_not_null and existing_has_not_null:
                alter_commands.append(f"ALTER TABLE {table_name} ALTER COLUMN {col_name} DROP NOT NULL;")
            # 检查默认值
            if existing_col["default"] != template_col.get("default"):
                if template_col.get("default") is not None:
                    alter_commands.append(f"ALTER TABLE {table_name} ALTER COLUMN {col_name} SET DEFAULT {template_col['default']};")
                else:
                    alter_commands.append(f"ALTER TABLE {table_name} ALTER COLUMN {col_name} DROP DEFAULT;")
    
    # 3. 处理索引（新增/删除）
    existing_indexes = {idx["name"]: idx for idx in existing_structure["indexes"]}
    template_indexes = {idx["name"]: idx for idx in template_structure.get("indexes", [])}
    # 新增索引
    for idx_name, template_idx in template_indexes.items():
        if idx_name not in existing_indexes:
            # 替换模板中的表名为实际表名
            idx_def = template_idx["definition"].replace("table1_instace1", table_name)
            alter_commands.append(idx_def + ";")
    # 删除索引
    for idx_name in existing_indexes:
        if idx_name not in template_indexes:
            alter_commands.append(f"DROP INDEX {idx_name};")
    
    # 注意：删除字段默认关闭，避免误删数据，需手动开启
    # for col_name in existing_cols:
    #     if col_name not in template_cols:
    #         alter_commands.append(f"ALTER TABLE {table_name} DROP COLUMN {col_name};")
    
    return alter_commands

步骤4：批量执行变更 #

实现分批处理逻辑，带事务执行变更，记录成功/失败日志：

def batch_update_templates(conn, template_name, template_structure):
    cursor = conn.cursor()
    # 获取所有该模板对应的实例表（比如table1_开头的表）
    cursor.execute("""
        SELECT table_name
        FROM information_schema.tables
        WHERE table_name LIKE %s AND table_schema = 'public'
    """, (f"{template_name}_%",))
    tables = [row[0] for row in cursor.fetchall()]
    
    # 分批处理，控制数据库压力
    batch_size = 100
    total_tables = len(tables)
    print(f"Found {total_tables} tables to process")
    
    for batch_idx in range(0, total_tables, batch_size):
        batch = tables[batch_idx:batch_idx+batch_size]
        print(f"Processing batch {batch_idx//batch_size + 1}/{(total_tables + batch_size -1)//batch_size}")
        
        for table in batch:
            try:
                existing_structure = get_existing_table_structure(conn, table)
                alter_cmds = generate_alter_commands(table, existing_structure, template_structure)
                
                if alter_cmds:
                    # 事务包裹，确保变更原子性
                    cursor.execute("BEGIN;")
                    for cmd in alter_cmds:
                        cursor.execute(cmd)
                    cursor.execute("COMMIT;")
                    print(f"✅ Updated: {table}")
                else:
                    print(f"ℹ️ No changes: {table}")
                    
            except Exception as e:
                cursor.execute("ROLLBACK;")
                error_msg = f"❌ Failed to update {table}: {str(e)}"
                print(error_msg)
                # 记录错误日志
                with open("table_update_errors.log", "a", encoding="utf-8") as f:
                    f.write(f"{error_msg}\n")
    
    cursor.close()

# 主程序入口
if __name__ == "__main__":
    # 数据库连接配置
    conn = psycopg2.connect(
        dbname="your_database",
        user="your_user",
        password="your_password",
        host="your_host"
    )
    
    # 执行批量更新
    batch_update_templates(conn, "table1", TABLE_TEMPLATES["table1"])
    
    conn.close()

关键注意事项 #

• 数据库兼容性：上述代码针对PostgreSQL，若使用MySQL，需修改系统表查询语句和ALTER语法（比如修改字段用MODIFY COLUMN）
• 性能优化：根据数据库承载能力调整batch_size，避免一次性操作过多表；可在低峰期执行
• 数据安全：执行前务必备份数据库；删除字段功能默认关闭，需谨慎开启
• 版本控制：将TABLE_TEMPLATES纳入代码版本控制，每次结构变更都更新模板，便于追溯
• 依赖处理：若涉及外键约束，需先处理关联表的变更，避免执行失败

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