Использование onAconflict_do_update () SQLAlchemy с уникальными ограничениями

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

(мне нужно было создать 2 уникальных индекса, а не 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 жалуется: 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 не существует.

На данный момент я просто не могу понять, как выполнить вышеуказанный запрос, и буду очень признателен за помощь.

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

А теперь, похоже, работает как положено.

Надеюсь, это поможет кому-то в будущем. Если у кого-то есть идея получше, мне интересно. :)