Merge pull request #5692 from opsmill/pog-bramch-merge-node-mutations-IFC-1045

Add DiffEvents class to generate changelog events from a diff

Author: ogenstad

backend/infrahub/core/changelog/diff.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Sequence
+
+from infrahub_sdk.utils import str_to_bool
+
+from infrahub.core.constants import DiffAction, RelationshipCardinality
+from infrahub.core.constants.database import DatabaseEdgeType
+
+from .models import (
+    AttributeChangelog,
+    NodeChangelog,
+    RelationshipCardinalityManyChangelog,
+    RelationshipCardinalityOneChangelog,
+    RelationshipPeerChangelog,
+)
+
+if TYPE_CHECKING:
+    from infrahub.core.branch import Branch
+    from infrahub.core.diff.model.path import (
+        EnrichedDiffAttribute,
+        EnrichedDiffNode,
+        EnrichedDiffRelationship,
+        EnrichedDiffRoot,
+    )
+    from infrahub.core.schema import MainSchemaTypes
+    from infrahub.database import InfrahubDatabase
+
+
+@dataclass
+class NodeInDiff:
+    node_id: str
+    kind: str
+    label: str
+
+
+class DiffChangelogCollector:
+    def __init__(self, diff: EnrichedDiffRoot, branch: Branch, db: InfrahubDatabase) -> None:
+        self._diff = diff
+        self._branch = branch
+        self._db = db
+        self._diff_nodes: dict[str, NodeInDiff]
+
+    def _populate_diff_nodes(self) -> None:
+        self._diff_nodes = {
+            node.uuid: NodeInDiff(node_id=node.uuid, kind=node.kind, label=node.label) for node in self._diff.nodes
+        }
+
+    def get_node(self, node_id: str) -> NodeInDiff:
+        return self._diff_nodes[node_id]
+
+    def _process_node(self, node: EnrichedDiffNode) -> NodeChangelog:
+        node_changelog = NodeChangelog(node_id=node.uuid, node_kind=node.kind, display_label=node.label)
+        schema = self._db.schema.get(node_changelog.node_kind, branch=self._branch, duplicate=False)
+        for attribute in node.attributes:
+            self._process_node_attribute(node=node_changelog, attribute=attribute, schema=schema)
+
+        for relationship in node.relationships:
+            self._process_node_relationship(node=node_changelog, relationship=relationship)
+
+        return node_changelog
+
+    def _process_node_attribute(
+        self, node: NodeChangelog, attribute: EnrichedDiffAttribute, schema: MainSchemaTypes
+    ) -> None:
+        schema_attribute = schema.get_attribute(name=attribute.name)
+        changelog_attribute = AttributeChangelog(name=attribute.name, kind=schema_attribute.kind)
+        for attr_property in attribute.properties:
+            match attr_property.property_type:
+                case DatabaseEdgeType.HAS_VALUE:
+                    # TODO deserialize correct value type from string
+                    changelog_attribute.value = attr_property.new_value
+                    changelog_attribute.value_previous = attr_property.previous_value
+                case DatabaseEdgeType.IS_PROTECTED:
+                    changelog_attribute.add_property(
+                        name="is_protected",
+                        value_current=self._convert_string_boolean_value(value=attr_property.new_value),
+                        value_previous=self._convert_string_boolean_value(value=attr_property.previous_value),
+                    )
+                case DatabaseEdgeType.IS_VISIBLE:
+                    changelog_attribute.add_property(
+                        name="is_visible",
+                        value_current=self._convert_string_boolean_value(value=attr_property.new_value),
+                        value_previous=self._convert_string_boolean_value(value=attr_property.previous_value),
+                    )
+                case DatabaseEdgeType.HAS_SOURCE:
+                    changelog_attribute.add_property(
+                        name="source",
+                        value_current=attr_property.new_value,
+                        value_previous=attr_property.previous_value,
+                    )
+                case DatabaseEdgeType.HAS_OWNER:
+                    changelog_attribute.add_property(
+                        name="owner",
+                        value_current=attr_property.new_value,
+                        value_previous=attr_property.previous_value,
+                    )
+
+        node.attributes[attribute.name] = changelog_attribute
+
+    def _process_node_relationship(self, node: NodeChangelog, relationship: EnrichedDiffRelationship) -> None:
+        match relationship.cardinality:
+            case RelationshipCardinality.ONE:
+                self._process_node_cardinality_one_relationship(node=node, relationship=relationship)
+
+            case RelationshipCardinality.MANY:
+                self._process_node_cardinality_many_relationship(node=node, relationship=relationship)
+
+    def _process_node_cardinality_one_relationship(
+        self, node: NodeChangelog, relationship: EnrichedDiffRelationship
+    ) -> None:
+        changelog_rel = RelationshipCardinalityOneChangelog(name=relationship.name)
+        for entry in relationship.relationships:
+            for rel_prop in entry.properties:
+                match rel_prop.property_type:
+                    case DatabaseEdgeType.IS_RELATED:
+                        if rel_prop.new_value:
+                            changelog_rel.peer_id = rel_prop.new_value
+                            changelog_rel.peer_kind = self.get_node(node_id=rel_prop.new_value).kind
+                        if rel_prop.previous_value:
+                            changelog_rel.peer_id_previous = rel_prop.previous_value
+                            changelog_rel.peer_kind_previous = self.get_node(node_id=rel_prop.previous_value).kind
+                    case DatabaseEdgeType.IS_PROTECTED:
+                        changelog_rel.add_property(
+                            name="is_protected",
+                            value_current=self._convert_string_boolean_value(value=rel_prop.new_value),
+                            value_previous=self._convert_string_boolean_value(value=rel_prop.previous_value),
+                        )
+                    case DatabaseEdgeType.IS_VISIBLE:
+                        changelog_rel.add_property(
+                            name="is_visible",
+                            value_current=self._convert_string_boolean_value(value=rel_prop.new_value),
+                            value_previous=self._convert_string_boolean_value(value=rel_prop.previous_value),
+                        )
+                    case DatabaseEdgeType.HAS_OWNER:
+                        changelog_rel.add_property(
+                            name="owner",
+                            value_current=rel_prop.new_value,
+                            value_previous=rel_prop.previous_value,
+                        )
+                    case DatabaseEdgeType.HAS_SOURCE:
+                        changelog_rel.add_property(
+                            name="source",
+                            value_current=rel_prop.new_value,
+                            value_previous=rel_prop.previous_value,
+                        )
+
+        node.add_relationship(relationship=changelog_rel)
+
+    def _convert_string_boolean_value(self, value: str | None) -> bool | None:
+        """Convert string based boolean for is_protected and is_visible."""
+        if value is not None:
+            return str_to_bool(value)
+
+        return None
+
+    def _process_node_cardinality_many_relationship(
+        self, node: NodeChangelog, relationship: EnrichedDiffRelationship
+    ) -> None:
+        changelog_rel = RelationshipCardinalityManyChangelog(name=relationship.name)
+        for peer in relationship.relationships:
+            peer_log = RelationshipPeerChangelog(
+                peer_id=peer.peer_id, peer_kind=self.get_node(node_id=peer.peer_id).kind, peer_status=peer.action
+            )
+            for peer_prop in peer.properties:
+                match peer_prop.property_type:
+                    case DatabaseEdgeType.IS_VISIBLE:
+                        peer_log.add_property(
+                            name="is_visible",
+                            value_current=self._convert_string_boolean_value(value=peer_prop.new_value),
+                            value_previous=self._convert_string_boolean_value(value=peer_prop.previous_value),
+                        )
+                    case DatabaseEdgeType.IS_PROTECTED:
+                        peer_log.add_property(
+                            name="is_protected",
+                            value_current=self._convert_string_boolean_value(value=peer_prop.new_value),
+                            value_previous=self._convert_string_boolean_value(value=peer_prop.previous_value),
+                        )
+                    case DatabaseEdgeType.HAS_OWNER:
+                        peer_log.add_property(
+                            name="owner",
+                            value_current=peer_prop.new_value,
+                            value_previous=peer_prop.previous_value,
+                        )
+                    case DatabaseEdgeType.HAS_SOURCE:
+                        peer_log.add_property(
+                            name="source",
+                            value_current=peer_prop.new_value,
+                            value_previous=peer_prop.previous_value,
+                        )
+
+            changelog_rel.peers.append(peer_log)
+
+        node.add_relationship(relationship=changelog_rel)
+
+    def collect_changelogs(self) -> Sequence[tuple[DiffAction, NodeChangelog]]:
+        self._populate_diff_nodes()
+        changelogs = [
+            (node.action, self._process_node(node=node))
+            for node in self._diff.nodes
+            if node.action != DiffAction.UNCHANGED
+        ]
+        return changelogs

