使用具有唯一约束的 SQLAlchemy 的 on_conflict_do_update()

Question

我正在使用 SQLAlchemy 1.3.4 和 PostgreSQL 11.3。

\n\n

我有以下（简化的）表定义：

\n\n

class MyModel(Base):\n    __tablename__ = \'mymodel\'\n\n    id = Column(Integer, primary_key=True)\n    col1 = Column(Unicode, nullable=False)\n    col2 = Column(Unicode, nullable=False)\n    col3 = Column(Unicode, nullable=False)\n    col4 = Column(Boolean)\n\n    created_at = Column(DateTime(timezone=True), nullable=False)\n    updated_at = Column(DateTime(timezone=True), nullable=False)\n\n    __table_args__ = (\n        Index(\'uq_mymodel_col1_col2_col3_col4\',\n              col1, col2, col3, col4,\n              unique=True, postgresql_where=col4.isnot(None)),\n        Index(\'uq_mymodel_col1_col2_col3\',\n              col1, col2, col3,\n              unique=True, postgresql_where=col4.is_(None)),\n    )\n

\n\n

（我必须创建 2 个唯一索引而不是 a，UniqueConstraint因为 aUniqueConstraint将允许多行具有相同的(col1, col2, col3)iscol4为空，这是我不想要的）

\n\n

我正在尝试执行以下查询：

\n\n

INSERT INTO mymodel (col1, col2, col3, col4, created_at, updated_at)\nVALUES (%(col1)s, %(col2)s, %(col3)s, %(col4)s, %(created_at)s, %(updated_at)s)\nON CONFLICT DO UPDATE SET updated_at = %(param_1)s\nRETURNING mymodel.id\n

\n\n

但我不知道如何正确使用 SQLAlchemy on_conflict_do_update()。:-/

\n\n

这是我尝试过的：

\n\n

values = {\xe2\x80\xa6}\n\nstmt = insert(MyModel.__table__).values(**values)\nstmt = stmt.returning(MyModel.__table__.c.id)\nstmt = stmt.on_conflict_do_update(set_={\'updated_at\': values[\'updated_at\']})\nresult = dbsession.connection().execute(stmt)\n

\n\n

然而 SQLAlchemy 抱怨道：Either constraint or index_elements, but not both, must be specified unless DO NOTHING

\n\n

我发现非常不清楚如何使用constraintor index_elements。

\n\n

我尝试了一些方法，但没有成功。例如：

\n\n

values = {\xe2\x80\xa6}\n\nstmt = insert(MyModel.__table__).values(**values)\nstmt = stmt.returning(MyModel.__table__.c.id)\nstmt = stmt.on_conflict_do_update(constraint=\'uq_mymodel_col1_col2_col3_col4\'\n                                  set_={\'updated_at\': values[\'updated_at\']})\nresult = dbsession.connection().execute(stmt)\n

\n\n

但这也不起作用：constraint "uq_mymodel_col1_col2_col3_col4" for table "mymodel" does not exist. 但它确实存在。（我什至从 pgsql 复制粘贴以确保我没有输入错误）

\n\n

无论如何，我有两个独特的约束可能会引发冲突，但on_conflict_do_update()似乎只采用一个。所以我也尝试像这样指定两者：

\n\n

values = {\xe2\x80\xa6}\n\nstmt = insert(MyModel.__table__).values(**values)\nstmt = stmt.returning(MyModel.__table__.c.id)\nstmt = stmt.on_conflict_do_update(constraint=\'uq_mymodel_col1_col2_col3_col4\'\n                                  set_={\'updated_at\': values[\'updated_at\']})\nstmt = stmt.on_conflict_do_update(constraint=\'uq_mymodel_col1_col2_col3\'\n                                  set_={\'updated_at\': values[\'updated_at\']})\nresult = dbsession.connection().execute(stmt)\n

\n\n

但我得到了同样的错误，即uq_mymodel_col1_col2_col3_col4不存在。

\n\n

此时我只是不知道如何执行上述查询，并且非常感谢一些帮助。

\n

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')
    )

现在它似乎正在按预期工作。

希望对将来的人有所帮助。如果有人有更好的主意，我很感兴趣。:)