SQLAlchemy仅返回含回车行的最后一行（SQL Server）问题解决

我编写了一条SQL查询，用于基于现有Schema生成表的创建DDL语句，以此实现Schema备份（因无SQL Server常规操作权限）。在SSMS中执行该查询时，能正常返回**单行单列（表头为"DDL"）**的完整DDL内容，但通过SQLAlchemy执行时，仅返回最后一行内容。

示例SQL代码：

SET NOCOUNT OFF;
IF OBJECT_ID('[<schema_name>].[<table_name>]','SN') IS NOT NULL DROP SYNONYM [<schema_name>].[<table_name>];
IF OBJECT_ID('[<schema_name>].[<table_name>]','U') IS NOT NULL DROP TABLE [<schema_name>].[<table_name>];
RAISERROR('CREATE TABLE %s',10,1, '[<schema_name>].[<table_name>]') WITH NOWAIT;

CREATE TABLE [<schema_name>].[<table_name>] (
 ... column definitions ...
)  
RAISERROR('CREATING INDEXES OF  %s',10,1, '[<schema_name>].[<table_name>]') WITH NOWAIT; 

对应的Python代码：

def get_create_ddls(table_name_list: list, conn) -> list:
    fd = open('CreateCreateDDL.sql', 'r')
    original_query = fd.read()
    fd.close()

    return_list = []
    for name in table_name_list:
        # prefixed the views as "V_"
        isView = name[:2] == "V_"

        if isView:
            tbl_insert_str = "('" + name[2:] + "')"
        else:
            tbl_insert_str = "('" + name + "')"

        query = original_query.replace("<INSERT_TABLE_NAME_HERE>", tbl_insert_str, 1)

        result = conn.execute(sqlalchemy.text(query))

        # TODO: Only reading last line of the row
        value_str = ""
        for row in result:
            temp_value = row.DDL 
            value_str += temp_value

        print(value_str)

        if isView:
            # put the "V_" back into the CREATE statement
            view_name = name[2:(name.find("].[") + 3)] + "V_" + name[(name.find("].[") + 3):]
            value_str = value_str.replace(name[2:], view_name)

        return_list.append(value_str)

    return return_list

请问如何无需手动删除源查询中的所有回车来解决该问题？

方法1：修改SQL，用单个SELECT返回完整DDL #

核心思路是将所有DDL片段拼接成一个完整的字符串，通过单个SELECT语句返回，避免多个结果集或消息干扰。

针对SQL Server 2017+（支持STRING_AGG） #

SELECT STRING_AGG(ddl_segment, CHAR(13) + CHAR(10)) AS DDL
FROM (
    -- 生成表创建语句头部
    SELECT 'CREATE TABLE [' + s.name + '].[' + t.name + '] (' AS ddl_segment
    FROM sys.tables t
    JOIN sys.schemas s ON t.schema_id = s.schema_id
    WHERE t.name = '<table_name>' AND s.name = '<schema_name>'

    UNION ALL

    -- 生成列定义
    SELECT '    [' + c.name + '] ' + tp.name + 
           CASE WHEN tp.name IN ('varchar', 'nvarchar', 'char', 'nchar') THEN '(' + CASE WHEN c.max_length = -1 THEN 'MAX' ELSE CAST(c.max_length AS VARCHAR) END + ')' ELSE '' END +
           CASE WHEN c.is_nullable = 0 THEN ' NOT NULL' ELSE ' NULL' END + ','
    FROM sys.columns c
    JOIN sys.tables t ON c.object_id = t.object_id
    JOIN sys.schemas s ON t.schema_id = s.schema_id
    JOIN sys.types tp ON c.system_type_id = tp.system_type_id
    WHERE t.name = '<table_name>' AND s.name = '<schema_name>'

    UNION ALL

    -- 生成表创建语句尾部
    SELECT ') ON [PRIMARY]' AS ddl_segment

    UNION ALL

    -- 生成索引语句
    SELECT 'CREATE ' + CASE WHEN i.is_unique = 1 THEN 'UNIQUE ' ELSE '' END + 'NONCLUSTERED INDEX [' + i.name + '] ON [' + s.name + '].[' + t.name + '] (' +
           STRING_AGG('[' + c.name + '] ' + CASE WHEN ic.is_descending_key = 1 THEN 'DESC' ELSE 'ASC' END, ', ') + ')'
    FROM sys.indexes i
    JOIN sys.tables t ON i.object_id = t.object_id
    JOIN sys.schemas s ON t.schema_id = s.schema_id
    JOIN sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id
    JOIN sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
    WHERE t.name = '<table_name>' AND s.name = '<schema_name>' AND i.type_desc = 'NONCLUSTERED'
    GROUP BY i.name, s.name, t.name, i.is_unique
) AS ddl_parts

