使用具有唯一约束的SQLAlchemy的on_conflict_do_update()
我使用的是SQLAlchemy 1.3.4和PostgreSQL 11.3。
我有以下(简化的)表格定义：
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)),
)(我必须创建2个唯一的索引，而不是UniqueConstraint，因为UniqueConstraint允许多个行具有相同的(col1, col2, col3)为col4为null，这是我不想要的)
我正在尝试执行以下查询：
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抱怨道：Either constraint or index_elements, but not both, must be specified unless DO NOTHING
我发现如何使用constraint或index_elements非常不清楚。
我试了几件事，都没有用。例如：
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不存在。
在这一点上，我只是不知道如何做上面的查询，真的很感谢一些帮助。
回答 1
Stack Overflow用户
发布于 2019-06-04 20:39:53
好了，我想我想通了。所以问题毕竟不是来自SQLAlchemy，我实际上是在滥用PostgreSQL。
首先，我在上面粘贴的SQL查询不起作用，因为像SQLAlchemy一样，PostgreSQL需要指定索引列或约束名称。
当我指定一个约束时，PostgreSQL给出了与SQLAlchemy相同的错误。这是因为我的约束实际上不是约束，而是唯一索引。似乎它真的必须是一个唯一的约束，而不是一个唯一的索引。(即使该索引与唯一约束具有相同的效果)
所以我重写了模型，如下所示：
# Feel free to use the following code under the MIT license
class NullableBoolean(TypeDecorator):
"""A three-states boolean, which allows working with UNIQUE constraints
In PostgreSQL, when making a composite UNIQUE constraint where one of the
columns is a nullable boolean, then null values for that column are counted
as always different.
So if you have:
class MyModel(Base):
__tablename__ = 'mymodel'
id = Column(Integer, primary_key=True)
col1 = Column(Unicode, nullable=False)
col2 = Column(Unicode, nullable=False)
col3 = Column(Boolean)
__table_args__ = (
UniqueConstraint(col1, col2, col3,
name='uq_mymodel_col1_col2_col3'),
}
Then you could INSERT multiple records which have the same (col1, col2)
when col3 is None.
If you want None to be considered a "proper" value that triggers the
unicity constraint, then use this type instead of a nullable Boolean.
"""
impl = Enum
def __init__(self, **kwargs):
kwargs['name'] = 'nullable_boolean_enum'
super().__init__('true', 'false', 'unknown', **kwargs)
def process_bind_param(self, value, dialect):
"""Convert the Python values into the SQL ones"""
return {
True: 'true',
False: 'false',
None: 'unknown',
}[value]
def process_result_value(self, value, dialect):
"""Convert the SQL values into the Python ones"""
return {
'true': True,
'false': False,
'unknown': None,
}[value]
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__ = (
UniqueConstraint(col1, col2, col3, col4,
name='uq_mymodel_col1_col2_col3_col4')
)现在，它看起来像预期的那样工作。
希望这对未来的某人有所帮助。不过，如果有人有更好的主意，我很感兴趣。:)
https://stackoverflow.com/questions/56441135
复制相似问题
腾讯云开发者
