SQLAlchemy IntegrityError und Massendatenimport

Question

Ich füge mehrere 10k Datensätze in eine Datenbank mit REF-Integritätsregeln ein. Einige der Datenzeilen sind leider Duplikate (da sie bereits in der Datenbank existieren). Es wäre zu teuer, die Existenz jeder Zeile in der Datenbank vor dem Einfügen zu überprüfen, also beabsichtige ich, die von SQLAlchemy ausgelösten IntegrityError-Exceptions zu behandeln, den Fehler zu protokollieren und dann fortzufahren.

Mein Code wird wie folgt aussehen:

# establish connection to db etc. 

tbl = obtain_binding_to_sqlalchemy_orm() 
datarows = load_rows_to_import() 

try: 
    conn.execute(tbl.insert(), datarows) 
except IntegrityError as ie: 
    # eat error and keep going 
except Exception as e: 
    # do something else 

Die (implizite) Annahme, die ich oben mache ist, dass SQLAlchemy nicht die mehrere Einsätze in einer Transaktion rollen. Wenn meine Annahme falsch ist, bedeutet dies, dass bei einem IntegrityError der Rest des Einsatzes abgebrochen wird. Kann irgendjemand bestätigen, ob das Pseudocode "Muster" wie erwartet funktioniert - oder werde ich am Ende Daten verlieren aufgrund von ausgelösten IntegrityError Exceptions?

Wenn jemand eine bessere Idee hat, dies zu tun, würde mich das interessieren.

Answer 1

kann es so funktionieren, wenn Sie keine Transaktion zuvor gestartet haben, wie in diesem Fall sqlalchemy autocommit feature tritt in. Aber Sie sollten explizit wie im Link beschrieben.

Answer 2

Ich stieß auch auf dieses Problem, als ich ASCII-Datendateien analysierte, um die Daten in eine Tabelle zu importieren. Das Problem ist, dass ich instinktiv und intuitiv wollte, dass SQLAlchemy die doppelten Zeilen überspringt und dabei die eindeutigen Daten zulässt. Oder es kann sein, dass aufgrund der aktuellen SQL-Engine ein zufälliger Fehler in einer Zeile ausgelöst wird, beispielsweise wenn Unicode-Strings nicht zulässig sind.

Dieses Verhalten ist jedoch außerhalb des Bereichs der Definition der SQL-Schnittstelle. SQL-APIs und damit SQLAlchemy verstehen nur Transaktionen und Commits und berücksichtigen dieses selektive Verhalten nicht. Darüber hinaus klingt es gefährlich, von der Autocommit-Funktion abhängig zu sein, da die Einfügung nach der Ausnahme anhält und den Rest der Daten übrig lässt.

Meine Lösung (die ich nicht sicher bin, ob es die eleganteste ist) besteht darin, jede Zeile in einer Schleife zu verarbeiten, Ausnahmen abzufangen und zu protokollieren und die Änderungen am Ende festzuschreiben.

Unter der Annahme, dass Sie irgendwie Daten in einer Liste von Listen erfasst haben, d. H. Liste von Zeilen, die Listen von Spaltenwerten sind. Dann lesen Sie jede Zeile in einer Schleife:

# Python 3.5 
from sqlalchemy import Table, create_engine 
import logging 

# Create the engine 
# Create the table 
# Parse the data file and save data in `rows` 

conn = engine.connect() 
trans = conn.begin() # Disables autocommit 

exceptions = {} 
totalRows = 0 
importedRows = 0 

ins = table.insert() 

for currentRowIdx, cols in enumerate(rows): 
    try: 
     conn.execute(ins.values(cols)) # try to insert the column values 
     importedRows += 1 

    except Exception as e: 
     exc_name = type(e).__name__ # save the exception name 
     if not exc_name in exceptions: 
      exceptions[exc_name] = [] 
     exceptions[exc_name].append(currentRowIdx) 

    totalRows += 1 

for key, val in exceptions.items(): 
    logging.warning("%d out of %d lines were not imported due to %s."%(len(val), totalRows, key)) 

logging.info("%d rows were imported."%(importedRows)) 

trans.commit() # Commit at the very end 
conn.close() 

Um die Geschwindigkeit in diesem Vorgang zu maximieren, sollten Sie Autocommit deaktivieren. Ich verwende diesen Code mit SQLite und es ist immer noch 3-5 Mal langsamer als meine ältere Version, die nur sqlite3 verwendet, auch wenn Autocommit deaktiviert ist. (Der Grund, dass ich nach SQLAlchemy portiert habe, war, es mit MySQL verwenden zu können.)

Es ist nicht die eleganteste Lösung in dem Sinne, dass es nicht so schnell ist wie eine direkte Schnittstelle zu SQLite. Wenn ich den Code profiliere und den Engpass in naher Zukunft finde, werde ich diese Antwort mit der Lösung aktualisieren.