Use do_orm_execute for DML driven invalidation

[whabanks]

Summary | Résumé

Experimentally branched off of: #3033

Many dao methods for updating / deleting leverage legacy sqlalchemy DML statements Query.update() and Query.delete(). These DML queries bypass the typical session ORM events like after_commit and after_flush. This prevents us from relying solely on ORM events for cache invalidation.

However, modernized DML syntax (session.execute(<someQuery>)) can be intercepted via the do_orm_execute event allowing us to follow the same event driven invalidation pattern for both session ORM changes and DML statement changes.

• Add do_orm_execute event listener
• Add cache_invalidating_dml wrapper to execute DML statements and attach cache metadata to the session.info
• Add _queue_cache_invalidation to store deduplicated cache invalidation actions to be executed post-commit
• Migrate / modernize DML statements from legacy Query.update()/delete() to session.execute(<query))

How does this work?

Bulk DML queries in SQLAlchemy bypass session.dirty and after_flush instance detection. Statements like the one below are both legacy syntax from v1.4 and do not trigger typical session ORM events. (after_flush, after_commit, etc.)

# Legacy syntax & no event triggers here
db.session.query(User).filter_by(id=user.id).update(updates)

The helper method cache_invalidating_dml() adds cache metadata to the SQL statement that can be later extracted. Note the updated query syntax, this facilitates listening to do_orm_execute events and is central to enabling event driven invalidation for DML queries.

statement = cache_invalidating_dml(
    delete(Permission).where(Permission.user_id == user.id),
    user_id=user.id,
)

db.session.execute(statement)

Behind the scenes cache metadata is included:

{
    "cache_invalidation_entity_ids": {
        "user_id": user.id,
    }
}

_intercept_bulk_operations() is attached to the do_orm_execute event. When db.session.execute(statement) is run, it executes and:

1. Verifies the statement is an update or delete
2. Determines the mapped model is of type Permission
3. calls _queue_model_invalidations()
   • Note: We queue the invalidation(s) in the transaction so that we can wait for either the after_commit or after_rollback events to fire, so that we only invalidate after successful mutation, or drop the invalidations when rolling back
4. Looks up the Permission entity in the CACHE_INVALIDATION_REGISTRY
5. Validates that the required id's are present and builds the invalidation actions
6. after_commit event fires, signalling successful data changes and _invalidate_cache_after_commit is called to invalidate the key

At a high level

flowchart TD
    A[Database mutation] --> B{Mutation style}

    B -->|ORM objects| C[Flush]
    C --> D[after_flush]
    D --> E[Resolve model through registry]

    B -->|Bulk DML| F[cache_invalidating_dml]
    F --> G[Session.execute]
    G --> H[do_orm_execute]
    H --> E

    E --> I[Store namespace and ID in Session.info]
    I --> J{Transaction result}

    J -->|Commit| K[after_commit]
    K --> L[Pop pending set]
    L --> M[Delete grouped Redis keys]

    J -->|Rollback| N[after_rollback]
    N --> O[Discard pending set]

Loading

Benefits of this approach

1. Developers need only to do two things in any given situation to leverage dogpile while everything else is generically taken care of in the background.

ORM session based actions

1. Annotate the dao method, specifying the group it belongs to. User for example:

@cache_on_arguments(namespace="user", group_by="user_id")
def dao_get_user_by_id(user_id) -> dict:
    . . .

2. Check that the CACHE_INVALIDATION_REGISTRY entries for the associated entity exist and clear the keys that it should.

    User: [
        CacheInvalidationRule(namespace="user", entity_id_attribute="id"),
    ],

DML based statements

1. Wrap any bulk update or delete statements that affect cached values with the helper method

statement = update(User).where(User.id == usr.id).values(**updates)
db.session.execute(cache_invalidating_dml(statement, id=usr.id))

2. Check that the CACHE_INVALIDATION_REGISTRY entries for the associated entity exist and clear the keys that it should.

    User: [
        CacheInvalidationRule(namespace="user", entity_id_attribute="id"),
    ],

2. The current approach is compatible with threaded workers

• SQLAlchemy sessions are scoped per request context. Concurrent requests should have distinct sessions
  • Therefore the events we are listening to are also run against distinct, scoped sessions
• While the CACHE_INVALIDATION_REGISTRY is technically global, it's read-only at runtime and will never be mutated.
• Redis clients (used for scan/delete invalidations) use thread-safe connection pools. Each invalidate_group_keys call maintains it's own local scan cursor, therefore parallel scans should not corrupt one another.

Drawbacks

1. Cache invalidation is synchronous with the SQLAlchemy commit path, and we add a small amount of overhead to DB write operations.

database COMMIT
    -> SQLAlchemy after_commit
    -> Redis SCAN
    -> Redis DELETE
    -> commit() returns to caller

This would add some latency to API responses when executing an update or delete. The silver lining here is that all subsequent gets on that entity would hit cache cache instead of the DB and thus should offset added latency. You trade one slightly longer API call for many subsequent faster API calls.

Test instructions | Instructions pour tester la modification

TODO: Fill in test instructions for the reviewer.

Release Instructions | Instructions pour le déploiement

None.

Reviewer checklist | Liste de vérification du réviseur

• This PR does not break existing functionality.
• This PR does not violate GCNotify's privacy policies.
• This PR does not raise new security concerns. Refer to our GC Notify Risk Register document on our Google drive.
• This PR does not significantly alter performance.
• Additional required documentation resulting of these changes is covered (such as the README, setup instructions, a related ADR or the technical documentation).

