Отметить одну строку в таблице в SQLAlchemy

Question

Представьте себе таблицу с двумя столбцами: id (целое число) и mark (логическое значение). Эта таблица может иметь произвольное количество строк, но ровно в одной из этих строк в столбце mark должно быть указано True.

Если я изменю базу данных для пометки другой записи от mark до True, то система должна сначала снять отметку с предыдущей записи, а затем отметить запрошенную запись.

Как бы вы подошли к этому в Python / SQLAlchemy?

Answer 1

Два комментария выше имеют правду Триггер - это хороший способ сделать это, а также шаблон «много ложных, одно истинное» предполагает, что, возможно, для ссылки на «истинную» строку может использоваться другая таблица, или даже что вся «истинная» строка может быть в другом месте. , Обычная модель здесь заключается в том, что ваша таблица хранит версионную информацию, а «True» представляет текущую «версию». У меня обычно либо есть «текущая» версия, на которую ссылается родительская запись, либо я использую отдельную таблицу с именем «history» для всех «не текущих» строк.

В любом случае, давайте посмотрим, какой самый быстрый способ сделать именно то, что вы просите в SQLAlchemy. Мы сделаем почти то же, что и триггер INSERT / UPDATE, с помощью событий ORM:

from sqlalchemy import Column, Integer, Boolean, create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event
Base = declarative_base()

class Widget(Base):
    __tablename__ = 'widget'
    id = Column(Integer, primary_key=True)
    is_current_widget = Column(Boolean, default=False,
                        nullable=False)

@event.listens_for(Widget, "after_insert")
@event.listens_for(Widget, "after_update")
def _check_current(mapper, connection, target):
    if target.is_current_widget:
        connection.execute(
            Widget.__table__.
                update().
                values(is_current_widget=False).
                where(Widget.id!=target.id)
        )

e = create_engine('sqlite://', echo=True)
Base.metadata.create_all(e)

s = Session(e)

w1, w2, w3, w4, w5 = [Widget() for i in xrange(5)]
s.add_all([w1, w2, w3, w4, w5])
s.commit()

# note each call to commit() expires
# the values on all the Widgets so that
# is_current_widget is refreshed.

w2.is_current_widget = True
s.commit()

assert w2.is_current_widget
assert not w5.is_current_widget

w4.is_current_widget = True
s.commit()

assert not w2.is_current_widget
assert not w5.is_current_widget
assert w4.is_current_widget

# test the after_insert event

w6 = Widget(is_current_widget=True)
s.add(w6)
s.commit()

assert w6.is_current_widget
assert not w5.is_current_widget
assert not w4.is_current_widget