conditional add statement in SQLAlchemy

Question

Suppose I want to upload several SQL records, to a table that may not be populated yet. If there is a record with a primary key("ID") that already exists, either in the table or in the records to be committed to a table, I want to replace the existing record with the new record. I'm using mssql, SQL server 2008.

My first guess would be

try:
    session.add(record)
    session.commit
except:
    session.query().\
        filter(Class.ID == record.ID).\
        update(some expression)
    session.commit()       

what should the expression be? and is there a cleaner(and safer!) way of doing this?

Answer 1

In general unless using statements that guarantee atomicity, you'll always have to account for race conditions that might arise from multiple actors trying to either insert or update (don't forget delete). Even the MERGE statement, though a single statement, can have race conditions if not used correctly.

Traditionally this kind of "upsert" is performed using stored procedures or other SQL or implementation specific features available, like the MERGE statement.

An SQLAlchemy solution has to either attempt the insert and perform an update if an integrity error is raised, or perform the udpate and attempt an insert if no rows were affected. It should be prepared to retry in case both operations fail (a row might get deleted or inserted in between):

from sqlalchemy.exc import IntegrityError

while True:  # Infinite loop, use a retry counter  if necessary
    try:
        # begin a save point, prevents the whole transaction failing
        # in case of an integrity error
        with session.begin_nested():
            session.add(record)
            # Flush instead of commit, we need the transaction intact
            session.flush()
            # If the flush is successful, break out of the loop as the insert
            # was performed
            break

    except IntegrityError:
        # Attempt the update. If the session has to reflect the changes
        # performed by the update, change the `synchronize_session` argument.
        if session.query(Class).\
                filter_by(ID=record.ID).\
                update({...},
                       syncronize_session=False):
            # 1 or more rows were affected (hopefully 1)
            break

        # Nothing was updated, perhaps a DELETE in between

    # Both operations have failed, retry

session.commit()

Regarding

If there is a record with a primary key("ID") that already exists, either in the table or in the records to be committed to a table, I want to replace the existing record with the new record.

If you can be sure that no concurrent updates to the table in question will happen, you can use Session.merge for this kind of task:

# Records have primary key set, on which merge can either load existing
# state and merge, or create a new record in session if none was found.
for record in records:
    merged_record = session.merge(record)
    # Note that merged_record is not record

session.commit()

The SQLAlchemy merge will first check if an instance with given primary key exists in the identity map. If it doesn't and load is passed as True it'll check the database for the primary key. If a given instance has no primary key or an instance cannot be found, a new instance will be created.

The merge will then copy the state of the given instance onto the located/created instance. The new instance is returned.