Cascading after_update operations

[FrankSiderio]

We have a parent/child table and want to be able to propagate updates up from the children/leaves to the parent root.

For example, given we have a structure like

{
  id: 1,
  state: "in-progress",
  children: [{
    id: 2,
    state: "in-progress",
    children: [{
      id: 3,
      state: "in-progress"
    }]
  }]
}

When child 3 state gets updated to completed we want to update the state of the parent 2 and its parent 1 to completed as well. We can do that when it's one level of separation, but for two levels the root node never gets updated. So node 2 will get updated to completed but 1 is stuck in-progress

Here's what we have

class OperationNode(db.Model):
    __tablename__ = "operation_node"

    id = Column(Integer, primary_key=True)

    # child -> parent
    parent_id = Column(
        Integer,
        ForeignKey("operation_node.id", ondelete="CASCADE", onupdate="CASCADE"),
        index=True,
    )

    # Track progress
    state = Column(String(15), default=OperationNodeStates.in_progress.value)


    # List of child objects with eager loading and max depth
    children = relationship("OperationNode",
                            foreign_keys=[parent_id],
                            cascade="all, delete-orphan",
                            lazy="joined",
                            join_depth=2)


@event.listens_for(OperationNode, "after_update")
def after_update_trigger_parent_sync(mapper, connection, operation_node):
    insp = db.inspect(operation_node)
    state_history = insp.attrs["state"].load_history()

    # Update aggregated fields of parent when a child has changes
    if operation_node.parent_id \
            and state_history.has_changes():
        operation_node_table = OperationNode.__table__
        connection.execute(
            operation_node_table.update().
            where(operation_node_table.c.id == operation_node.parent_id).
            values(state="completed")
        )

so when we do

operation_node = OperationNode.query.get(3)
operation_node.state = "completed"
db.session.merge(operation_node)
db.session.commit()

only the direct parent 2 gets updated from the after_update and not the root 1. The after_update doesn't trigger the second time :(. Would be happy to share more code/context if needed.

Are we missing something? Or is there a good example of this same use case out there? Any help is appreciated as we are very stuck!

Using sqlalchemy 1.3.24 and Flask-SQLAlchemy 2.4.4

This is copied from my stack overflow question here

[zzzeek]

OK so two general strategies:

1. cascade upwards in your after_update_trigger using SQL
2. use the before_flush() event handler instead and cascade the operation using the object references.

For option one, you need to query the database to get a hold of not just id=2, but also id=1. the after_update event is run after the session has already figured out all the changes it will be making, so there's no automatic system that an UPDATE on row id=2 would cause the unit of work to actually trigger on that row.

in other words, inside of after_update(), you need to emit the SQL for everything that needs to happen, not just part of it.

if you're on PostgreSQL you could probably write a single UPDATE statement that uses WITH RECURSIVE to update all the rows at once, background on this syntax is at https://www.postgresql.org/docs/current/queries-with.html which SQLAlchemy supports.

otherwise, the less fancy way is to keep SELECTing the parent row until parent_id=NULL and run an UPDATE on each one, something like this

@event.listens_for(OperationNode, "after_update")
def after_update_trigger_parent_sync(mapper, connection, operation_node):
    insp = db.inspect(operation_node)
    state_history = insp.attrs["state"].load_history()

    # Update aggregated fields of parent when a child has changes
    if operation_node.parent_id \
            and state_history.has_changes():
        operation_node_table = OperationNode.__table__

        parent_id = operation_node.parent_id
        while parent_id is not None:
            connection.execute(
                operation_node_table.update().
                where(operation_node_table.c.id == parent_id).
                values(state="completed")
            )
            parent_id = connection.scalar(
                select(operation_node_table.c.parent_id).
                where(operation_node_table.c.id == parent_id)
            )

[CaselIT]

Wouldn't it also work to instead load the parent OperationNode with the session and update it by setting its status, so that a new event is triggered for the parted etc.

Of course the above version is more optimized since it does not need to trigger multiple events

[zzzeek]

yes, that would be the before_flush approach.

[zzzeek]

taht is, dont use the Session to load things inside of the after_update handler, that kind of thing can be only in before_flush or not in an event at all.

[CaselIT]

taht is, dont use the Session to load things inside of the after_update handler, that kind of thing can be only in before_flush or not in an event at all.

ok, there is a dragon in the docs that mentions not to load other stuff with the session

[FrankSiderio]

This is all very helpful so thank you! It sounds like the solution suggested above (cascade upwards in your after_update_trigger using SQL) would be the best approach?

Initially, we wanted to delegate this task using something like sqs because we're using this for long running processes. But that was giving us the same issues with after_update not triggering. Our biggest concern with a different approach would be scale.