使用 psycopg2 构建动态 INSERT 查询答案

【问题标题】：Build dynamic INSERT query with psycopg2使用 psycopg2 构建动态 INSERT 查询
【发布时间】：2020-02-12 16:12:35
【问题描述】：

我目前正在编写一个小工具来将数据从 SQL Sever DB 移动到 postgres DB。到目前为止，连接基本查询一切正常。

我发现动态创建查询非常困难。它需要是动态的，因为需要移动许多表和许多列。此外，表和列的顺序和名称不一定从 getgo 匹配。我已经将新旧列与数据匹配，我需要做的最后一件事是构建插入语句。按列进行也不是一种选择，因为非空约束不允许一次只插入一个值。

Python 3.7.2

Postgres 12

Windows 10 64 位

所以基本上我想要的，在伪代码中是这样的：

INSERT INTO {tableName} {columnList} VALUES {valueList}

在尝试了本质上是字符串操作的多种变体并参考了这些资源/SO 帖子之后：

http://aklaver.org/wordpress/2018/04/21/building-dynamic-sql-using-psycopg2/

https://www.psycopg.org/docs/usage.html#query-parameters

https://www.psycopg.org/docs/sql.html#module-psycopg2.sql

build SQL dynamic query with psycopg2 python library and using good conversion type tools

到目前为止，我想出了这个代码：

# wfData -- List of Lists which contain data, of mixed data types(int,str,NULL), every list is a row from #the database
# tableName -- Name of the table i want to insert into, as a string
# columnMap -- List of lists, which contain the matched column names --> [[col1_old,col2_old,..],[col1_new,col2_new,...]]
def addData(self, wfData, tableName, columnMap ):
    cursor = self.cursor
    for row in wfData:
        insert_str = sql.SQL("INSERT INTO {tableName} ({}) VALUES ({})").format(
        sql.SQL(",").join(map(sql.Identifier, columnMap[1])),
        sql.SQL(",").join(map(sql.Placeholder, row)),
        sql.Identifier(tableName)
        )
        try:
            cursor.execute(insert_str)
        except Exception as e:
            print(e)
        return

尝试运行此代码会产生以下错误： expected string or None as name, got 1 这让我迷失了方向，因为我真的不知道这是否来自某些类型转换问题，我怀疑这是不是我想在这里做的事情是错误的。

感谢您的建议。

编辑 1

我已更改参数的顺序，但无济于事。错误还是一样。但是我有了一个有趣的发现，在重新访问上面提到的资源之后，我设法让整个事情至少产生了一个 SQL 字符串，这仍然不起作用，但 progres 是 proges。根据我所做的更改，我的数据结构似乎导致了错误，直到现在数据被组织在列表中。我现在已经把它改成了一个名为 test 的字典：

# wfData -- List of Lists which contain data, of mixed data types(int,str,NULL), every list is a row from #the database
# tableName -- Name of the table i want to insert into, as a string
# columnMap -- List of lists, which contain the matched column names --> [[col1_old,col2_old,..],[col1_new,col2_new,...]]
    def addData(self, wfData, tableName, columnMap ):
        cursor = self.cursor
        test = {"id": 1}
        for row in wfData:
            try:
                insert_str = sql.SQL("INSERT INTO {} ({}) VALUES ({})").format(
                sql.Identifier(tableName),
                sql.SQL(",").join(map(sql.Identifier, test)),
                sql.SQL(",").join(map(sql.Placeholder, test))
                )
                cursor.execute(insert_str)

            except Exception as e:
                print(e)
            exit()
        return

现在我得到这个错误：

ERROR:  Syntaxerror at »%«
LINE 1: INSERT INTO "gr_akt_a" ("id") VALUES (%(id)s)
#The printed sql string :
Composed([SQL('INSERT INTO '), Identifier('graves_cur_h'), SQL(' ('), Composed([Identifier('id')]), SQL(') VALUES ('), Composed([Placeholder('id')]), SQL(')')])

【问题讨论】：

改用 SQLAlchemy，它会让你变得非常容易。
请显示一些输入：wfData, tableName, columnMap
您的参数是否向后列出？ sql.Identifier(tableName) 应该是第一个？

标签： python postgresql psycopg2

【解决方案1】：

我已经做到了。问题是我的数据结构。我试图用列表处理所有事情，并将一些列表处理与一些字典处理混为一谈，简而言之，这是一团糟，但我现在设法让事情正常进行：

def addData(self, wfData, tableName, columnMap ):
    cursor = self.cursor      
    for row in wfData:
        test = dict(zip(columnMap[1],row)) #this is where the magic starts
        try:
            insert_str = sql.SQL("INSERT INTO {} ({}) VALUES ({})").format(
            sql.Identifier(tableName),
            sql.SQL(",").join(map(sql.Identifier, test)),
            sql.SQL(",").join(map(sql.Placeholder, test))
            )
            print(type(insert_str))
            cursor.execute(insert_str, test)# the 'test' here is crucial i just forgot it
        except Exception as e:
            print(e)
        exit()
    return

现在插入查询被执行，它仍然崩溃，但这是一些与数据相关的问题。截至目前，我试图将几何 vom sql 服务器移动到 postgis 而不进行任何处理。欢迎对这类事情提出任何建议。

【讨论】：