在 sqlalchemy 中触发

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)

所以，我想在 obs.state 更改为值 2 时更新相关的 task.state。目前我正在手动进行（使用称为任务的关系）

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，但找不到路。不知道有没有可能。

Answer 1

您可以使用DDL class在数据库中创建触发器：

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。您必须再添加一个触发器或事件处理程序以支持双向更改传播。