Add support for converting object type (#6343)

Author: LucasG0

backend/infrahub/core/convert_object_type/__init__.py

@@ -0,0 +1,122 @@
+from typing import Any
+
+from pydantic import BaseModel
+
+from infrahub.core.attribute import BaseAttribute
+from infrahub.core.branch import Branch
+from infrahub.core.constants import RelationshipCardinality
+from infrahub.core.manager import NodeManager
+from infrahub.core.node import Node
+from infrahub.core.node.create import create_node
+from infrahub.core.query.relationship import GetAllPeersIds
+from infrahub.core.relationship import RelationshipManager
+from infrahub.core.schema import NodeSchema
+from infrahub.database import InfrahubDatabase
+
+
+class InputDataForDestField(BaseModel):  # Only one of these fields can be not None
+    attribute_value: Any | None = None
+    peer_id: str | None = None
+    peers_ids: list[str] | None = None
+
+    @property
+    def value(self) -> Any:
+        fields = [self.attribute_value, self.peer_id, self.peers_ids]
+        set_fields = [f for f in fields if f is not None]
+        if len(set_fields) != 1:
+            raise ValueError("Exactly one of attribute_value, peer_id, or peers_ids must be set")
+        return set_fields[0]
+
+
+class InputForDestField(BaseModel):  # Only one of these fields can be not None
+    source_field: str | None = None
+    data: InputDataForDestField | None = None
+
+    @property
+    def value(self) -> Any:
+        if self.source_field is not None and self.data is not None:
+            raise ValueError("Only one of source_field or data can be set")
+        if self.source_field is None and self.data is None:
+            raise ValueError("Either source_field or data must be set")
+        return self.source_field if self.source_field is not None else self.data
+
+
+async def get_out_rels_peers_ids(node: Node, db: InfrahubDatabase) -> list[str]:
+    all_peers: list[Node] = []
+    for name in node._relationships:
+        relm: RelationshipManager = getattr(node, name)
+        peers = await relm.get_peers(db=db)
+        all_peers.extend(peers.values())
+    return [peer.id for peer in all_peers]
+
+
+async def build_data_new_node(db: InfrahubDatabase, mapping: dict[str, InputForDestField], node: Node) -> dict:
+    """Value of a given field on the target kind to convert is either an input source attribute/relationship of the source node,
+    or a raw value."""
+
+    data = {}
+    for dest_field_name, input_for_dest_field in mapping.items():
+        value = input_for_dest_field.value
+        if isinstance(value, str):  # source_field
+            item = getattr(node, value)
+            if isinstance(item, BaseAttribute):
+                data[dest_field_name] = item.value
+            elif isinstance(item, RelationshipManager):
+                if item.schema.cardinality == RelationshipCardinality.ONE:
+                    peer = await item.get_peer(db=db)
+                    if peer is not None:
+                        data[dest_field_name] = {"id": peer.id}
+                    # else, relationship is optional, and if the target relationship is mandatory an error will be raised during creation
+                elif item.schema.cardinality == RelationshipCardinality.MANY:
+                    data[dest_field_name] = [{"id": peer.id} for _, peer in (await item.get_peers(db=db)).items()]
+                else:
+                    raise ValueError(f"Unknown cardinality {item.schema.cardinality=}")
+        else:  # user input data
+            data[dest_field_name] = value.value
+    return data
+
+
+async def get_unidirectional_rels_peers_ids(node: Node, branch: Branch, db: InfrahubDatabase) -> list[str]:
+    """
+    Returns peers ids of nodes connected to input `node` through an incoming unidirectional relationship.
+    """
+
+    out_rels_identifier = [rel.identifier for rel in node.get_schema().relationships]
+    query = await GetAllPeersIds.init(db=db, node_id=node.id, branch=branch, exclude_identifiers=out_rels_identifier)
+    await query.execute(db=db)
+    return query.get_peers_uuids()
+
+
+async def convert_object_type(
+    node: Node, target_schema: NodeSchema, mapping: dict[str, InputForDestField], branch: Branch, db: InfrahubDatabase
+) -> Node:
+    """Delete the node and return the new created one. If creation fails, the node is not deleted, and raise an error.
+    An extra check is performed on input node peers relationships to make sure they are still valid."""
+
+    node_schema = node.get_schema()
+    if not isinstance(node_schema, NodeSchema):
+        raise ValueError(f"Only a node with a NodeSchema can be converted, got {type(node_schema)}")
+
+    async with db.start_transaction() as dbt:  # noqa: PLR1702
+        deleted_node_out_rels_peer_ids = await get_out_rels_peers_ids(node=node, db=dbt)
+        deleted_node_unidir_rels_peer_ids = await get_unidirectional_rels_peers_ids(node=node, db=dbt, branch=branch)
+
+        deleted_nodes = await NodeManager.delete(db=dbt, branch=branch, nodes=[node], cascade_delete=False)
+        if len(deleted_nodes) != 1:
+            raise ValueError(f"Deleted {len(deleted_nodes)} nodes instead of 1")
+
+        data_new_node = await build_data_new_node(dbt, mapping, node)
+        node_created = await create_node(
+            data=data_new_node,
+            db=dbt,
+            branch=branch,
+            schema=target_schema,
+        )
+
+        # Make sure relationships with constraints are not broken by retrieving them
+        peers_ids = deleted_node_out_rels_peer_ids + deleted_node_unidir_rels_peer_ids
+        peers = await NodeManager.get_many(ids=peers_ids, db=dbt, prefetch_relationships=True, branch=branch)
+        for peer in peers.values():
+            peer.validate_relationships()
+
+        return node_created

backend/infrahub/core/convert_object_type/conversion.py

@@ -0,0 +1,56 @@
+from pydantic import BaseModel
+
+from infrahub.core.constants import RelationshipCardinality
+from infrahub.core.schema import NodeSchema
+
+
+class SchemaMappingValue(BaseModel):
+    is_mandatory: bool
+    source_field_name: str | None = None  # None means there is no corresponding source field name
+    relationship_cardinality: RelationshipCardinality | None = None
+
+
+SchemaMapping = dict[str, SchemaMappingValue]
+
+
+def get_schema_mapping(source_schema: NodeSchema, target_schema: NodeSchema) -> SchemaMapping:
+    """
+    Return fields mapping meant to be used for converting a node from `source_kind` to `target_kind`.
+    For any field of the target kind, field of the source kind will be matched if:
+    - It's an attribute with identical name and type.
+    - It's a relationship with identical peer kind and cardinality.
+    If there is no match, the mapping will only indicate whether the field is mandatory or not.
+    """
+
+    target_field_to_source_field = {}
+
+    # Create lookup dictionaries for source attributes and relationships
+    source_attrs = {attr.name: attr for attr in source_schema.attributes}
+    source_rels = {rel.name: rel for rel in source_schema.relationships}
+
+    # Process attributes
+    for target_attr in target_schema.attributes:
+        source_attr = source_attrs.get(target_attr.name)
+        if source_attr and source_attr.kind == target_attr.kind:
+            target_field_to_source_field[target_attr.name] = SchemaMappingValue(
+                source_field_name=source_attr.name, is_mandatory=not target_attr.optional
+            )
+        else:
+            target_field_to_source_field[target_attr.name] = SchemaMappingValue(is_mandatory=not target_attr.optional)
+
+    # Process relationships
+    for target_rel in target_schema.relationships:
+        source_rel = source_rels.get(target_rel.name)
+        if source_rel and source_rel.peer == target_rel.peer and source_rel.cardinality == target_rel.cardinality:
+            target_field_to_source_field[target_rel.name] = SchemaMappingValue(
+                source_field_name=source_rel.name,
+                is_mandatory=not target_rel.optional,
+                relationship_cardinality=target_rel.cardinality,
+            )
+        else:
+            target_field_to_source_field[target_rel.name] = SchemaMappingValue(
+                is_mandatory=not target_rel.optional,
+                relationship_cardinality=target_rel.cardinality,
+            )
+
+    return target_field_to_source_field

backend/infrahub/core/convert_object_type/schema_mapping.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from copy import copy
 from functools import reduce
 from typing import TYPE_CHECKING, Any, Iterable, Literal, TypeVar, overload
 
@@ -1339,22 +1340,24 @@ async def delete(
         nodes: list[Node],
         branch: Branch | str | None = None,
         at: Timestamp | str | None = None,
+        cascade_delete: bool = True,
     ) -> list[Node]:
         """Returns list of deleted nodes because of cascading deletes"""
         branch = await registry.get_branch(branch=branch, db=db)
-        node_delete_validator = NodeDeleteValidator(db=db, branch=branch)
-        ids_to_delete = await node_delete_validator.get_ids_to_delete(nodes=nodes, at=at)
-        node_ids = {node.get_id() for node in nodes}
-        missing_ids_to_delete = ids_to_delete - node_ids
-        if missing_ids_to_delete:
-            node_map = await cls.get_many(db=db, ids=list(missing_ids_to_delete), branch=branch, at=at)
-            nodes += list(node_map.values())
-        deleted_nodes = []
-        for node in nodes:
+        nodes_to_delete = copy(nodes)
+        if cascade_delete:
+            node_delete_validator = NodeDeleteValidator(db=db, branch=branch)
+            ids_to_delete = await node_delete_validator.get_ids_to_delete(nodes=nodes, at=at)
+            node_ids = {node.get_id() for node in nodes}
+            missing_ids_to_delete = ids_to_delete - node_ids
+            if missing_ids_to_delete:
+                node_map = await cls.get_many(db=db, ids=list(missing_ids_to_delete), branch=branch, at=at)
+                nodes_to_delete += list(node_map.values())
+
+        for node in nodes_to_delete:
             await node.delete(db=db, at=at)
-            deleted_nodes.append(node)
 
-        return deleted_nodes
+        return nodes_to_delete
 
 
 def _get_cardinality_one_identifiers_by_kind(

backend/infrahub/core/manager.py

@@ -22,7 +22,14 @@
 from infrahub.core.constants.schema import SchemaElementPathType
 from infrahub.core.protocols import CoreNumberPool, CoreObjectTemplate
 from infrahub.core.query.node import NodeCheckIDQuery, NodeCreateAllQuery, NodeDeleteQuery, NodeGetListQuery
-from infrahub.core.schema import AttributeSchema, NodeSchema, ProfileSchema, RelationshipSchema, TemplateSchema
+from infrahub.core.schema import (
+    AttributeSchema,
+    NodeSchema,
+    NonGenericSchemaTypes,
+    ProfileSchema,
+    RelationshipSchema,
+    TemplateSchema,
+)
 from infrahub.core.timestamp import Timestamp
 from infrahub.exceptions import InitializationError, NodeNotFoundError, PoolExhaustedError, ValidationError
 from infrahub.types import ATTRIBUTE_TYPES
@@ -66,7 +73,7 @@ def __init_subclass_with_meta__(cls, _meta=None, default_filter=None, **options)
         _meta.default_filter = default_filter
         super().__init_subclass_with_meta__(_meta=_meta, **options)
 
-    def get_schema(self) -> NodeSchema | ProfileSchema | TemplateSchema:
+    def get_schema(self) -> NonGenericSchemaTypes:
         return self._schema
 
     def get_kind(self) -> str:
@@ -872,7 +879,7 @@ def _get_parent_relationship_name(self) -> str | None:
             if relationship.kind == RelationshipKind.PARENT:
                 return relationship.name
 
-    async def get_object_template(self, db: InfrahubDatabase) -> Node | None:
+    async def get_object_template(self, db: InfrahubDatabase) -> CoreObjectTemplate | None:
         object_template: RelationshipManager = getattr(self, OBJECT_TEMPLATE_RELATIONSHIP_NAME, None)
         return await object_template.get_peer(db=db) if object_template is not None else None
 
@@ -888,3 +895,8 @@ def get_relationships(
             for relationship in self.get_schema().relationships
             if relationship.name not in exclude and relationship.kind == kind
         ]
+
+    def validate_relationships(self) -> None:
+        for name in self._relationships:
+            relm: RelationshipManager = getattr(self, name)
+            relm.validate()