backend/infrahub/core/changelog/models.py

@@ -47,8 +47,8 @@ def value_update_status(self) -> DiffAction:
 
 class AttributeChangelog(BaseModel):
     name: str = Field(..., description="The name of the attribute")
-    value: Any = Field(..., description="The current value of the attribute")
-    value_previous: Any = Field(..., description="The previous value of the attribute")
+    value: Any = Field(default=None, description="The current value of the attribute")
+    value_previous: Any = Field(default=None, description="The previous value of the attribute")
     properties: dict[str, PropertyChangelog] = Field(
         default_factory=dict, description="The properties that were updated during this update"
     )
@@ -155,6 +155,9 @@ class RelationshipPeerChangelog(BaseModel):
         default_factory=dict, description="Changes to properties of this relationship if any were made"
     )
 
+    def add_property(self, name: str, value_current: bool | str | None, value_previous: bool | str | None) -> None:
+        self.properties[name] = PropertyChangelog(name=name, value=value_current, value_previous=value_previous)
+
 
 class RelationshipCardinalityManyChangelog(BaseModel):
     name: str

backend/infrahub/core/diff/merger/merger.py

@@ -9,6 +9,7 @@
 
 if TYPE_CHECKING:
     from infrahub.core.branch import Branch
+    from infrahub.core.diff.model.path import EnrichedDiffRoot
     from infrahub.core.diff.repository.repository import DiffRepository
     from infrahub.core.timestamp import Timestamp
     from infrahub.database import InfrahubDatabase
