Add history events for changes made during deployments (#1600)

* Add history events for changes made during deployments

* Save history inline

* Remove save history events callable from deployment context

Author: shangyian

datajunction-server/datajunction_server/api/deployments.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import logging
-from typing import Callable
 import uuid
 
 from fastapi import Depends, BackgroundTasks, Request
@@ -16,7 +15,6 @@
 from datajunction_server.internal.caching.cachelib_cache import get_cache
 from datajunction_server.internal.caching.interface import Cache
 from datajunction_server.service_clients import QueryServiceClient
-from datajunction_server.api.helpers import get_save_history
 from datajunction_server.models.deployment import (
     DeploymentResult,
     DeploymentSpec,
@@ -159,7 +157,6 @@ async def create_deployment(
     session: AsyncSession = Depends(get_session),
     current_user: User = Depends(get_current_user),
     query_service_client: QueryServiceClient = Depends(get_query_service_client),
-    save_history: Callable = Depends(get_save_history),
     cache: Cache = Depends(get_cache),
     validate_access: access.ValidateAccessFn = Depends(
         validate_access,
@@ -176,7 +173,6 @@ async def create_deployment(
             current_user=current_user,
             request=request,
             query_service_client=query_service_client,
-            save_history=save_history,
             validate_access=validate_access,
             background_tasks=background_tasks,
             cache=cache,

datajunction-server/datajunction_server/internal/deployment/orchestrator.py

@@ -89,6 +89,8 @@
     LinkType,
 )
 from datajunction_server.models.history import ActivityType
+from datajunction_server.database.history import History
+from datajunction_server.internal.history import EntityType
 from datajunction_server.models.node import (
     NodeStatus,
     NodeType,
@@ -638,7 +640,6 @@ async def _bulk_delete_links(self, to_delete, node_spec) -> list[DeploymentResul
                     and link.role == delete_link.role
                 ):
                     link_ids_to_delete.append(link.id)  # type: ignore
-                    # await self.session.delete(link)
                     delete_results.append(
                         DeploymentResult(
                             name=link_name,
@@ -647,6 +648,23 @@ async def _bulk_delete_links(self, to_delete, node_spec) -> list[DeploymentResul
                             operation=DeploymentResult.Operation.DELETE,
                         ),
                     )
+
+                    # Track history for link deletion
+                    self.session.add(
+                        History(
+                            entity_type=EntityType.LINK,
+                            entity_name=node.name,
+                            node=node.name,
+                            activity_type=ActivityType.DELETE,
+                            details={
+                                "dimension_node": delete_link.rendered_dimension_node,
+                                "role": delete_link.role,
+                                "deployment_id": self.deployment_id,
+                            },
+                            user=self.context.current_user.username,
+                        ),
+                    )
+
         if link_ids_to_delete:
             await self.session.execute(
                 DimensionLink.__table__.delete().where(
@@ -754,6 +772,31 @@ async def _create_or_update_dimension_link(
             )
         role_suffix = f"[{link_spec.role}]" if link_spec.role else ""
         await self.session.flush()
+
+        # Track history for dimension link create/update (skip REFRESH/NOOP)
+        if activity_type in (ActivityType.CREATE, ActivityType.UPDATE):
+            link_details = {
+                "dimension_node": dimension_node.name,
+                "link_type": link_spec.type,
+                "role": link_spec.role,
+                "deployment_id": self.deployment_id,
+            }
+            if link_spec.type == LinkType.JOIN:
+                join_link = cast(DimensionJoinLinkSpec, link_spec)
+                link_details["join_type"] = join_link.join_type
+                link_details["join_on"] = join_link.rendered_join_on
+
+            self.session.add(
+                History(
+                    entity_type=EntityType.LINK,
+                    entity_name=new_revision.name,
+                    node=new_revision.name,
+                    activity_type=activity_type,
+                    details=link_details,
+                    user=self.context.current_user.username,
+                ),
+            )
+
         return DeploymentResult(
             name=f"{new_revision.name} -> {dimension_node.name}" + role_suffix,
             deploy_type=DeploymentResult.Type.LINK,
@@ -1132,6 +1175,24 @@ async def _create_cubes_from_validation(
                     )
                 )
 
+                # Track history for cube create/update operations
+                activity_type = ActivityType.UPDATE if existing else ActivityType.CREATE
+                self.session.add(
+                    History(
+                        entity_type=EntityType.NODE,
+                        entity_name=cube_spec.rendered_name,
+                        node=cube_spec.rendered_name,
+                        activity_type=activity_type,
+                        details={
+                            "version": new_node.current_version,
+                            "deployment_id": self.deployment_id,
+                            "metrics": cube_spec.rendered_metrics,
+                            "dimensions": cube_spec.rendered_dimensions,
+                        },
+                        user=self.context.current_user.username,
+                    ),
+                )
+
                 # Create deployment result
                 deployment_result = DeploymentResult(
                     name=cube_spec.rendered_name,
@@ -1219,12 +1280,16 @@ async def _delete_nodes(self, to_delete: list[NodeSpec]) -> list[DeploymentResul
         ]
 
     async def _deploy_delete_node(self, name: str) -> DeploymentResult:
+        async def add_history(event, session):
+            """Add history to session without committing"""
+            session.add(event)
+
         try:
             await hard_delete_node(
                 name=name,
                 session=self.session,
                 current_user=self.context.current_user,
-                save_history=self.context.save_history,
+                save_history=add_history,
             )
             return DeploymentResult(
                 name=name,
@@ -1523,6 +1588,23 @@ async def _process_valid_node_deploy(
         self.session.add(new_node)
         self.session.add(new_revision)
         await self.session.flush()
+
+        # Track history for create/update operations
+        activity_type = ActivityType.UPDATE if existing else ActivityType.CREATE
+        self.session.add(
+            History(
+                entity_type=EntityType.NODE,
+                entity_name=result.spec.rendered_name,
+                node=result.spec.rendered_name,
+                activity_type=activity_type,
+                details={
+                    "version": new_node.current_version,
+                    "deployment_id": self.deployment_id,
+                },
+                user=self.context.current_user.username,
+            ),
+        )
+
         deployment_result = DeploymentResult(
             name=result.spec.rendered_name,
             deploy_type=DeploymentResult.Type.NODE,

datajunction-server/datajunction_server/internal/deployment/utils.py

@@ -1,4 +1,3 @@
-from typing import Callable
 from fastapi import Request, BackgroundTasks
 
 from collections import defaultdict
@@ -115,7 +114,6 @@ class DeploymentContext:
     current_user: User
     request: Request
     query_service_client: QueryServiceClient
-    save_history: Callable
     validate_access: access.ValidateAccessFn
     background_tasks: BackgroundTasks
     cache: Cache

datajunction-server/tests/api/deployments_test.py

@@ -2853,3 +2853,194 @@ async def test_print_roads_spec(roads_nodes):
     )
     print("Roads Spec:", json.dumps(spec.model_dump()))
     assert 1 == 2
+
+
+@pytest.mark.xdist_group(name="deployments")
+class TestDeploymentHistoryTracking:
+    """Tests for history tracking during YAML deployments"""
+
+    @pytest.mark.asyncio
+    async def test_deployment_creates_history_for_nodes(
+        self,
+        client,
+        default_hard_hats,
+        default_us_states,
+        default_us_state,
+    ):
+        """
+        Test that deploying nodes creates history entries for each create/update operation
+        """
+        namespace = "history_test"
+        dim_spec = DimensionSpec(
+            name="default.hard_hat",
+            description="""Hard hat dimension""",
+            query="""
+                SELECT
+                    hard_hat_id,
+                    last_name,
+                    first_name,
+                    state
+                FROM ${prefix}default.hard_hats
+            """,
+            primary_key=["hard_hat_id"],
+            owners=["dj"],
+        )
+
+        # Deploy nodes
+        data = await deploy_and_wait(
+            client,
+            DeploymentSpec(
+                namespace=namespace,
+                nodes=[
+                    default_hard_hats,
+                    default_us_states,
+                    default_us_state,
+                    dim_spec,
+                ],
+            ),
+        )
+        assert data["status"] == DeploymentStatus.SUCCESS.value
+
+        # Check history for source node
+        response = await client.get(
+            f"/history/node/{namespace}.default.hard_hats/",
+        )
+        assert response.status_code == 200
+        history = response.json()
+        assert len(history) >= 1
+        # Find the create event
+        create_events = [h for h in history if h["activity_type"] == "create"]
+        assert len(create_events) == 1
+        assert create_events[0]["entity_type"] == "node"
+        assert "deployment_id" in create_events[0]["details"]
+
+        # Check history for dimension node
+        response = await client.get(
+            f"/history/node/{namespace}.default.hard_hat/",
+        )
+        assert response.status_code == 200
+        history = response.json()
+        assert len(history) >= 1
+        create_events = [h for h in history if h["activity_type"] == "create"]
+        assert len(create_events) == 1
+        assert create_events[0]["entity_type"] == "node"
+        assert "deployment_id" in create_events[0]["details"]
+
+    @pytest.mark.asyncio
+    async def test_deployment_creates_history_for_updates(
+        self,
+        client,
+        default_hard_hats,
+    ):
+        """
+        Test that updating nodes via deployment creates history entries
+        """
+        namespace = "history_update_test"
+
+        # First deployment - create
+        data = await deploy_and_wait(
+            client,
+            DeploymentSpec(
+                namespace=namespace,
+                nodes=[default_hard_hats],
+            ),
+        )
+        assert data["status"] == DeploymentStatus.SUCCESS.value
+
+        # Second deployment - update description
+        updated_hard_hats = SourceSpec(
+            name=default_hard_hats.name,
+            description="Updated description for hard hats table",
+            catalog=default_hard_hats.catalog,
+            schema=default_hard_hats.schema_,
+            table=default_hard_hats.table,
+            columns=default_hard_hats.columns,
+            owners=default_hard_hats.owners,
+        )
+        data = await deploy_and_wait(
+            client,
+            DeploymentSpec(
+                namespace=namespace,
+                nodes=[updated_hard_hats],
+            ),
+        )
+        assert data["status"] == DeploymentStatus.SUCCESS.value
+
+        # Check history shows both create and update
+        response = await client.get(
+            f"/history/node/{namespace}.default.hard_hats/",
+        )
+        assert response.status_code == 200
+        history = response.json()
+        create_events = [h for h in history if h["activity_type"] == "create"]
+        update_events = [h for h in history if h["activity_type"] == "update"]
+        assert len(create_events) >= 1
+        assert len(update_events) >= 1
+        # Verify deployment_id is tracked
+        assert "deployment_id" in create_events[0]["details"]
+        assert "deployment_id" in update_events[0]["details"]
+
+    @pytest.mark.asyncio
+    async def test_deployment_creates_history_for_dimension_links(
+        self,
+        client,
+        default_hard_hats,
+        default_us_states,
+        default_us_state,
+    ):
+        """
+        Test that deploying nodes with dimension links creates history entries for links
+        """
+        namespace = "history_link_test"
+        dim_spec = DimensionSpec(
+            name="default.hard_hat",
+            description="""Hard hat dimension""",
+            query="""
+                SELECT
+                    hard_hat_id,
+                    last_name,
+                    first_name,
+                    state
+                FROM ${prefix}default.hard_hats
+            """,
+            primary_key=["hard_hat_id"],
+            dimension_links=[
+                DimensionJoinLinkSpec(
+                    dimension_node="${prefix}default.us_state",
+                    join_type="inner",
+                    join_on="${prefix}default.hard_hat.state = ${prefix}default.us_state.state_short",
+                ),
+            ],
+            owners=["dj"],
+        )
+
+        # Deploy nodes with dimension links
+        data = await deploy_and_wait(
+            client,
+            DeploymentSpec(
+                namespace=namespace,
+                nodes=[
+                    default_hard_hats,
+                    default_us_states,
+                    default_us_state,
+                    dim_spec,
+                ],
+            ),
+        )
+        assert data["status"] == DeploymentStatus.SUCCESS.value
+
+        # Check history for dimension links
+        response = await client.get(
+            f"/history?node={namespace}.default.hard_hat",
+        )
+        assert response.status_code == 200
+        history = response.json()
+
+        # Should have link creation history
+        link_events = [h for h in history if h["entity_type"] == "link"]
+        assert len(link_events) >= 1
+        link_create_events = [h for h in link_events if h["activity_type"] == "create"]
+        assert len(link_create_events) >= 1
+        # Verify link details are tracked
+        assert "dimension_node" in link_create_events[0]["details"]
+        assert "deployment_id" in link_create_events[0]["details"]