SQLAlchemy BEI DUPLIKAT VORHANDENEM SCHLÜSSEL AKTUALISIEREN

Question

Gibt es eine elegante Möglichkeit, ein INSERT ... ON DUPLICATE KEY UPDATE in SQLAlchemy durchzuführen? Ich meine etwas mit einer Syntax ähnlich wie inserter.insert().execute(list_of_dictionaries) ?

Answer 1

ON DUPLICATE KEY UPDATE Post-Version-1.2 für MySQL

Diese Funktionalität ist jetzt nur für MySQL in SQLAlchemy integriert. Die Antwort von somada141 unten hat die beste Lösung: https://stackoverflow.com/a/48373874/319066

ON DUPLICATE KEY UPDATE im SQL-Statement

Wenn Sie möchten, dass das generierte SQL tatsächlich ON DUPLICATE KEY UPDATE enthält, erfolgt der einfachste Weg über die Verwendung eines @compiles -Dekorators.

Der Code (verlinkt von einem guten Thread zum Thema auf reddit) für ein Beispiel kann auf github gefunden werden:

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert

@compiles(Insert)
def append_string(insert, compiler, **kw):
    s = compiler.visit_insert(insert, **kw)
    if 'append_string' in insert.kwargs:
        return s + " " + insert.kwargs['append_string']
    return s

my_connection.execute(my_table.insert(append_string = 'ON DUPLICATE KEY UPDATE foo=foo'), my_values)

Beachten Sie jedoch, dass Sie in diesem Ansatz manuell den append_string erstellen müssen. Sie könnten die append_string-Funktion wahrscheinlich ändern, damit sie automatisch den Einfügestring in ein Einfügen mit 'ON DUPLICATE KEY UPDATE'-Zeichenfolge ändert, aber ich werde das hier wegen der Faulheit nicht tun.

ON DUPLICATE KEY UPDATE Funktionalität innerhalb des ORM

SQLAlchemy bietet keine Schnittstelle für ON DUPLICATE KEY UPDATE oder MERGE oder eine ähnliche Funktionalität in seiner ORM-Schicht. Dennoch verfügt es über die session.merge()-Funktion, die die Funktionalität nur dann replizieren kann, wenn der Schlüssel in Frage ein Primärschlüssel ist.

session.merge(ModelObject) überprüft zunächst, ob bereits eine Zeile mit demselben Primärschlüsselwert vorhanden ist, indem es eine SELECT-Abfrage sendet (oder es lokal nachschaut). Wenn dies der Fall ist, wird irgendwo ein Flag gesetzt, das anzeigt, dass das ModelObject bereits in der Datenbank vorhanden ist, und dass SQLAlchemy eine UPDATE-Abfrage verwenden sollte. Beachten Sie, dass merge etwas komplizierter ist als das, aber es repliziert die Funktionalität gut mit Primärschlüsseln.

Aber was ist, wenn Sie die ON DUPLICATE KEY UPDATE-Funktionalität mit einem Nicht-Primärschlüssel (zum Beispiel einem anderen eindeutigen Schlüssel) möchten? Leider hat SQLAlchemy eine solche Funktion nicht. Stattdessen müssen Sie etwas erstellen, das Django's get_or_create() ähnelt. Eine weitere StackOverflow-Antwort behandelt dies, und ich werde hier der Einfachheit halber eine modifizierte, funktionierende Version davon einfügen.

def get_or_create(session, model, defaults=None, **kwargs):
    instance = session.query(model).filter_by(**kwargs).first()
    if instance:
        return instance
    else:
        params = dict((k, v) for k, v in kwargs.iteritems() if not isinstance(v, ClauseElement))
        if defaults:
            params.update(defaults)
        instance = model(**params)
        return instance

Answer 2

Ich sollte erwähnen, dass seit der Version 1.2 das SQLAlchemy 'core' eine Lösung für das oben genannte Problem hat, die integriert ist und unter hier (kopierter Ausschnitt unten) gesehen werden kann:

from sqlalchemy.dialects.mysql import insert

insert_stmt = insert(my_table).values(
    id='some_existing_id',
    data='eingefügter Wert')

on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
    data=insert_stmt.inserted.data,
    status='U'
)

conn.execute(on_duplicate_key_stmt)

Answer 3

Basierend auf phsource's Antwort und für den spezifischen Anwendungsfall der Verwendung von MySQL und dem vollständigen Überschreiben der Daten für denselben Schlüssel ohne Ausführung einer DELETE-Anweisung kann man den folgenden durch @compiles dekorierten Insert-Ausdruck verwenden:

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert

@compiles(Insert)
def append_string(insert, compiler, **kw):
    s = compiler.visit_insert(insert, **kw)
    if insert.kwargs.get('on_duplicate_key_update'):
        fields = s[s.find("(") + 1:s.find(")")].replace(" ", "").split(",")
        generated_directive = ["{0}=VALUES({0})".format(field) for field in fields]
        return s + " ON DUPLICATE KEY UPDATE " + ",".join(generated_directive)
    return s

Answer 4

Mein Weg

import typing
from datetime import datetime
from sqlalchemy.dialects import mysql

class MyRepository:

    def model(self):
        return MySqlAlchemyModel

    def upsert(self, data: typing.List[typing.Dict]):
        if not data:
            return
        model = self.model()
        if hasattr(model, 'created_at'):
            for item in data:
                item['created_at'] = datetime.now()

        stmt = mysql.insert(getattr(model, '__table__')).values(data)
        for_update = []
        for k, v in data[0].items():
            for_update.append(k)

        dup = {k: getattr(stmt.inserted, k) for k in for_update}
        stmt = stmt.on_duplicate_key_update(**dup)
        self.db.session.execute(stmt)
        self.db.session.commit()

Verwendung:

myrepo.upsert([
    {
        "field11": "value11",
        "field21": "value21",
        "field31": "value31",
    },
    {
        "field12": "value12",
        "field22": "value22",
        "field32": "value32",
    },
])

Answer 5

Eine einfachere Lösung:

from sqlalchemy.ext.compilerimport compiles
from sqlalchemy.sql.expression import Insert

@compiles(Insert)
def replace_string(insert, compiler, **kw):
    s = compiler.visit_insert(insert, **kw)
    s = s.replace("INSERT INTO", "REPLACE INTO")
    return s

my_connection.execute(my_table.insert(replace_string=""), my_values)