How to do an upsert with SqlAlchemy?

This allows access to the underlying models based on string names

def get_class_by_tablename(tablename):
"""Return class reference mapped to table.
https://stackoverflow.com/questions/11668355/sqlalchemy-get-model-from-table-name-this-may-imply-appending-some-function-to
:param tablename: String with name of table.
:return: Class reference or None.
"""
for c in Base._decl_class_registry.values():
if hasattr(c, '__tablename__') and c.__tablename__ == tablename:
return c




sqla_tbl = get_class_by_tablename(table_name)


def handle_upsert(record_dict, table):
"""
handles updates when there are primary key conflicts


"""
try:
self.active_session().add(table(**record_dict))
except:
# Here we'll assume the error is caused by an integrity error
# We do this because the error classes are passed from the
# underlying package (pyodbc / sqllite) SQLAlchemy doesn't mask
# them with it's own code - this should be updated to have
# explicit error handling for each new db engine


# <update>add explicit error handling for each db engine</update>
active_session.rollback()
# Query for conflic class, use update method to change values based on dict
c_tbl_primary_keys = [i.name for i in table.__table__.primary_key] # List of primary key col names
c_tbl_cols = dict(sqla_tbl.__table__.columns) # String:Col Object crosswalk


c_query_dict = {k:record_dict[k] for k in c_tbl_primary_keys if k in record_dict} # sub-dict from data of primary key:values
c_oo_query_dict = {c_tbl_cols[k]:v for (k,v) in c_query_dict.items()} # col-object:query value for primary key cols


c_target_record = session.query(sqla_tbl).filter(*[k==v for (k,v) in oo_query_dict.items()]).first()


# apply new data values to the existing record
for k, v in record_dict.items()
setattr(c_target_record, k, v)

There are multiple answers and here comes yet another answer (YAA). Other answers are not that readable due to the metaprogramming involved. Here is an example that

• 使用 SQLAlchemy ORM

• 演示如何使用 on_conflict_do_nothing创建零行的行

• 演示如何在不使用 on_conflict_do_update创建新行的情况下更新现有行(如果有的话)

• 使用表主键作为 constraint

the original question what this code is related to中的一个更长的示例。

import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy import text
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session


class PairState(Base):


__tablename__ = "pair_state"


# This table has 1-to-1 relationship with Pair
pair_id = sa.Column(sa.ForeignKey("pair.id"), nullable=False, primary_key=True, unique=True)
pair = orm.relationship(Pair,
backref=orm.backref("pair_state",
lazy="dynamic",
cascade="all, delete-orphan",
single_parent=True, ), )




# First raw event in data stream
first_event_at = sa.Column(sa.TIMESTAMP(timezone=True), nullable=False, server_default=text("TO_TIMESTAMP(0)"))


# Last raw event in data stream
last_event_at = sa.Column(sa.TIMESTAMP(timezone=True), nullable=False, server_default=text("TO_TIMESTAMP(0)"))


# The last hypertable entry added
last_interval_at = sa.Column(sa.TIMESTAMP(timezone=True), nullable=False, server_default=text("TO_TIMESTAMP(0)"))


@staticmethod
def create_first_event_if_not_exist(dbsession: Session, pair_id: int, ts: datetime.datetime):
"""Sets the first event value if not exist yet."""
dbsession.execute(
insert(PairState).
values(pair_id=pair_id, first_event_at=ts).
on_conflict_do_nothing()
)


@staticmethod
def update_last_event(dbsession: Session, pair_id: int, ts: datetime.datetime):
"""Replaces the the column last_event_at for a named pair."""
# Based on the original example of https://stackoverflow.com/a/49917004/315168
dbsession.execute(
insert(PairState).
values(pair_id=pair_id, last_event_at=ts).
on_conflict_do_update(constraint=PairState.__table__.primary_key, set_={"last_event_at": ts})
)


@staticmethod
def update_last_interval(dbsession: Session, pair_id: int, ts: datetime.datetime):
"""Replaces the the column last_interval_at for a named pair."""
dbsession.execute(
insert(PairState).
values(pair_id=pair_id, last_interval_at=ts).
on_conflict_do_update(constraint=PairState.__table__.primary_key, set_={"last_interval_at": ts})
)

对于 sqlite，可以在定义 UniqueConstraint时使用 sqlite_on_conflict='REPLACE'选项，在定义单个列的唯一约束时使用 sqlite_on_conflict_unique选项。然后 session.add将以一种类似于 upsert的方式工作。见官方 文件。

由于我们在生成缺省 id 和引用时遇到了问题，这些问题会导致 ForeignKeyViviation-Errorlike

update or delete on table "..." violates foreign key constraint
Key (id)=(...) is still referenced from table "...".

我们必须排除更新 dict 的 id，否则它将总是作为新的默认值生成。

此外，该方法还返回创建/更新的实体。

from sqlalchemy.dialects.postgresql import insert # Important to use the postgresql insert




def upsert(session, data, key_columns, model):


stmt = insert(model).values(data)
    

# Important to exclude the ID for update!
exclude_for_update = [model.id.name, *key_columns]
update_dict = {c.name: c for c in stmt.excluded if c.name not in exclude_for_update}


stmt = stmt.on_conflict_do_update(
index_elements=key_columns,
set_=update_dict
).returning(model)


orm_stmt = (
select(model)
.from_statement(stmt)
.execution_options(populate_existing=True)
)


return session.execute(orm_stmt).scalar()

例如:

class UpsertUser(Base):
__tablename__ = 'upsert_user'
id = Column(Id, primary_key=True, default=uuid.uuid4)
name: str = Column(sa.String, nullable=False)
user_sid: str = Column(sa.String, nullable=False, unique=True)
house_admin = relationship('UpsertHouse', back_populates='admin', uselist=False)




class UpsertHouse(Base):
__tablename__ = 'upsert_house'
id = Column(Id, primary_key=True, default=uuid.uuid4)
admin_id: Id = Column(Id, ForeignKey('upsert_user.id'), nullable=False)
admin: UpsertUser = relationship('UpsertUser', back_populates='house_admin', uselist=False)


# Usage


upserted_user = upsert(session, updated_user, [UpsertUser.user_sid.name], UpsertUser)

注意: 只在 postgreql 上进行了测试，但也可以在其他支持 ON DUPLICATE KEY UPDATE 的 DBs 上进行测试，例如 MySQL