ENG-3324: Add system-stewards change hook for inheritance propagation

Temporary single-purpose callback registry under ``fides.api`` that lets
the fidesplus monitor-stewardship-inheritance feature react when a
user is added/removed as a data steward (system manager) of a system.

The three v1 system-manager mutation routes (``bulk_assign_steward``,
``update_managed_systems``, ``remove_user_as_system_manager``, and
the approver-demotion branch of ``update_user_permissions``) gain a
``BackgroundTasks`` injection and call
``notify_system_stewards_changed(background_tasks, system.id)`` after
each successful set/remove. The model methods themselves are
unchanged so callers from tests/conftest don't trigger propagation.

This is intentionally a tiny ~50-line shim. The module docstring
documents the cutover: when the event framework lands (fides
PR #8096), trigger sites swap to ``publish_after_commit(...)`` and
this file is deleted. Do not generalize this into a ``hooks`` package.

Pairs with fidesplus ``ENG-3324/monitor-steward-inheritance-propagation``,
which registers the inheritance-reconcile callback at app startup.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adamsachs

src/fides/api/system_steward_change_hooks.py

@@ -0,0 +1,59 @@
+"""
+System data-steward change hook — TEMPORARY EVENT-FRAMEWORK SHIM.
+
+A single-purpose callback registry that lets fidesplus react when a user is
+added/removed as a data steward (system manager) of a system. The only known
+consumer today is fidesplus's monitor-stewardship-inheritance propagation.
+
+This module exists ONLY because the event framework in fides PR #8096 is
+not yet merged. When it lands:
+
+  - Call sites of ``notify_system_stewards_changed`` (in the three v1
+    system-manager routes) are replaced with
+    ``publish_after_commit(session, SystemDataStewardsChanged(...))``.
+  - Fidesplus's registered callback is replaced with a
+    ``@subscribes_to(SystemDataStewardsChanged)`` handler.
+  - This file is deleted.
+
+DO NOT use this as a precedent for adding more cross-repo hooks. If you need
+a similar shim before the framework lands, create a separate single-purpose
+module — do not generalize this into a ``hooks`` package.
+"""
+
+from typing import Callable, List
+
+from fastapi import BackgroundTasks
+from loguru import logger
+
+SystemStewardChangeHook = Callable[[BackgroundTasks, str], None]
+
+_HOOKS: List[SystemStewardChangeHook] = []
+
+
+def register_system_steward_change_hook(hook: SystemStewardChangeHook) -> None:
+    """Register a callback invoked when a system's data stewards change.
+
+    Idempotent: registering the same hook twice is a no-op.
+    """
+    if hook not in _HOOKS:
+        _HOOKS.append(hook)
+
+
+def notify_system_stewards_changed(
+    background_tasks: BackgroundTasks, system_id: str
+) -> None:
+    """Invoke every registered hook with the given ``system_id``.
+
+    Each hook is wrapped in try/except so one failure doesn't suppress the rest;
+    failures are logged and swallowed. Hooks are responsible for scheduling
+    their own background work via ``background_tasks``.
+    """
+    for hook in _HOOKS:
+        try:
+            hook(background_tasks, system_id)
+        except Exception:
+            logger.exception(
+                "System-stewards-change hook {} raised for system_id={}",
+                hook,
+                system_id,
+            )

src/fides/api/v1/endpoints/system.py

@@ -1,6 +1,6 @@
 from typing import Annotated, Dict, List, Literal, Optional, Union
 
-from fastapi import Depends, HTTPException, Query, Response, Security
+from fastapi import BackgroundTasks, Depends, HTTPException, Query, Response, Security
 from fastapi_pagination import Page, Params
 from fastapi_pagination.bases import AbstractPage
 from fastapi_pagination.ext.sqlalchemy import paginate
@@ -57,6 +57,7 @@
     BasicSystemResponse,
     SystemResponse,
 )
