我使用的是SQLAlchemy 1.3.4和PostgreSQL 11.3。
以下是我简化后的表定义:
我不得不创建两个唯一索引而不是一个
我正在尝试执行以下查询:
我不知道如何正确使用SQLAlchemy的
以下是我尝试过的内容:
以下是我简化后的表定义:
class MyModel(Base):
__tablename__ = 'mymodel'
id = Column(Integer, primary_key=True)
col1 = Column(Unicode, nullable=False)
col2 = Column(Unicode, nullable=False)
col3 = Column(Unicode, nullable=False)
col4 = Column(Boolean)
created_at = Column(DateTime(timezone=True), nullable=False)
updated_at = Column(DateTime(timezone=True), nullable=False)
__table_args__ = (
Index('uq_mymodel_col1_col2_col3_col4',
col1, col2, col3, col4,
unique=True, postgresql_where=col4.isnot(None)),
Index('uq_mymodel_col1_col2_col3',
col1, col2, col3,
unique=True, postgresql_where=col4.is_(None)),
)
我不得不创建两个唯一索引而不是一个
UniqueConstraint
,因为UniqueConstraint
允许多行具有相同的(col1,col2,col3)
,即使 col4
为空,这不是我想要的。我正在尝试执行以下查询:
INSERT INTO mymodel (col1, col2, col3, col4, created_at, updated_at)
VALUES (%(col1)s, %(col2)s, %(col3)s, %(col4)s, %(created_at)s, %(updated_at)s)
ON CONFLICT DO UPDATE SET updated_at = %(param_1)s
RETURNING mymodel.id
我不知道如何正确使用SQLAlchemy的
on_conflict_do_update()
。:-/以下是我尝试过的内容:
values = {…}
stmt = insert(MyModel.__table__).values(**values)
stmt = stmt.returning(MyModel.__table__.c.id)
stmt = stmt.on_conflict_do_update(set_={'updated_at': values['updated_at']})
result = dbsession.connection().execute(stmt)
然而SQLAlchemy报错:除非使用 DO NOTHING,否则必须指定约束或索引元素,但不是两者都要指定
我觉得如何使用约束
或索引元素
非常不清楚。
我尝试了一些方法,但都没有成功。例如:
values = {…}
stmt = insert(MyModel.__table__).values(**values)
stmt = stmt.returning(MyModel.__table__.c.id)
stmt = stmt.on_conflict_do_update(constraint='uq_mymodel_col1_col2_col3_col4'
set_={'updated_at': values['updated_at']})
result = dbsession.connection().execute(stmt)
但是这也不起作用:constraint "uq_mymodel_col1_col2_col3_col4" for table "mymodel" does not exist
。但它确实存在。(我甚至从pgsql中复制粘贴以确保我没有打错)
无论如何,我有两个唯一的约束条件可能会引发冲突,但是on_conflict_do_update()
似乎只能处理一个。所以我也尝试像这样指定两个:
values = {…}
stmt = insert(MyModel.__table__).values(**values)
stmt = stmt.returning(MyModel.__table__.c.id)
stmt = stmt.on_conflict_do_update(constraint='uq_mymodel_col1_col2_col3_col4'
set_={'updated_at': values['updated_at']})
stmt = stmt.on_conflict_do_update(constraint='uq_mymodel_col1_col2_col3'
set_={'updated_at': values['updated_at']})
result = dbsession.connection().execute(stmt)
但我得到了相同的错误,即 uq_mymodel_col1_col2_col3_col4
不存在。
目前,我无法弄清如何执行上述查询,并真的很需要一些帮助。