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)
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