In the overwrite=True case I wonder if it’s a good idea to update the existing row instead of a delete-insert combination. Maybe even upsert on supported database backends?

That's fair. I tried to get updates working, but ran into IntegrityErrors a bunch due to how flushes work. I fell back on delete/flush/create, but I'll take another look today to see if I can get update working.

Thanks for the review!

In case this helps any future readers (or me 😁), the IntegrityErrors were due to the fact that I thought conn_id was the PK for connections. When I merged a new Connection into the session to overwrite an existing one with the same conn_id, the session wouldn't reconcile the existing and new entries.

While there is a unique constraint on conn_id, the PK is id. The two entries having different PKs with the same conn_id is what caused the IntegrityError.

I fixed that by setting conn.id = existing_conn.id and have upserts working.

The overwrite part is technically unnecessary; if there is an existing_conn, overwrite is always True.

Maybe we could even refactor this entire block into

existing_id = session.query(Connection.id).filter(Connection.conn_id == conn_id).first()
if existing_id is not None:
    if not overwrite:
        continue
    conn.id = existing_id
session.merge(conn)

(I omitted some code not related to the flow control.)

You're right, I'll refactor that.

Incorporated the feedback above with one tweak:

Using .first() yields a tuple representing the select clause, like (1,).

Using .scalar() gets the first row and returns the first item in the tuple (or None, if no rows are returned).