使用psycopg2执行Postgres动态TRUNCATE表名时遇单引号语法错误

我正在开发一个Python Web项目，后端采用Postgres数据库，使用psycopg2包。此前为演示目的编写的静态TRUNCATE查询可正常运行，改为动态传入表名时出现问题。

相关代码如下：

def clear_tables(query: str, vars_list: list[Any]):
    # psycopg2 connection engine code
    
    cursor.executemany(query=query, vars_list=vars_list)
    
    conn.commit()


def foo():
    vars_list: list[Any] = list()
    table_name: str = "person"
    record: tuple[Any] = (table_name,)
    vars_list.append(record)
    
    query = """
        TRUNCATE TABLE %s RESTART IDENTITY;
        """
     
    clear_tables(query=query, vars_list=vars_list)      

执行时出现语法错误，表名被自动添加了多余的单引号，错误信息如下：

psycopg2.errors.SyntaxError: syntax error at or near "'person'"
LINE 2:        TRUNCATE TABLE 'person' RESTART IDENTITY;

为啥会出这问题 #

psycopg2的%s占位符是为普通数据值设计的，比如字符串、数字这类，它会自动给传入的字符串加单引号来避免SQL注入。但表名属于SQL对象标识符，不是普通值，用这种方式传入就会被当成字符串字面量，直接导致语法错误。

解决方法 #

用psycopg2内置的psycopg2.sql模块来安全构造包含动态表名的SQL语句，这个模块会正确处理标识符的转义，不会添加多余的单引号，还能防范SQL注入。

修改后的代码示例：

from psycopg2 import sql
from typing import Any

def clear_tables(query: sql.Composed):
    # psycopg2 connection engine code（确保conn和cursor已正确初始化）
    
    cursor.execute(query)
    conn.commit()


def foo():
    table_name: str = "person"
    
    # 用Identifier包装表名，构造动态SQL
    query = sql.SQL("TRUNCATE TABLE {} RESTART IDENTITY;").format(
        sql.Identifier(table_name)
    )
     
    clear_tables(query=query)

处理多个表的情况 #

如果需要一次性清空多个表，可以这样写：

def foo():
    table_names = ["person", "address", "order"]
    
    query = sql.SQL("TRUNCATE TABLE {} RESTART IDENTITY;").format(
        sql.SQL(', ').join(map(sql.Identifier, table_names))
    )
     
    clear_tables(query=query)

注意事项 #

• 绝对不要直接用字符串拼接构造带表名的SQL，这会留下严重的SQL注入漏洞。
• 如果表名包含特殊字符或者Postgres保留字，sql.Identifier会自动给表名加上双引号，避免语法错误。

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