如何将字典转换为SQL查询？动态生成SQL遇问题求助

需求说明 #

给定如下结构的payload：

payload={
    'Tables':['A', 'B'],
    'Cols':{"A":["Product_Name","Product_Desc","Sales"],
            "B":['Product_Size', 'Product_Desc']},
     'Function': {
         'agg': [{'Type': 'sum', 'Value': 'Sales'}],  # 注：原payload中Value为Enrollments，与期望输出的SUM(Sales)不符，此处修正为Sales
        "groupby": ['Product_Name','Product_Desc']
}}

期望生成的SQL查询结果：

Query={
    'A':"SELECT Product_Name, Product_Desc, SUM(Sales) FROM A GROUP BY Product_Name, Product_Desc;",
    'B': "SELECT Product_Size, Product_Desc FROM B"
}

现有代码问题 #

当前代码仅处理了GROUP BY子句的判断逻辑，既没有构建完整的SELECT语句，也未处理聚合函数的生成，且直接修改了原col_info的结构，逻辑不完整，无法生成符合要求的SQL。

正确实现代码 #

# 初始化存储查询结果的字典
Query = {}

# 从payload中提取关键信息
tables = payload['Tables']
cols_map = payload['Cols']
function = payload.get('Function', {})
groupby_cols = function.get('groupby', [])
agg_funcs = function.get('agg', [])

# 遍历每个表生成对应SQL
for table in tables:
    table_cols = cols_map[table]
    select_parts = []
    
    # 检查当前表是否包含所有分组列
    has_groupby = all(col in table_cols for col in groupby_cols)
    
    if has_groupby and agg_funcs:
        # 添加分组列到SELECT部分
        select_parts.extend(groupby_cols)
        # 生成聚合函数表达式
        for agg in agg_funcs:
            agg_type = agg['Type'].upper()
            agg_col = agg['Value']
            # 确保聚合列存在于当前表中
            if agg_col in table_cols:
                select_parts.append(f"{agg_type}({agg_col})")
        # 拼接完整SQL
        select_clause = ", ".join(select_parts)
        groupby_clause = ", ".join(groupby_cols)
        sql = f"SELECT {select_clause} FROM {table} GROUP BY {groupby_clause};"
    else:
        # 不满足分组条件时，查询当前表所有列
        select_clause = ", ".join(table_cols)
        sql = f"SELECT {select_clause} FROM {table}"
    
    Query[table] = sql

print(Query)

代码说明 #

• 容错性处理：用get方法提取Function配置，避免因key不存在抛出异常
• 表级独立判断：针对每个表单独校验是否满足分组聚合条件，适配不同表的列结构差异
• 规范生成聚合函数：将聚合类型转为大写（符合SQL语法规范），并验证聚合列是否存在于当前表中
• 分支构建SQL：根据是否满足分组条件，分别生成带GROUP BY的聚合查询或普通全列查询

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