python - Trigger in sqlachemy

Question

I have two tables related via a foreign key, here they are using Declarative Mapping

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)

So, I would like to update the related task.state when obs.state is changed to value 2. Currently I'm doing it by hand (using a relationship called task)

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

But I would prefer doing it using a trigger. I have checked that this works in 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;

But I can't find how to express this in sqlalchemy. I have read insert update defaults several times, but can't find the way. I don't know if it's even possible.

Answer 1

You can create trigger in the database with 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)

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

Both my examples work one way, i.e. they update task when obs is changes, but don't touch obs when task is updated. You have to add one more trigger or event handler to support change propagation in both directions.