使用psycopg拼接列时去除结果引号并避免SQL注入的方法

嗨，我来帮你解决这个问题！你的核心困扰在于手动把psycopg的标识符对象转成字符串拼接——这既破坏了psycopg内置的防SQL注入机制，又导致CONCAT里的列名要么带多余引号，要么被当成列值输出。咱们用psycopg原生的SQL组合方式，就能完美兼顾安全和你想要的输出格式。

问题根源拆解 #

你之前的代码犯了两个关键错误：

1. 把psycopg.sql.Identifier转成字符串：比如Identifier('salary').as_string()会返回"salary"（带双引号），拼进CONCAT后就会输出带引号的结果。
2. 手动拼接SQL字符串：绕过了psycopg的安全处理机制，既容易出错，又埋下注入风险。

而你想要的结果是Houston-Alice-salary，本质是CONCAT(分组列的值 + 分隔符 + 列名的纯字符串)——这里的列名是作为字符串常量存在，不是列的值，也不是带引号的SQL标识符。

正确的解决方案：用psycopg原生组合SQL #

全程依赖psycopg.sql模块的Composable对象（SQL、Identifier、Literal）来构建查询，不要手动转字符串拼接。这样既安全防注入，又能精准控制输出格式。

修改后的完整代码

import psycopg
from psycopg import sql
import pandas as pd

def so_question(columns, group_by_columns, table):
    """
    安全构建SQL查询，生成符合要求的CONCAT结果
    """
    # 用Identifier包装表名和列名，防SQL注入
    table_ident = sql.Identifier(table)
    group_by_idents = [sql.Identifier(col) for col in group_by_columns]

    sql_statements = []
    for data_col in columns:
        # 构建CONCAT的各个部分：分组列 + 分隔符 + 列名字符串
        concat_parts = []
        for g_col in group_by_idents:
            concat_parts.append(g_col)
            # 每个分组列后加分隔符'-'
            concat_parts.append(sql.Literal('-'))
        # 最后加上列名的纯字符串（用Literal包装，安全转义）
        concat_parts.append(sql.Literal(data_col))
        
        # 组合成CONCAT表达式
        concat_expr = sql.SQL("CONCAT({})").format(sql.SQL(', ').join(concat_parts))
        
        # 构建GROUP BY子句：分组列 + 当前数据列
        group_by_expr = sql.SQL(', ').join(group_by_idents + [sql.Identifier(data_col)])
        
        # 用SQL模板组合整个查询
        sql_template = sql.SQL("""
            WITH sql_cte as (
                SELECT
                    {concat_expr} as concat_columns,
                    {data_col_ident} as value
                FROM {table}
                GROUP BY {group_by_expr}
            )
            SELECT * FROM sql_cte
        """)
        
        # 填充模板参数
        current_sql = sql_template.format(
            concat_expr=concat_expr,
            data_col_ident=sql.Identifier(data_col),
            table=table_ident,
            group_by_expr=group_by_expr
        )
        sql_statements.append(current_sql)
    
    # 如果是单列的情况，返回单个SQL对象；多列则返回列表
    return sql_statements[0] if len(sql_statements) == 1 else sql_statements

def execute_sql_get_dataframe(sql):
    """
    安全执行SQL并返回DataFrame，支持psycopg的Composable对象
    """
    try:
        # 替换为你的数据库连接参数
        with psycopg.connect("dbname=mydatabase user=myuser password=mypassword") as conn:
            with conn.cursor() as cur:
                # 直接执行Composable对象，不用转字符串
                cur.execute(sql)
                tuples_list = cur.fetchall()
                column_names = [desc[0] for desc in cur.description]
                return pd.DataFrame(tuples_list, columns=column_names)
    except Exception as exc:
        print(f"执行出错: {exc}")
        raise

# 测试调用
data_columns = ['salary']
group_by_columns_in_order_of_grouping = ['city', 'name']

_sql = so_question(
    columns=data_columns,
    group_by_columns=group_by_columns_in_order_of_grouping,
    table='sample_table'
)
dataframe = execute_sql_get_dataframe(sql=_sql)
print(dataframe)

关键细节说明 #

1. 防注入的核心：

   • 所有用户输入的表名、列名都用sql.Identifier包装，psycopg会自动处理标识符的转义，避免注入。
   • 列名作为字符串常量时用sql.Literal(data_col)包装，即使输入包含特殊字符或注入语句，也会被当成普通字符串处理。

2. 输出格式的控制：

   • sql.Literal(data_col)会生成不带多余引号的字符串常量，比如Literal('salary')在SQL里就是'salary'，CONCAT后结果就是Houston-Eve-salary。
   • 全程不用手动转字符串，psycopg会自动处理所有拼接和转义，不会出现多余的双引号。

3. 为什么之前的尝试失败：

   • 你之前用Identifier(data_col)转字符串，得到的是带双引号的标识符（比如"salary"），拼进CONCAT后自然会带引号。
   • 直接用data_col拼SQL会有注入风险，而用Literal则既安全又能得到纯字符串。

测试结果 #

执行上述代码后，你会得到完全符合预期的输出：

备注：内容来源于stack exchange，提问作者Python_Learner