⚠ If boxes cannot be checked off before merging the PR, they should be moved to the "Release Instructions" section with appropriate steps required to verify before release. For example, changes to celery code may require tests on staging to verify that performance has not been affected.

[Copilot]

🟡 Changes recommended

Bulk session.execute(update(...)) calls can leave in-session ORM instances stale unless session synchronization (or explicit expire/refresh) is applied, risking incorrect behavior for callers that use those instances after mutation.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR extends the ORM-event-driven cache invalidation approach to also cover bulk DML (UPDATE/DELETE) statements by intercepting SQLAlchemy’s do_orm_execute event and attaching cache metadata to session.execute(...) statements.

Changes:

• Add a do_orm_execute listener to queue cache invalidations for bulk ORM UPDATE/DELETE statements using execution options metadata.
• Introduce cache_invalidating_dml(...) to tag DML statements with the entity IDs required by the invalidation registry.
• Migrate several legacy Query.update()/delete() call sites to session.execute(update/delete(...)) and add/extend tests for rollback/commit invalidation behavior.

File summaries

File	Description
tests/app/cache/test_cache_events.py	Adds test coverage for DML-driven invalidation and rollback behavior.
app/dao/users_dao.py	Migrates user update paths to bulk update(User) wrapped with cache invalidation metadata.
app/dao/templates_dao.py	Migrates template bulk updates to update(Template) wrapped with cache invalidation metadata.
app/dao/service_permissions_dao.py	Converts permission removal to delete(ServicePermission) with DML invalidation tagging.
app/dao/permissions_dao.py	Converts permission deletions to delete(Permission) with DML invalidation tagging and reuse helper.
app/cache/cache_events.py	Adds DML interception via do_orm_execute and centralizes invalidation queueing helpers.
app/cache/cache_dml.py	Adds helper for tagging statements with invalidation entity IDs via execution options.

Review details

Suppressed comments (1)

app/dao/users_dao.py:51

• When update_dict is provided, save_model_user now issues a bulk UPDATE via session.execute(update(...)). Without synchronize_session (or expiring/refreshing usr), callers may observe stale values on the passed-in usr instance within the same session compared to the previous attribute-assignment implementation.

        _remove_values_for_keys_if_present(updates, ["id", "password_changed_at"])
        statement = update(User).where(User.id == usr.id).values(**updates)
        db.session.execute(cache_invalidating_dml(statement, id=usr.id))

• Files reviewed: 7/7 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[jimleroyer]

@copilot what do you mean that the users could remain stale? That these wouldn't be updated on the update? or after the update? what if we don't expect these user objects to change after and we are okay with these remaining stale?

[Copilot]

Stale means the bulk UPDATE changes the row in Postgres, but any User objects already loaded in the same SQLAlchemy session are not rewritten in memory automatically. If nothing reads those objects again, leaving them stale can be acceptable; the risk is that callers later in the same request/session can still observe old values. In this branch I made the two user bulk-update paths use synchronize_session="fetch" so the DAO is safe for callers that reuse the User after the update (2480c81).

[Copilot]

Stale means the bulk UPDATE changes the row in Postgres, but any User objects already loaded in the same SQLAlchemy session are not rewritten in memory automatically. If nothing reads those objects again, leaving them stale can be acceptable; the risk is that callers later in the same request/session can still observe old values. In this branch I made the two user bulk-update paths use synchronize_session="fetch" so the DAO is safe for callers that reuse the User after the update (2480c81).

[whabanks]

Yea this is the case here and is specific to the code that follows this save operation. By default [SQLAlchemy will choose synchronize_session='auto'](https://docs.sqlalchemy.org/en/21/orm/queryguide/dml.html#selecting-a-synchronization-strategy) which could resolve to either evaluate(doesn't refresh the model in session) orfetch` (does refresh).

Needing to manually specify this is extra cognitive load but I think it's a good thing to explicitly document and guarantee object synchronization when it's needed. We already explicitly specify this in our code:

• https://github.com/cds-snc/notification-api/blob/main/app/dao/api_key_dao.py#L84-L85
• https://github.com/cds-snc/notification-api/blob/main/app/dao/inbound_sms_dao.py#L108-L109

[jimleroyer]

Very nice additions. This likely needs some more logic to hook in the actual listener registration with _intercept_bulk_operations and more unit-tests.

[jimleroyer]

This is a string constant named _CACHE_INVALIDATION_ENTITY_IDS_OPTION defined on that line, as well as an import with the same name at line 13, but a tuple1(string). It's the end of the day so I might be confused, but yes I am confused atm. 😅

[whabanks]

Cleaned up!

[jimleroyer]

Let's add type annotations on parameters and return types.

[whabanks]

Added a bunch of type annotations. Since value here can be quite a few things, I defined a custom type EntityIdValue for this. Let me know what you think.

[jimleroyer]

Ah there we actually refer to the entity via entity_ids (and not model). 😛

[jimleroyer]

We didn't hook this function to any listener yet, right?

[whabanks]

This is hooked up to do_orm_execute in register_cache_orm_events

[jimleroyer]

cool cool

[jimleroyer]

@copilot what do you mean that the users could remain stale? That these wouldn't be updated on the update? or after the update? what if we don't expect these user objects to change after and we are okay with these remaining stale?