使用 execute_batch 插入时出现 Psycopg2 类型错误

Question

我正在尝试使用 psycopg2 将批量插入到 postgres 数据库中。我正在使用 %s 和元组列表，但失败并出现以下错误：

File ".../python3.6/site-packages/psycopg2/extras.py", line 1183, in execute_batch
    sqls = [cur.mogrify(sql, args) for args in page]
  File ".../python3.6/site-packages/psycopg2/extras.py", line 1183, in <listcomp>
    sqls = [cur.mogrify(sql, args) for args in page]
TypeError: not all arguments converted during string formatting

这是我的代码：

import psycopg2
import psycopg2.extras
    conn = psycopg2.connect(
        database='mydb',
        user='name',
        password='pass')
    cur = conn.cursor()
    query = "INSERT INTO my_table (tweet_id, user_id, time, text, 
        reply_to_user_id, reply_to_tweet_id, reply_to_handle, is_retweet, 
        is_quote, quote_usr_id, quote_usr_handle, quote_id, quote_text, 
        retweet_usr_id, retweet_usr_handle, retweet_id, longitude, latitude, 
        location, time_zone) VALUES (%s);"
    #vals are values to insert, a list of tuples
    vals = [(123, 123, datetime.datetime(2017, 1, 18, 17, 12, 33), 
        "'Some Text'", None, None, None, None, None, None, None, None, 
        None, 1234, "'username'", 1234, None, None, "'Somewhere'", 
        "'Pacific Time (US & Canada)'"), 
        (321, 321, datetime.datetime(2017, 1, 18, 15, 43, 19), 
        "'More text'", 321, 321, "'person'", None, None, None, None, None,  
        None, None, None, None, None,None, "'faraway'", 
        "'Pacific Time (US & Canada)'")]
    psycopg2.extras.execute_batch(cur,query,vals)

我也尝试用“NULL”字符串替换 Nones 无济于事。

vals 是一个元组列表，所以它不是 faq 和 various stackoverflow posts. 的常见问题

我希望这是我插入的其中一种类型的问题，但不知道是哪一种。我已经将 psycopg2 python 与 sql datatype conversion chart 进行了比较，似乎应该检查一下。

Answer 1

将值的数量显式传递到您的查询中。

import psycopg2
import psycopg2.extras
conn = psycopg2.connect(
    database='mydb',
    user='name',
    password='pass')
    cur = conn.cursor()
query = "INSERT INTO my_table (tweet_id, user_id, time, text, 
    reply_to_user_id, reply_to_tweet_id, reply_to_handle, is_retweet, 
    is_quote, quote_usr_id, quote_usr_handle, quote_id, quote_text, 
    retweet_usr_id, retweet_usr_handle, retweet_id, longitude, latitude, 
    location, time_zone) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);"
#vals are values to insert, a list of tuples
vals = [(123, 123, datetime.datetime(2017, 1, 18, 17, 12, 33), 
    "'Some Text'", None, None, None, None, None, None, None, None, 
    None, 1234, "'username'", 1234, None, None, "'Somewhere'", 
    "'Pacific Time (US & Canada)'"), 
    (321, 321, datetime.datetime(2017, 1, 18, 15, 43, 19), 
    "'More text'", 321, 321, "'person'", None, None, None, None, None,  
    None, None, None, None, None,None, "'faraway'", 
    "'Pacific Time (US & Canada)'")]
psycopg2.extras.execute_batch(cur,query,vals)

Answer 2

我建议您编写一个类型的函数，而不是编写查询（如果需要，则使用 UPSERT）：

def execute_upsert(cur, table, champs, rows, champs_unique=None):
    """
        Fonction qui renvoie la requete pour execute_batch(cur, requete, list_rows) de psycopg2
        :param cur: cursor psycopg2
        :param table: Table postgresql, pour l'insertion
        :param champs: List ou Tuple des Champs de la table, pour les insertions
        :param rows: Tableau (list, tuple, generator) des lignes à insérer
        :param champs_unique: Si Upsert, on donne un tuple des champs d'unicité
    """

    def get_sql(t, col, c_u):
        """
            Fonction qui renvoie la requete pour execute_batch(cur, requete, list_rows) de psycopg2
            :param t: Table postgresql, pour l'insertion
            :param col: List ou Tuple des Champs de la table, pour les insertions
            :param c_u: Si Upsert, on donne un tuple des champs d'unicité
            :return: Retourne la Requête
        """
        sql_insert = f'INSERT INTO "{t}" ('
        champs_insertion = []

        for champ in col:
            sql_insert += f'"{champ}", '
            test_champ = "%s"
            champs_insertion.append(test_champ)

        if len(champs_insertion) > 1:
            c = str(tuple(champs_insertion)).replace("'", "")
            sql_insert = sql_insert[:-2] + f') VALUES {c} '
        else:
            c = "(" + str(champs_insertion[0]).replace("'", "") + ")"
            sql_insert = sql_insert[:-2] + f') VALUES {c} '

        if c_u is None:
            sql_insert += 'ON CONFLICT DO NOTHING;'

        else:
            if champs_unique is not None:
                chu = "("
                for value in champs_unique:
                    chu += f'"{value}", '
                chu = f'{chu[:-2]})'
                sql_insert += f' ON CONFLICT {chu} DO UPDATE SET '
                for champ in champs:
                    if champ not in champs_unique:
                        sql_insert += f'"{str(champ)}" = excluded."{str(champ)}", '

                sql_insert = sql_insert[:-2] + ';'

            else:
                sql_insert += ';'

        return sql_insert

    execute_batch(cur, get_sql(table, champs, champs_unique), rows)

之后很简单：

execute_upsert(
    cur=cur,
    table=table_projets_colonnesprojet,
    champs=champs_table,
    rows=list_rows,
    champs_unique=("code_comptable",)
)