pythonsqlitesqlalchemyupsert

Insert or update when importing from JSON


My SQLAlchemy ORM model is populated by a JSON file that occasionally changes. The JSON file does not provide an integer primary key but has a unique alphanumeric ProductCode. My model:

class ProductDescriptor(Base):
    __tablename__ = 'product_descriptor'

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    ProductCode: Mapped[str] = mapped_column(String(50), unique=True)

    DisplayName: Mapped[str] = mapped_column(String(50))
    Description: Mapped[str] = mapped_column(String(1000))
    ...

This answer makes sense until this line:

on_duplicate_stmt = insert_stmt.on_duplicate_key_update(dict(txt=insert_stmt.inserted.txt))

Because the incoming data lacks a key I have nothing to compare. Do I need to change the ProductCode definition? I am using Unique=True. My code:

product_list = []
for product in products:

    # Create filtered list of dicts to be send to the DB
    product_list.append({
        'ProductName': product.get('ProductName'),
        'DisplayName': product.get('DisplayName'),
        'Description': product.get('Description'),
        ... more columns
    })

    insert_stmt = insert(ProductDescriptor).values(product_list)

    # This is where it goes wrong
    on_duplicate_stmt = insert_stmt.on_duplicate_key_update()

    # Trying to deal with integrity errors and rollback requests
    for product in product_list:
        try:
            self.session.add(resource)
            self.session.commit()
        except IntegrityError:
            pass

In Django I would be using the update_or_create method where I can specify the key field and provide a defaults dictionary:

Profile.objects.update_or_create(custid=user_profile.custid, defaults=defaults)

Solution

  • SQLite lets us specify the matching columns for ON CONFLICT, like so:

    from sqlalchemy.dialects.sqlite import insert
    
    new_values = json.loads("""\
    [
      {"ProductCode": "code_1", "DisplayName": "display_1", "Description": "description_1"},
      {"ProductCode": "code_2", "DisplayName": "display_2", "Description": "description_2"}
    ]
    """)
    insert_stmt = insert(ProductDescriptor).values(new_values)
    do_update_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["ProductCode"],
        set_=dict(
            DisplayName=insert_stmt.excluded.DisplayName,
            Description=insert_stmt.excluded.Description,
        ),
    )
    
    engine.echo = True
    with engine.begin() as conn:
        conn.execute(do_update_stmt)
    """
    BEGIN (implicit)
    INSERT INTO product_descriptor ("ProductCode", "DisplayName", "Description") VALUES (?, ?, ?), (?, ?, ?) ON CONFLICT ("ProductCode") DO UPDATE SET "DisplayName" = excluded."DisplayName", "Description" = excluded."Description"
    [no key 0.00093s] ('code_1', 'display_1', 'description_1', 'code_2', 'display_2', 'description_2')
    COMMIT
    """
    

    Note that if "ProductCode" is a unique non-nullable column then it is in fact a natural primary key, so the autoincrement integer "id" column is not really necessary