@@ -33,7 +34,7 @@ def __init__(
         self.diff_repository = diff_repository
         self.serializer = serializer
 
-    async def merge_graph(self, at: Timestamp) -> None:
+    async def merge_graph(self, at: Timestamp) -> EnrichedDiffRoot:
         tracking_id = BranchTrackingId(name=self.source_branch.name)
         enriched_diffs = await self.diff_repository.get_roots_metadata(
             diff_branch_names=[self.source_branch.name],
@@ -77,6 +78,7 @@ async def merge_graph(self, at: Timestamp) -> None:
         self.source_branch.branched_from = at.to_string()
         await self.source_branch.save(db=self.db)
         registry.branch[self.source_branch.name] = self.source_branch
+        return enriched_diff
 
     async def rollback(self, at: Timestamp) -> None:
         rollback_query = await DiffMergeRollbackQuery.init(

backend/infrahub/core/merge.py

@@ -19,13 +19,15 @@
     from infrahub.core.branch import Branch
     from infrahub.core.diff.coordinator import DiffCoordinator
     from infrahub.core.diff.merger.merger import DiffMerger
+    from infrahub.core.diff.model.path import EnrichedDiffRoot
     from infrahub.core.diff.repository.repository import DiffRepository
     from infrahub.core.models import SchemaUpdateConstraintInfo, SchemaUpdateMigrationInfo
     from infrahub.core.schema.manager import SchemaDiff
     from infrahub.core.schema.schema_branch import SchemaBranch
     from infrahub.database import InfrahubDatabase
     from infrahub.services import InfrahubServices
 
+
 log = get_logger()
 
 
@@ -172,7 +174,7 @@ async def calculate_validations(self, target_schema: SchemaBranch) -> list[Schem
     async def merge(
         self,
         at: Optional[Union[str, Timestamp]] = None,
-    ) -> None:
+    ) -> EnrichedDiffRoot:
         """Merge the current branch into main."""
         if self.source_branch.name == registry.default_branch:
             raise ValidationError(f"Unable to merge the branch '{self.source_branch.name}' into itself")
@@ -198,8 +200,9 @@ async def merge(
         # TODO need to find a way to properly communicate back to the user any issue that could come up during the merge
         # From the Graph or From the repositories
         self._merge_at = Timestamp(at)
-        await self.diff_merger.merge_graph(at=self._merge_at)
+        branch_diff = await self.diff_merger.merge_graph(at=self._merge_at)
         await self.merge_repositories()
+        return branch_diff
 
     async def rollback(self) -> None:
         await self.diff_merger.rollback(at=self._merge_at)

backend/tests/integration/proposed_change/test_proposed_change_conflict.py

@@ -28,6 +28,7 @@
 if TYPE_CHECKING:
     from infrahub_sdk import InfrahubClient
 
+    from infrahub.core.diff.model.path import EnrichedDiffRoot
     from infrahub.core.timestamp import Timestamp
     from infrahub.database import InfrahubDatabase
     from tests.adapters.message_bus import BusSimulator
@@ -37,7 +38,7 @@ class ErroringBranchMerger(BranchMerger):
     async def merge(
         self,
         at: str | Timestamp | None = None,
-    ) -> None:
+    ) -> EnrichedDiffRoot:
         raise ValueError("This will always fail")