Блог

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

 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. В настоящее время я делаю это вручную (используя отношение, называемое task)

 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. Я несколько раз прочитал insert update defaults, но не могу найти способ. Я не знаю, возможно ли это вообще.

Вы можете создать триггер в базе данных с помощью класса 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 при обновлении задачи. Вам нужно добавить еще один триггер или обработчик событий для поддержки распространения изменений в обоих направлениях.