SQLAlchemy - Session.merge() returning duplicate key integrity error

Question

I'm attempting to either insert or update some database rows based on information from a CSV file. I am able to read the CSV file into a list of dict() objects and process those dicts into class objects. However, attempting a Session.merge() on these objects gives the following error:

sqlalchemy.exc.IntegrityError: (MySQLdb._exceptions.IntegrityError) (1062, "Duplicate entry '27445-5810' for key 'SiteId'")

When previously working with sqlalchemy, merge() would automatically execute an UPDATE when the INSERT failed. Why would I be getting an integrity error in this case?

Here is my script:

from sqlalchemy import create_engine,and_,or_,func
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.automap import automap_base
from io import StringIO
from csv import reader
import xmltodict
from requests import get
from json import loads
from sys import argv
from datetime import datetime, timedelta
from traceback import print_exc
from tqdm import tqdm

#Set-Up DB connection/classes
engine = create_engine(
    "<DATABSE_URI_HERE>"
)
Base = automap_base()
Base.prepare(engine, reflect=True)
ProductSource = Base.classes.ProductSource
Products = Base.classes.Products
Session = sessionmaker()

siteid = argv[2] if 'site' in argv else input('SiteId: ')

def _findhead(obj, key):
    if key in obj: return obj[key]
    for _, v in obj.items():
        if isinstance(v,dict):
            item = _findhead(v, key)
            if item is not None:
                return item


conn = Session(bind=engine)
source = conn.query(ProductSource).filter(ProductSource.SiteId == siteid).first()
feed_file = get(source.Url).text
mapping = loads(source.CsvMap)

try:
    if source.FileType == 'csv':
        delimiters = {
            'comma' : ',',
            'tab' : '\t',
            'pipe' : '|',
            'semicolon' : ';'
        }
        with StringIO(feed_file) as f:
            products = list(reader(f, delimiter=delimiters[source.Other]))
    else:
        parsed = xmltodict.parse(feed_file)
        products = _findhead(parsed, source.HeadElement)
except:
    conn.close()
    print('Malformed file, cannot read this feed.')
    exit(1)

updated = 0
failed = 0
valid_keys =[]
started = datetime.now()
source.Started = started
source.Updated = updated
conn.commit()
for key in mapping.keys():
    if mapping[key] != None:
        valid_keys.append(key)
if 'Description' not in valid_keys or 'ProductId' not in valid_keys:
    print('Invalid mapping, nothing to do')
    conn.close()
    exit(1)
for product in tqdm(products):
    try:
        new_product_row = {'SiteId' : siteid}
        for key in valid_keys:
            new_product_row[key] = product[mapping[key]]
        conn.merge(Products(**new_product_row))
        conn.commit()
        updated += 1
    except:
        conn.rollback()
        failed += 1
finished = datetime.now()
next_read = started + timedelta(days=source.ReadInterval)
source.LastRead = finished
source.NextRead = next_read
source.Added = added
source.Updated = updated
conn.commit()
print(f'{source.Updated} products updated.')
print(f'{failed} products failed.')
conn.close()

Answer 1

This might be late but as Ilja stated, session.merge() only considers the primary key. In this case you would have to "lie" and make the unique constraint a composite primary key when defining your model.