Wie kann ich in PostgreSQL UPSERT (MERGE, INSERT ... ON DUPLICATE UPDATE) durchführen?

Ich versuche, mit den PostgreSQL-Versionen vor 9.5 einen Beitrag zu einer anderen Lösung für das Problem des einmaligen Einfügens zu leisten. Die Idee besteht einfach darin, zuerst das Einfügen durchzuführen und, falls der Datensatz bereits vorhanden ist, ihn zu aktualisieren:

do $$
begin 
  insert into testtable(id, somedata) values(2,'Joe');
exception when unique_violation then
  update testtable set somedata = 'Joe' where id = 2;
end $$;

Beachten Sie, dass diese Lösung angewendet werden kann nur wenn keine Zeilen der Tabelle gelöscht wurden.

Ich weiß nichts über die Effizienz dieser Lösung, aber sie scheint mir vernünftig genug zu sein.

Hier einige Beispiele für insert ... on conflict ... (pg 9.5+): 

• Einfügen, bei Konflikt - nichts tun .
  insert into dummy(id, name, size) values(1, 'new_name', 3) on conflict do nothing; 

• Einfügen bei Konflikt - Update durchführen , Konfliktziel über Spalte angeben .
  insert into dummy(id, name, size) values(1, 'new_name', 3) on conflict(id) do update set name = 'new_name', size = 3; 

• Einfügen bei Konflikt - Aktualisierung durchführen , Konfliktziel über Einschränkungsname angeben.
  insert into dummy(id, name, size) values(1, 'new_name', 3) on conflict on constraint dummy_pkey do update set name = 'new_name', size = 4;

WITH UPD AS (UPDATE TEST_TABLE SET SOME_DATA = 'Joe' WHERE ID = 2 
RETURNING ID),
INS AS (SELECT '2', 'Joe' WHERE NOT EXISTS (SELECT * FROM UPD))
INSERT INTO TEST_TABLE(ID, SOME_DATA) SELECT * FROM INS

Getestet auf Postgresql 9.3

SQLAlchemy Upsert für Postgres> = 9.5

Da der große Beitrag oben viele verschiedene SQL-Ansätze für Postgres-Versionen abdeckt (nicht nur Nicht-9.5 wie in der Frage), möchte ich hinzufügen, wie dies in SQLAlchemy geschieht, wenn Sie Postgres 9.5 verwenden. Anstelle eines eigenen Upserts können Sie auch die Funktionen von SQLAlchemy verwenden (die in SQLAlchemy 1.1 hinzugefügt wurden). Ich persönlich würde empfehlen, diese zu verwenden, wenn möglich. Nicht nur wegen der Bequemlichkeit, sondern auch weil PostgreSQL alle Race-Bedingungen handhaben kann.

Cross-Posting aus einer anderen Antwort, die ich gestern gegeben habe ( https://stackoverflow.com/a/44395983/2156909 )

SQLAlchemy unterstützt ON CONFLICT jetzt mit zwei Methoden on_conflict_do_update() und on_conflict_do_nothing():

Kopieren aus der Dokumentation:

from sqlalchemy.dialects.postgresql import insert

stmt = insert(my_table).values(user_email='[email protected]', data='inserted data')
stmt = stmt.on_conflict_do_update(
    index_elements=[my_table.c.user_email],
    index_where=my_table.c.user_email.like('%@gmail.com'),
    set_=dict(data=stmt.excluded.data)
    )
conn.execute(stmt)

http://docs.sqlalchemy.org/de/latest/dialects/postgresql.html?highlight=conflict#insert-on-conflict-upsert

Da diese Frage geschlossen wurde, poste ich hier, wie Sie SQLAlchemy verwenden. Durch Rekursion wird eine Masseneinfügung oder ein Update wiederholt, um race-Bedingungen und Validierungsfehler zu bekämpfen.

Zuerst die Importe 

import itertools as it

from functools import partial
from operator import itemgetter

from sqlalchemy.exc import IntegrityError
from app import session
from models import Posts

Jetzt funktioniert ein paar Helfer

def chunk(content, chunksize=None):
    """Groups data into chunks each with (at most) `chunksize` items.
    https://stackoverflow.com/a/22919323/408556
    """
    if chunksize:
        i = iter(content)
        generator = (list(it.islice(i, chunksize)) for _ in it.count())
    else:
        generator = iter([content])

    return it.takewhile(bool, generator)


def gen_resources(records):
    """Yields a dictionary if the record's id already exists, a row object 
    otherwise.
    """
    ids = {item[0] for item in session.query(Posts.id)}

    for record in records:
        is_row = hasattr(record, 'to_dict')

        if is_row and record.id in ids:
            # It's a row but the id already exists, so we need to convert it 
            # to a dict that updates the existing record. Since it is duplicate,
            # also yield True
            yield record.to_dict(), True
        Elif is_row:
            # It's a row and the id doesn't exist, so no conversion needed. 
            # Since it's not a duplicate, also yield False
            yield record, False
        Elif record['id'] in ids:
            # It's a dict and the id already exists, so no conversion needed. 
            # Since it is duplicate, also yield True
            yield record, True
        else:
            # It's a dict and the id doesn't exist, so we need to convert it. 
            # Since it's not a duplicate, also yield False
            yield Posts(**record), False

Und zum Schluss die Aufwärtsfunktion

def upsert(data, chunksize=None):
    for records in chunk(data, chunksize):
        resources = gen_resources(records)
        sorted_resources = sorted(resources, key=itemgetter(1))

        for dupe, group in it.groupby(sorted_resources, itemgetter(1)):
            items = [g[0] for g in group]

            if dupe:
                _upsert = partial(session.bulk_update_mappings, Posts)
            else:
                _upsert = session.add_all

            try:
                _upsert(items)
                session.commit()
            except IntegrityError:
                # A record was added or deleted after we checked, so retry
                # 
                # modify accordingly by adding additional exceptions, e.g.,
                # except (IntegrityError, ValidationError, ValueError)
                db.session.rollback()
                upsert(items)
            except Exception as e:
                # Some other error occurred so reduce chunksize to isolate the 
                # offending row(s)
                db.session.rollback()
                num_items = len(items)

                if num_items > 1:
                    upsert(items, num_items // 2)
                else:
                    print('Error adding record {}'.format(items[0]))

So verwenden Sie es

>>> data = [
...     {'id': 1, 'text': 'updated post1'}, 
...     {'id': 5, 'text': 'updated post5'}, 
...     {'id': 1000, 'text': 'new post1000'}]
... 
>>> upsert(data)

Der Vorteil gegenüber bulk_save_objects besteht darin, dass Beziehungen, Fehlerprüfungen usw. beim Einfügen verarbeitet werden können (im Gegensatz zu bulk-Vorgängen ).