Триггер в sqlachemy

Question

У меня есть две таблицы, связанные через внешний ключ, здесь они используют декларативное сопоставление

class Task(DeclarativeBase):
    __tablename__ = 'task'
    id = Column(Integer, primary_key=True)
    state = Column(Integer, default=0)
    obs_id = Column(Integer, ForeignKey('obs.id'), nullable=False)

class Obs(DeclarativeBase):
    __tablename__ = 'obs'
    id = Column(Integer, primary_key=True)
    state = Column(Integer, default=0)

Итак, я хотел бы обновить связанный файл task.state, когда obs.state изменяется на значение 2.В настоящее время я делаю это вручную (используя отношение, называемое задачей)

obs.state = 2
obs.task.state = 2

Но я бы предпочел сделать это с помощью триггера.Я проверил, что это работает в sqlite

CREATE TRIGGER update_task_state UPDATE OF state ON obs
  BEGIN
    UPDATE task SET state = 2 WHERE (obs_id = old.id) and (new.state = 2);
  END;

Но я не могу найти, как выразить это в sqlalchemy.Я прочитал вставить обновление по умолчанию несколько раз, но не могу найти способ.Я не знаю, возможно ли это вообще.

Answer 1

Вы можете создать триггер в базе данных с помощью класса DDL :

update_task_state = DDL('''\
CREATE TRIGGER update_task_state UPDATE OF state ON obs
  BEGIN
    UPDATE task SET state = 2 WHERE (obs_id = old.id) and (new.state = 2);
  END;''')
event.listen(Obs.__table__, 'after_create', update_task_state)

Это самый надежный способ: он будет работать для массовых обновлений, когда ORM не используется, и даже дляобновления за пределами вашего приложения.Однако есть и недостатки:

• Вы должны следить за тем, чтобы ваш триггер существовал и был в актуальном состоянии;
• Он не переносимый, поэтому вам придется переписать его, если вы измените базу данных;
• SQLAlchemy не изменит новое состояние уже загруженного объекта, пока не истечет срок его действия (например, с помощью какого-либо обработчика событий).

Ниже приведен менее надежный (он будет работать при внесении изменений вТолько на уровне ORM), но гораздо более простое решение:

from sqlalchemy.orm import validates

class Obs(DeclarativeBase):
    __tablename__ = 'obs'
    id = Column(Integer, primary_key=True)
    state = Column(Integer, default=0)
    @validates('state')
    def update_state(self, key, value):
        self.task.state = value
        return value

Оба моих примера работают в одном направлении, то есть они обновляют задачу, когда изменяется obs, но не касаются obs, когда задача обновляется.Вы должны добавить еще один обработчик триггера или события для поддержки распространения изменений в обоих направлениях.