trigger in sqlachemy

 

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)

This is the most reliable way: it will work for bulk updates when ORM is not used and even for updates outside your application. However there disadvantages too:

• You have to take care your trigger exists and up to date;
• It's not portable so you have to rewrite it if you change database;
• SQLAlchemy won't change the new state of already loaded object unless you expire it (e.g. with some event handler).

Below is a less reliable (it will work when changes are made at ORM level only), but much simpler solution:

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