如何用Python+pyodbc实现SQL Server表结构与存储过程的选择性复制？

我正在编写脚本，将SQL Server生产库（prod）中的指定表和存储过程复制到开发库（dev），无需复制整个数据库（敏感数据脱敏已妥善处理）。需求如下：

• 复制指定表的完整库定义/架构，包括所有约束；
• 复制所有存储过程。

目前已通过内置存储过程sp_helptext实现存储过程的复制，代码如下：

prod_cursor.execute("sp_helptext '<procedure-name>'")
rows = prod_cursor.fetchall()
definition = ''.join(row[0] for row in rows)
dev_cursor.execute(definition)
dev_cursor.commit()

我已掌握数据复制方法，但卡在表结构/完整定义的复制上，希望能生成Azure Data Studio中右键表选择“脚本为创建”时输出的文本。希望用Python脚本实现该功能，仅替换指定表和存储过程而非整个实例，请问是否有更优方案或现成工具？

解决方案 #

一、Python脚本实现表结构完整复制 #

要生成类似Azure Data Studio的建表脚本，可通过查询SQL Server系统视图拼接完整表定义（含列、约束、索引等），核心实现思路如下：

1. 获取表列基础定义
   查询sys.tables、sys.columns、sys.types等系统视图，提取列名、数据类型、长度、精度、是否可为空等信息，拼接CREATE TABLE的主体部分。

2. 提取各类约束

• 主键约束：通过sys.key_constraints和sys.index_columns关联获取主键列；
• 外键约束：查询sys.foreign_keys、sys.foreign_key_columns获取关联表与列信息；
• 唯一、默认、检查约束：分别从sys.key_constraints、sys.default_constraints、sys.check_constraints提取完整定义。

3. 拼接完整建表脚本
   以下是可复用的代码片段：

def generate_create_table_script(cursor, table_name):
    # 查询表列信息
    cursor.execute("""
        SELECT 
            c.name AS column_name,
            t.name AS data_type,
            c.max_length,
            c.precision,
            c.scale,
            c.is_nullable
        FROM sys.columns c
        JOIN sys.types t ON c.system_type_id = t.system_type_id
        WHERE c.object_id = OBJECT_ID(?)
        ORDER BY c.column_id
    """, (table_name,))
    columns = cursor.fetchall()
    
    # 拼接列定义
    column_defs = []
    for col in columns:
        col_name = col[0]
        data_type = col[1]
        # 处理带参数的数据类型
        if data_type in ('varchar', 'nvarchar') and col[2] == -1:
            type_str = f"{data_type}(MAX)"
        elif data_type in ('decimal', 'numeric'):
            type_str = f"{data_type}({col[3]}, {col[4]})"
        else:
            type_str = data_type
        # 处理空值属性
        null_str = "NULL" if col[5] else "NOT NULL"
        column_defs.append(f"    {col_name} {type_str} {null_str}")
    
    # 查询主键约束
    cursor.execute("""
        SELECT 
            kc.name AS constraint_name,
            STRING_AGG(c.name, ', ') WITHIN GROUP (ORDER BY ic.key_ordinal) AS pk_columns
        FROM sys.key_constraints kc
        JOIN sys.index_columns ic ON kc.parent_object_id = ic.object_id AND kc.unique_index_id = ic.index_id
        JOIN sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
        WHERE kc.parent_object_id = OBJECT_ID(?) AND kc.type = 'PK'
        GROUP BY kc.name
    """, (table_name,))
    pk_info = cursor.fetchone()
    pk_def = f"\n    CONSTRAINT {pk_info[0]} PRIMARY KEY ({pk_info[1]})" if pk_info else ""
    
    # 拼接完整建表语句
    create_script = f"CREATE TABLE {table_name}\n(\n" + ",\n".join(column_defs) + pk_def + "\n)"
    
    # 可按需扩展：补充外键、检查、默认约束的查询与拼接逻辑
    # ...
    
    return create_script

生成脚本后直接在dev库执行：

create_script = generate_create_table_script(prod_cursor, "your_target_table")
dev_cursor.execute(create_script)
dev_cursor.commit()

二、存储过程复制的优化方案 #

sp_helptext可能出现格式错乱或换行丢失的问题，改用查询sys.sql_modules直接获取完整定义更可靠：

def get_procedure_definition(cursor, proc_name):
    cursor.execute("""
        SELECT definition FROM sys.sql_modules 
        WHERE object_id = OBJECT_ID(?)
    """, (proc_name,))
    return cursor.fetchone()[0]

# 使用示例
proc_def = get_procedure_definition(prod_cursor, "your_procedure")
dev_cursor.execute(proc_def)
dev_cursor.commit()

三、现成工具推荐 #

若不想手动编写脚本，可选择以下工具：

• SSMS/Azure Data Studio：通过「生成脚本」向导，精准选择指定表和存储过程，生成脚本后在dev库执行；
• dbForge Schema Compare for SQL Server：可视化对比并同步指定对象的架构，适合频繁同步场景；
• SQL Server 内置生成脚本功能：通过T-SQL的Generate Scripts向导，命令行或GUI均可操作。

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