+from fides.api.system_steward_change_hooks import notify_system_stewards_changed
 from fides.api.util.api_router import APIRouter
 from fides.api.util.connection_util import (
     delete_connection_config,
@@ -402,6 +403,7 @@ async def system_bulk_delete(
 )
 async def bulk_assign_steward(
     data: AssignStewardRequest,
+    background_tasks: BackgroundTasks,
     db: Session = Depends(deps.get_db),
 ) -> Dict:
     """Assign the given `data_steward` (username) as a system manager for the list of `system_keys`.
@@ -467,6 +469,7 @@ async def bulk_assign_steward(
     for system in systems:
         if user not in system.data_stewards:
             user.set_as_system_manager(db, system)
+            notify_system_stewards_changed(background_tasks, system.id)
             updated_count += 1
 
     return {

src/fides/api/v1/endpoints/user_endpoints.py

@@ -4,7 +4,7 @@
 from datetime import datetime, timedelta, timezone
 from typing import Dict, List, Optional
 
-from fastapi import Depends, HTTPException, Request, Response, Security
+from fastapi import BackgroundTasks, Depends, HTTPException, Request, Response, Security
 from fastapi.security import SecurityScopes
 from fastapi_pagination import Page, Params
 from fastapi_pagination.bases import AbstractPage
@@ -64,6 +64,7 @@
     UserResponse,
     UserUpdate,
 )
+from fides.api.system_steward_change_hooks import notify_system_stewards_changed
 from fides.api.util.api_router import APIRouter
 from fides.api.util.errors import FidesError, MessageDispatchException
 from fides.api.util.rate_limit import fides_limiter
@@ -341,6 +342,7 @@ def update_managed_systems(
     db: Session = Depends(deps.get_db),
     user_id: str,
     systems: List[FidesKey],
+    background_tasks: BackgroundTasks,
 ) -> List[SystemSchema]:
     """
     Endpoint to override the systems for which a user is "system manager".
@@ -379,11 +381,13 @@ def update_managed_systems(
     for system in retrieved_systems:
         if user not in system.data_stewards:
             user.set_as_system_manager(db, system)
+            notify_system_stewards_changed(background_tasks, system.id)
 
     # Removing systems for which the user in no longer a manager
     for system in user.systems.copy():
         if system not in retrieved_systems:
             user.remove_as_system_manager(db, system)
+            notify_system_stewards_changed(background_tasks, system.id)
 
     return user.systems
 
@@ -473,7 +477,11 @@ async def get_managed_system_details(
     status_code=HTTP_204_NO_CONTENT,
 )
 def remove_user_as_system_manager(
-    *, db: Session = Depends(deps.get_db), user_id: str, system_key: FidesKey
+    *,
+    db: Session = Depends(deps.get_db),
+    user_id: str,
+    system_key: FidesKey,
+    background_tasks: BackgroundTasks,
 ) -> None:
     """
     Endpoint to remove user as system manager from the given system
@@ -488,6 +496,7 @@ def remove_user_as_system_manager(
         )
 
     user.remove_as_system_manager(db, system)
+    notify_system_stewards_changed(background_tasks, system.id)
     logger.info("Removed user {} as system manager of {}", user_id, system.fides_key)
 
 

src/fides/api/v1/endpoints/user_permission_endpoints.py

@@ -1,6 +1,6 @@
 from typing import List, Optional
 
-from fastapi import Depends, HTTPException, Security
+from fastapi import BackgroundTasks, Depends, HTTPException, Security
 from fastapi.security import SecurityScopes
 from loguru import logger
 from sqlalchemy.orm import Session
@@ -16,6 +16,7 @@
     UserPermissionsEdit,
     UserPermissionsResponse,
 )
+from fides.api.system_steward_change_hooks import notify_system_stewards_changed
 from fides.api.util.api_router import APIRouter
 from fides.common import urn_registry as urls
 from fides.common.scope_registry import (
@@ -97,6 +98,7 @@ async def update_user_permissions(
     user_id: str,
     authorization: str = Security(oauth2_scheme),
     permissions: UserPermissionsEdit,
+    background_tasks: BackgroundTasks,
 ) -> FidesUserPermissions:
     """Update a user's role(s).  The UI assigns one role at a time, but multiple
     roles are technically supported.
@@ -128,6 +130,7 @@ async def update_user_permissions(
                 system.fides_key,
             )
             user.remove_as_system_manager(db, system)
+            notify_system_stewards_changed(background_tasks, system.id)
 
     return updated_user_perms
 

tests/api/test_system_steward_change_hooks.py

@@ -0,0 +1,61 @@
+"""Tests for the system-stewards change hook registry."""
+
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import BackgroundTasks
+
+from fides.api import system_steward_change_hooks as system_stewards
+
+
+@pytest.fixture(autouse=True)
+def isolated_registry(monkeypatch):
+    """Each test starts with an empty registry."""
+    monkeypatch.setattr(system_stewards, "_HOOKS", [])
+    yield
+
+
+def test_register_appends_hook():
+    hook = MagicMock()
+    system_stewards.register_system_steward_change_hook(hook)
+    assert system_stewards._HOOKS == [hook]
+
+
+def test_register_is_idempotent():
+    hook = MagicMock()
+    system_stewards.register_system_steward_change_hook(hook)
+    system_stewards.register_system_steward_change_hook(hook)
+    assert system_stewards._HOOKS == [hook]
+
+
+def test_notify_calls_each_registered_hook():
+    hook_a = MagicMock()
+    hook_b = MagicMock()
+    system_stewards.register_system_steward_change_hook(hook_a)
+    system_stewards.register_system_steward_change_hook(hook_b)
+
+    bg = BackgroundTasks()
+    system_stewards.notify_system_stewards_changed(bg, "sys-1")
+
+    hook_a.assert_called_once_with(bg, "sys-1")
+    hook_b.assert_called_once_with(bg, "sys-1")
+
+
+def test_notify_isolates_hook_failures():
+    """One hook raising must not prevent the rest from firing."""
+    raising = MagicMock(side_effect=RuntimeError("boom"))
+    survivor = MagicMock()
+    system_stewards.register_system_steward_change_hook(raising)
+    system_stewards.register_system_steward_change_hook(survivor)
+
+    bg = BackgroundTasks()
+    system_stewards.notify_system_stewards_changed(bg, "sys-2")
+
+    raising.assert_called_once()
+    survivor.assert_called_once_with(bg, "sys-2")
+
+
+def test_notify_with_no_hooks_is_noop():
+    bg = BackgroundTasks()
+    # Should not raise even if registry empty
+    system_stewards.notify_system_stewards_changed(bg, "sys-3")