Use "on_conflict_do_update()" with sqlalchemy ORM

Question

I am currently using SQLAlchemy ORM to deal with my db operations. Now I have a SQL command which requires ON CONFLICT (id) DO UPDATE. The method on_conflict_do_update() seems to be the correct one to use. But the post here says the code have to switch to SQLAlchemy core and the high-level ORM functionalities are missing. I am confused by this statement since I think the code like the demo below can achieve what I want while keep the functionalities of SQLAlchemy ORM.

class Foo(Base):
  ...
  bar = Column(Integer)


foo = Foo(bar=1)


insert_stmt = insert(Foo).values(bar=foo.bar)
do_update_stmt = insert_stmt.on_conflict_do_update(
    set_=dict(
        bar=insert_stmt.excluded.bar,
    )
)

session.execute(do_update_stmt)

I haven't tested it on my project since it will require a huge amount of modification. Can I ask if this is the correct way to deal with ON CONFLICT (id) DO UPDATE with SQLALchemy ORM?

Answer 1

As noted in the documentation, the constraint= argument is

The name of a unique or exclusion constraint on the table, or the constraint object itself if it has a .name attribute.

so we need to pass the name of the PK constraint to .on_conflict_do_update().

We can get the PK constraint name via the inspection interface:

insp = inspect(engine)
pk_constraint_name = insp.get_pk_constraint(Foo.__tablename__)["name"]
print(pk_constraint_name)  # tbl_foo_pkey

new_bar = 123
insert_stmt = insert(Foo).values(id=12345, bar=new_bar)
do_update_stmt = insert_stmt.on_conflict_do_update(
    constraint=pk_constraint_name, set_=dict(bar=new_bar)
)

with Session(engine) as session, session.begin():
    session.execute(do_update_stmt)