创建审计表存储过程需求：同步原表结构并新增控制列

核心逻辑 #

针对你的需求，核心是对比原表与审计表的列结构：

• 审计表不存在时：复制原表空结构，追加审计控制列
• 审计表已存在时：找出原表新增的列，批量添加到审计表中

实现步骤（以SQL Server为例，用pyodbc驱动） #

以下是可直接复用的代码框架，你可以根据自己的数据库类型调整系统表查询语句：

1. 数据库连接与基础函数

先封装获取表列、判断表存在的通用函数：

import pyodbc

def get_db_connection():
    # 替换为你的数据库连接信息
    conn_str = (
        "DRIVER={ODBC Driver 17 for SQL Server};"
        "SERVER=your_server_name;"
        "DATABASE=your_db_name;"
        "UID=your_username;"
        "PWD=your_password;"
    )
    return pyodbc.connect(conn_str)

def get_table_columns(conn, table_name):
    """获取指定表的列名及完整DDL定义"""
    cursor = conn.cursor()
    cursor.execute(f"""
        SELECT 
            c.name AS column_name,
            t.name AS data_type,
            c.is_nullable,
            c.max_length,
            c.precision,
            c.scale
        FROM sys.columns c
        JOIN sys.types t ON c.system_type_id = t.system_type_id
        WHERE c.object_id = OBJECT_ID('{table_name}')
        ORDER BY c.column_id
    """)
    columns = []
    for row in cursor.fetchall():
        # 拼接列的完整定义，兼容常见数据类型
        col_def = f"[{row.column_name}] {row.data_type}"
        if row.data_type in ('varchar', 'nvarchar', 'char', 'nchar'):
            col_def += f"({row.max_length if row.max_length != -1 else 'MAX'})"
        elif row.data_type in ('decimal', 'numeric'):
            col_def += f"({row.precision}, {row.scale})"
        col_def += " NULL" if row.is_nullable else " NOT NULL"
        columns.append((row.column_name, col_def))
    cursor.close()
    return columns

def table_exists(conn, table_name):
    """判断指定表是否存在"""
    cursor = conn.cursor()
    cursor.execute(f"""
        SELECT 1 FROM sys.tables WHERE name = '{table_name}'
    """)
    exists = cursor.fetchone() is not None
    cursor.close()
    return exists

2. 审计表创建与列同步主函数

def sync_audit_table(source_table, audit_table):
    conn = get_db_connection()
    try:
        source_columns = get_table_columns(conn, source_table)
        source_col_names = [col[0] for col in source_columns]
        
        if not table_exists(conn, audit_table):
            # 1. 审计表不存在：复制原表空结构
            conn.execute(f"SELECT TOP 0 * INTO [{audit_table}] FROM [{source_table}]")
            # 2. 添加审计控制列
            control_columns = [
                "[AudiEstado] BIT NOT NULL DEFAULT 1",
                "[AuditDMLAccion] VARCHAR(10) NOT NULL",
                "[AuditUsuario] VARCHAR(50) NOT NULL",
                "[AuditFecha] DATETIME NOT NULL DEFAULT GETDATE()"
            ]
            for col_def in control_columns:
                conn.execute(f"ALTER TABLE [{audit_table}] ADD {col_def}")
            print(f"审计表 {audit_table} 已创建并添加控制列")
        else:
            # 获取审计表现有列
            audit_columns = get_table_columns(conn, audit_table)
            audit_col_names = [col[0] for col in audit_columns]
            # 排除审计控制列，找出原表新增的列
            control_col_names = ["AudiEstado", "AuditDMLAccion", "AuditUsuario", "AuditFecha"]
            new_columns = [col for col in source_columns if col[0] not in audit_col_names and col[0] not in control_col_names]
            
            if new_columns:
                # 批量添加新增列
                for col_name, col_def in new_columns:
                    conn.execute(f"ALTER TABLE [{audit_table}] ADD {col_def}")
                    print(f"已为审计表 {audit_table} 添加列：{col_name}")
            else:
                print(f"审计表 {audit_table} 与原表 {source_table} 列结构一致，无需更新")
        
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

# 调用示例
sync_audit_table("原表名", "审计表名")

关键细节说明 #

• 数据库适配：如果使用PostgreSQL、MySQL等其他数据库，需要替换get_table_columns和table_exists中的查询语句（比如改用information_schema.columns）
• 特殊类型兼容：代码已处理常见数据类型，若涉及XML、自定义类型等特殊字段，需补充对应的DDL拼接逻辑
• 事务保障：所有结构修改操作包裹在事务中，避免出现表结构不一致的情况

内容的提问来源于stack exchange，提问作者Juan Jose Zuniga