Python+pyodbc拼接含动态列的UNION查询SQL实现方法

你现在拼不出完整SQL的核心原因是把外层UNION块生成和动态列生成拆成了两个独立的遍历流程，输出的片段是散的，没法按SurveyId一一对应。调整逻辑如下：

• 把动态答案列的生成逻辑直接内嵌到遍历所有SurveyId的循环内部，处理单个问卷时一次性生成该问卷对应的完整SELECT语句块，避免片段错位
• 每处理一个SurveyId，先用临时列表存当前问卷对应的所有答案列片段，不要直接打印输出
• 单问卷的列全部生成后，直接替换外层SELECT语句里的占位位置，拼成完整的单问卷查询块，存入总列表
• 所有问卷的查询块生成完成后，用UNION关键字把所有块拼接成最终可执行的完整SQL

修正后代码 #

# 假设cursor为已初始化的pyodbc游标对象
# 1. 获取所有需要查询的问卷ID
allSurveyID = cursor.execute("SELECT DISTINCT SurveyId FROM Survey ORDER BY SurveyId").fetchall()

# 2. 存储每个问卷对应的独立SELECT查询块
union_query_blocks = []

for survey_row in allSurveyID:
    current_survey_id = survey_row[0]
    
    # 3. 查询当前问卷包含/缺失的所有问题，生成动态列
    question_configs = cursor.execute("""
        SELECT *
        FROM
        (
            SELECT
                SurveyId,
                QuestionId,
                1 as InSurvey
            FROM SurveyStructure
            WHERE SurveyId = ?
            UNION
            SELECT 
                ? as SurveyId,
                Q.QuestionId,
                0 as InSurvey
            FROM Question as Q
            WHERE NOT EXISTS
            (
                SELECT *
                FROM SurveyStructure as S
                WHERE S.SurveyId = ? AND S.QuestionId = Q.QuestionId
            )
        ) as t
        ORDER BY QuestionId
    """, current_survey_id, current_survey_id, current_survey_id).fetchall()

    # 收集当前问卷的所有答案列SQL片段
    answer_col_list = []
    for q_row in question_configs:
        q_id = q_row[1]
        is_in_survey = q_row[2]
        if is_in_survey == 1:
            # 问卷包含该问题，生成关联答案的COALESCE列
            col_sql = f",COALESCE((SELECT a.Answer_Value FROM dbo.Answer a WHERE a.UserId = u.UserId AND a.SurveyId = {current_survey_id} AND a.QuestionId = {q_id}),-1) AS ANS_Q{q_id}"
        else:
            # 问卷不包含该问题，生成NULL列
            col_sql = f",NULL AS ANS_Q{q_id}"
        answer_col_list.append(col_sql)
    
    # 拼接当前问卷的完整SELECT块
    answer_cols_sql = "\n    ".join(answer_col_list)
    current_block = f"""SELECT
    UserId
    , {current_survey_id} as SurveyId
    {answer_cols_sql}
FROM
    dbo.[User] as u
WHERE EXISTS
(
    SELECT * FROM Answer as a
    WHERE u.UserId = a.UserId AND a.SurveyId = {current_survey_id}
)"""
    union_query_blocks.append(current_block)

# 4. 用UNION拼接所有查询块，得到最终SQL
final_sql = "\nUNION\n".join(union_query_blocks)

# 打印校验或直接执行
print(final_sql)
# result = cursor.execute(final_sql).fetchall()

额外说明 #

• 原代码里SELECT UserId, X as SurveyId Dynamic_Answers的写法本身有语法错误，SurveyId后面缺逗号，这里的Dynamic_Answers本来就是占位符，直接替换成生成的动态列即可
• 原动态列生成逻辑里每次循环都重置questionIds/surveyIds列表、再取索引0值的写法属于冗余逻辑，直接用循环拿到的当前question_id、current_survey_id变量即可，避免索引取值错误
• 用列表收集SQL片段再统一join，比循环里反复做字符串拼接效率更高，格式也更容易调整

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