针对SQL Server 2016及更早版本（用FOR XML PATH拼接） #

SELECT 
    STUFF((
        SELECT CHAR(13) + CHAR(10) + ddl_segment
        FROM (
            -- 表头部
            SELECT 'CREATE TABLE [' + s.name + '].[' + t.name + '] (' AS ddl_segment
            FROM sys.tables t
            JOIN sys.schemas s ON t.schema_id = s.schema_id
            WHERE t.name = '<table_name>' AND s.name = '<schema_name>'

            UNION ALL

            -- 列定义
            SELECT '    [' + c.name + '] ' + tp.name + 
                   CASE WHEN tp.name IN ('varchar', 'nvarchar', 'char', 'nchar') THEN '(' + CASE WHEN c.max_length = -1 THEN 'MAX' ELSE CAST(c.max_length AS VARCHAR) END + ')' ELSE '' END +
                   CASE WHEN c.is_nullable = 0 THEN ' NOT NULL' ELSE ' NULL' END + ','
            FROM sys.columns c
            JOIN sys.tables t ON c.object_id = t.object_id
            JOIN sys.schemas s ON t.schema_id = s.schema_id
            JOIN sys.types tp ON c.system_type_id = tp.system_type_id
            WHERE t.name = '<table_name>' AND s.name = '<schema_name>'

            UNION ALL

            -- 表尾部
            SELECT ') ON [PRIMARY]' AS ddl_segment

            UNION ALL

            -- 索引
            SELECT 'CREATE ' + CASE WHEN i.is_unique = 1 THEN 'UNIQUE ' ELSE '' END + 'NONCLUSTERED INDEX [' + i.name + '] ON [' + s.name + '].[' + t.name + '] (' +
                   STUFF((
                       SELECT ', [' + c2.name + '] ' + CASE WHEN ic2.is_descending_key = 1 THEN 'DESC' ELSE 'ASC' END
                       FROM sys.index_columns ic2
                       JOIN sys.columns c2 ON ic2.object_id = c2.object_id AND ic2.column_id = c2.column_id
                       WHERE ic2.object_id = i.object_id AND ic2.index_id = i.index_id
                       FOR XML PATH(''), TYPE
                   ).value('.', 'NVARCHAR(MAX)'), 1, 2, '') + ')'
            FROM sys.indexes i
            JOIN sys.tables t ON i.object_id = t.object_id
            JOIN sys.schemas s ON t.schema_id = s.schema_id
            WHERE t.name = '<table_name>' AND s.name = '<schema_name>' AND i.type_desc = 'NONCLUSTERED'
        ) AS parts
        FOR XML PATH(''), TYPE
    ).value('.', 'NVARCHAR(MAX)'), 1, 2, '') AS DDL

这样修改后，SQL会返回一个包含完整DDL的单行结果，SQLAlchemy只需读取这一行即可，无需处理多个结果集。

方法2：修改Python代码，遍历所有结果集 #

如果无法修改SQL，可调整Python代码，让SQLAlchemy遍历所有结果集，收集所有包含DDL列的内容：

def get_create_ddls(table_name_list: list, conn) -> list:
    fd = open('CreateCreateDDL.sql', 'r')
    original_query = fd.read()
    fd.close()

    return_list = []
    for name in table_name_list:
        isView = name[:2] == "V_"

        if isView:
            tbl_insert_str = "('" + name[2:] + "')"
        else:
            tbl_insert_str = "('" + name + "')"

        query = original_query.replace("<INSERT_TABLE_NAME_HERE>", tbl_insert_str, 1)

        result = conn.execute(sqlalchemy.text(query))
        value_str = ""
        
        # 遍历所有结果集
        current_result = result
        while current_result is not None:
            try:
                # 读取当前结果集的所有行
                for row in current_result:
                    if hasattr(row, 'DDL'):
                        value_str += row.DDL
            except Exception:
                # 跳过无DDL列的结果集（如DROP语句的影响行数）
                pass
            # 获取下一个结果集
            current_result = current_result.nextset()

        print(value_str)

        if isView:
            view_name = name[2:(name.find("].[") + 3)] + "V_" + name[(name.find("].[") + 3):]
            value_str = value_str.replace(name[2:], view_name)

        return_list.append(value_str)

    return return_list

这段代码会逐个处理SQL执行产生的所有结果集，收集所有包含DDL字段的内容，避免只获取最后一个结果集的问题。

方法3：调整SQL的消息输出逻辑 #

如果SQL中的RAISERROR只是用于进度提示，而非DDL内容的一部分，可将这些语句的级别调整为不干扰结果集的形式，或改用PRINT语句（SQLAlchemy默认不捕获PRINT消息，不会影响结果集读取）。同时确保核心DDL内容通过单个SELECT语句返回。

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