使用SQLAlchemy的on_conflict_do_update()与唯一约束条件

Question

我使用的是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)),
    )

我不得不创建两个唯一索引而不是一个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 不存在。

目前，我无法弄清如何执行上述查询，并真的很需要一些帮助。

- Mathieu Bridon

Answer 1

好的，我想我明白了。所以问题实际上并不是来自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')
    )

现在看起来它正如预期的那样工作。

希望这能帮到未来的一些人。如果有人有更好的想法，请告诉我。:)

- Mathieu Bridon