使用psycopg2.copy_expert导入PostgreSQL数据时遇语法错误求助

问题背景 #

• 原使用copy_from函数实现PostgreSQL数据导入，因新版本安全限制，copy_from不再支持schema_name.table_name格式的表名，需改用copy_expert
• 测试环境：
  • 已存在dev_schema schema，表结构：

    create table dev_schema.testtab(numval integer not null);
    

  • 数据文件Data.txt内容：

    100
    120
    200
    800
    500
    

  • psql中执行以下命令可成功导入：

    \copy dev_schema.testtab from 'D:/Data.txt' with (format CSV, NULL '', delimiter '|', quote '"');
    

两次错误尝试及原因分析 #

第一次错误代码及报错 #

代码：

import psycopg2
from psycopg2 import sql


def main():
    connection = psycopg2.connect(
        dbname="testdb",
        user="dev_user",
        password="Here I write my password",
        host="localhost",
        port=5432
    )
    cursor = connection.cursor()
    filepath = "D:/Data.txt"
    with open(
        file=filepath,
        mode="r",
        encoding="UTF-8"
    ) as file_desc:
        option_values = [
            "format CSV",
            "NULL ''",
            "delimiter '|'",
            "quote '\"'"
        ]
        copy_options = sql.SQL(', ').join(
            sql.Identifier(n) for n in option_values
        )
        cursor.copy_expert(
            sql=sql.SQL(
                "copy {} from stdin with ({})"
            ).format(
                sql.Identifier("dev_schema", "testtab"),
                copy_options
            ),
            file=file_desc
        )

    cursor.close()
    connection.close()
    print("done")


if __name__ == "__main__":
    main()

报错：

cursor.copy_expert(
psycopg2.errors.SyntaxError: ERROR:  option « format CSV » not recognized
LINE 1: copy "dev_schema"."testtab" from stdin with ("format CSV", "...

错误原因：
sql.Identifier()仅用于处理数据库对象名（如表名、列名），但代码中用它包裹了format CSV这类COPY选项，导致生成的SQL里选项被加上双引号，PostgreSQL无法识别带引号的选项语法。

第二次错误代码及报错 #

代码：

import psycopg2
from psycopg2 import sql


def main():
    connection = psycopg2.connect(
        dbname="testdb",
        user="dev_user",
        password="Here I write my password",
        host="localhost",
        port=5432
    )
    cursor = connection.cursor()
    filepath = "D:/Data.txt"
    with open(
        file=filepath,
        mode="r",
        encoding="UTF-8"
    ) as file_desc:
        option_values = "".join(
            [
                "format CSV, ",
                "NULL '', ",
                "delimiter '|', ",
                "quote '\"',"
            ]
        )
        cursor.copy_expert(
            sql=sql.SQL("".join(
                [
                    "copy {} from stdin with (",
                    option_values,
                    ")"
                ]).format(sql.Identifier("dev_schema", "testtab"))
            ),
            file=file_desc
        )

    cursor.close()
    connection.close()
    print("done")


if __name__ == "__main__":
    main()

报错：

psycopg2.errors.SyntaxError: ERROR : Syntax error on or near   'dev_schema'  
LINE 1: copy Identifier('dev_schema', 'testtab') from stdin with (fo...

错误原因：
错误混用了Python字符串的format()方法和psycopg2的sql.SQL()对象，导致sql.Identifier()被直接转为字符串Identifier('dev_schema', 'testtab')，而非生成正确的带引号表名"dev_schema"."testtab"，触发SQL语法错误。

正确解决方案 #

需严格遵循psycopg2的SQL拼接规则：

• 用sql.Identifier("dev_schema", "testtab")正确生成带schema的表名
• COPY选项直接作为SQL片段传入，无需用sql.Identifier()包裹
• 所有动态SQL拼接必须通过sql.SQL()的format()方法完成

正确代码：

import psycopg2
from psycopg2 import sql


def main():
    connection = psycopg2.connect(
        dbname="testdb",
        user="dev_user",
        password="Here I write my password",
        host="localhost",
        port=5432
    )
    cursor = connection.cursor()
    filepath = "D:/Data.txt"
    
    with open(file=filepath, mode="r", encoding="UTF-8") as file_desc:
        # 定义COPY选项的SQL片段
        copy_options = sql.SQL("format CSV, NULL '', delimiter '|', quote '\"'")
        
        # 正确拼接完整COPY语句
        copy_sql = sql.SQL("COPY {} FROM STDIN WITH ({})").format(
            sql.Identifier("dev_schema", "testtab"),
            copy_options
        )
        
        cursor.copy_expert(sql=copy_sql, file=file_desc)
    
    # 必须提交事务，否则数据不会持久化
    connection.commit()
    cursor.close()
    connection.close()
    print("done")


if __name__ == "__main__":
    main()

关键注意事项：

1. 务必调用connection.commit()，否则导入的数据不会写入数据库（原代码遗漏此步骤）
2. 仅数据库对象名（表、列、schema等）需要用sql.Identifier()处理，SQL语法部分直接作为sql.SQL()参数即可
3. 禁止混用Python字符串拼接/格式化与psycopg2的SQL对象，避免语法错误和注入风险

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