Support converting repository type

Author: LucasG0

backend/infrahub/core/convert_object_type/conversion.py

@@ -2,16 +2,21 @@
 
 from pydantic import BaseModel
 
+from infrahub.auth import AccountSession
+from infrahub.context import InfrahubContext
 from infrahub.core.attribute import BaseAttribute
 from infrahub.core.branch import Branch
 from infrahub.core.constants import RelationshipCardinality
+from infrahub.core.constants.infrahubkind import READONLYREPOSITORY, REPOSITORY
 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.repositories.create_repository import post_create_repository
 from infrahub.core.schema import NodeSchema
 from infrahub.database import InfrahubDatabase
+from infrahub.services import InfrahubServices
 
 
 class InputDataForDestField(BaseModel):  # Only one of these fields can be not None
@@ -88,7 +93,14 @@ async def get_unidirectional_rels_peers_ids(node: Node, branch: Branch, db: Infr
 
 
 async def convert_object_type(
-    node: Node, target_schema: NodeSchema, mapping: dict[str, InputForDestField], branch: Branch, db: InfrahubDatabase
+    node: Node,
+    target_schema: NodeSchema,
+    mapping: dict[str, InputForDestField],
+    branch: Branch,
+    db: InfrahubDatabase,
+    account_session: AccountSession,
+    services: InfrahubServices,
+    context: InfrahubContext,
 ) -> 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."""
@@ -106,6 +118,7 @@ async def convert_object_type(
             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,
@@ -119,4 +132,17 @@ async def convert_object_type(
         for peer in peers.values():
             peer.validate_relationships()
 
-        return node_created
+    # We can't apply post creation steps within above transaction as a sdk call tries to fetch the node
+    # created within the transaction from a different process, therefore that would not run within this transaction
+    # so the node ends up being not found
+    if target_schema.kind in [REPOSITORY, READONLYREPOSITORY]:
+        await post_create_repository(
+            obj=node_created,  # type: ignore
+            db=db,
+            branch=branch,
+            account_session=account_session,
+            services=services,
+            context=context,
+        )
+
+    return node_created

backend/infrahub/core/node/create.py

@@ -2,13 +2,17 @@
 
 from typing import TYPE_CHECKING, Any, Mapping
 
+from infrahub import lock
 from infrahub.core import registry
 from infrahub.core.constants import RelationshipCardinality, RelationshipKind
 from infrahub.core.constraint.node.runner import NodeConstraintRunner
 from infrahub.core.manager import NodeManager
 from infrahub.core.node import Node
+from infrahub.core.node.lock_utils import get_kind_lock_names_on_object_mutation
 from infrahub.core.protocols import CoreObjectTemplate
+from infrahub.core.schema import GenericSchema
 from infrahub.dependencies.registry import get_component_registry
+from infrahub.lock import InfrahubMultiLock
 
 if TYPE_CHECKING:
     from infrahub.core.branch import Branch
@@ -170,10 +174,13 @@ async def create_node(
     data: dict,
     db: InfrahubDatabase,
     branch: Branch,
-    schema: NonGenericSchemaTypes,
+    schema: MainSchemaTypes,
 ) -> Node:
     """Create a node in the database if constraint checks succeed."""
 
+    if isinstance(schema, GenericSchema):
+        raise ValueError(f"Node of generic schema `{schema.name=}` can not be instantiated.")
+
     component_registry = get_component_registry()
     node_constraint_runner = await component_registry.get_component(
         NodeConstraintRunner, db=db.start_session() if not db.is_transaction else db, branch=branch
@@ -183,27 +190,54 @@ async def create_node(
         node_class = registry.node[schema.kind]
 
     fields_to_validate = list(data)
+    schema_branch = db.schema.get_schema_branch(name=branch.name)
+    lock_names = get_kind_lock_names_on_object_mutation(kind=schema.kind, branch=branch, schema_branch=schema_branch)
+
     if db.is_transaction:
-        obj = await _do_create_node(
-            node_class=node_class,
-            node_constraint_runner=node_constraint_runner,
-            db=db,
-            schema=schema,
-            branch=branch,
-            fields_to_validate=fields_to_validate,
-            data=data,
-        )
-    else:
-        async with db.start_transaction() as dbt:
+        if lock_names:
+            async with InfrahubMultiLock(lock_registry=lock.registry, locks=lock_names):
+                obj = await _do_create_node(
+                    node_class=node_class,
+                    node_constraint_runner=node_constraint_runner,
+                    db=db,
+                    schema=schema,
+                    branch=branch,
+                    fields_to_validate=fields_to_validate,
+                    data=data,
+                )
+        else:
             obj = await _do_create_node(
                 node_class=node_class,
                 node_constraint_runner=node_constraint_runner,
-                db=dbt,
+                db=db,
                 schema=schema,
                 branch=branch,
                 fields_to_validate=fields_to_validate,
                 data=data,
             )
+    else:
+        async with db.start_transaction() as dbt:
+            if lock_names:
+                async with InfrahubMultiLock(lock_registry=lock.registry, locks=lock_names):
+                    obj = await _do_create_node(
+                        node_class=node_class,
+                        node_constraint_runner=node_constraint_runner,
+                        db=dbt,
+                        schema=schema,
+                        branch=branch,
+                        fields_to_validate=fields_to_validate,
+                        data=data,
+                    )
+            else:
+                obj = await _do_create_node(
+                    node_class=node_class,
+                    node_constraint_runner=node_constraint_runner,
+                    db=dbt,
+                    schema=schema,
+                    branch=branch,
+                    fields_to_validate=fields_to_validate,
+                    data=data,
+                )
 
     if await get_profile_ids(db=db, obj=obj):
         obj = await refresh_for_profile_update(db=db, branch=branch, schema=schema, obj=obj)

backend/infrahub/core/node/lock_utils.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from infrahub.core.constants.infrahubkind import GENERICGROUP
+from infrahub.core.schema import GenericSchema
+from infrahub.lock import build_object_lock_name
+
+if TYPE_CHECKING:
+    from infrahub.core.branch import Branch
+    from infrahub.core.schema.schema_branch import SchemaBranch
+
+
+KINDS_CONCURRENT_MUTATIONS_NOT_ALLOWED = [GENERICGROUP]
+
+
+def get_kind_lock_names_on_object_mutation(kind: str, branch: Branch, schema_branch: SchemaBranch) -> list[str]:
+    """
+    Return objects kind for which we want to avoid concurrent mutation (create/update). Except for some specific kinds,
+    concurrent mutations are only allowed on non-main branch as objects validations will be performed at least when merging in main branch.
+    """
+
+    if not branch.is_default and not _should_kind_be_locked_on_any_branch(kind, schema_branch):
+        return []
+
+    lock_kinds = _get_kinds_to_lock_on_object_mutation(kind, schema_branch)
+    lock_names = [build_object_lock_name(kind) for kind in lock_kinds]
+    return lock_names
+
+
+def _should_kind_be_locked_on_any_branch(kind: str, schema_branch: SchemaBranch) -> bool:
+    """
+    Check whether kind or any kind generic is in KINDS_TO_LOCK_ON_ANY_BRANCH.
+    """
+
+    if kind in KINDS_CONCURRENT_MUTATIONS_NOT_ALLOWED:
+        return True
+
+    node_schema = schema_branch.get(name=kind)
+    if isinstance(node_schema, GenericSchema):
+        return False
+
+    for generic_kind in node_schema.inherit_from:
+        if generic_kind in KINDS_CONCURRENT_MUTATIONS_NOT_ALLOWED:
+            return True
+    return False
+
+
+def _get_kinds_to_lock_on_object_mutation(kind: str, schema_branch: SchemaBranch) -> list[str]:
+    """
+    Return kinds for which we want to lock during creating / updating an object of a given schema node.
+    Lock should be performed on schema kind and its generics having a uniqueness_constraint defined.
+    If a generic uniqueness constraint is the same as the node schema one,
+    it means node schema overrided this constraint, in which case we only need to lock on the generic.
+    """
+
+    node_schema = schema_branch.get(name=kind)
+
+    schema_uc = None
+    kinds = []
+    if node_schema.uniqueness_constraints:
+        kinds.append(node_schema.kind)
+        schema_uc = node_schema.uniqueness_constraints
+
+    if isinstance(node_schema, GenericSchema):
+        return kinds
+
+    generics_kinds = node_schema.inherit_from
+
+    node_schema_kind_removed = False
+    for generic_kind in generics_kinds:
+        generic_uc = schema_branch.get(name=generic_kind).uniqueness_constraints
+        if generic_uc:
+            kinds.append(generic_kind)
+            if not node_schema_kind_removed and generic_uc == schema_uc:
+                # Check whether we should remove original schema kind as it simply overrides uniqueness_constraint
+                # of a generic
+                kinds.pop(0)
+                node_schema_kind_removed = True
+    return kinds

backend/infrahub/core/repositories/__init__.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from infrahub.core.constants import RepositoryInternalStatus
+from infrahub.core.constants.infrahubkind import READONLYREPOSITORY
+from infrahub.core.protocols import CoreGenericRepository, CoreReadOnlyRepository, CoreRepository
+from infrahub.exceptions import ValidationError
+from infrahub.git.models import GitRepositoryAdd, GitRepositoryAddReadOnly
+from infrahub.log import get_logger
+from infrahub.message_bus import messages
+from infrahub.message_bus.messages.git_repository_connectivity import GitRepositoryConnectivityResponse
+from infrahub.workflows.catalogue import GIT_REPOSITORY_ADD, GIT_REPOSITORY_ADD_READ_ONLY
+
+if TYPE_CHECKING:
+    from infrahub.auth import AccountSession
+    from infrahub.context import InfrahubContext
+    from infrahub.core.branch import Branch
+    from infrahub.database import InfrahubDatabase
+    from infrahub.services import InfrahubServices
+
+log = get_logger()
+
+
+async def post_create_repository(
+    obj: CoreGenericRepository,
+    branch: Branch,
+    db: InfrahubDatabase,
+    account_session: AccountSession,
+    services: InfrahubServices,
+    context: InfrahubContext,
+) -> None:
+    # First check the connectivity to the remote repository
+    # If the connectivity is not good, we remove the repository to allow the user to add a new one
+
+    message = messages.GitRepositoryConnectivity(
+        repository_name=obj.name.value,
+        repository_location=obj.location.value,
+    )
+    response = await services.message_bus.rpc(message=message, response_class=GitRepositoryConnectivityResponse)
+
+    if response.data.success is False:
+        await obj.delete(db=db)
+        raise ValidationError(response.data.message)
+    # If we are in the default branch, we set the sync status to Active
+    # If we are in another branch, we set the sync status to Staging
+    if branch.is_default:
+        obj.internal_status.value = RepositoryInternalStatus.ACTIVE.value
+    else:
+        obj.internal_status.value = RepositoryInternalStatus.STAGING.value
+    await obj.save(db=db)
+
+    # Create the new repository in the filesystem.
+    log.info("create_repository", name=obj.name.value)
+    authenticated_user = None
+    if account_session and account_session.authenticated:
+        authenticated_user = account_session.account_id
+    if obj.get_kind() == READONLYREPOSITORY:
+        obj = cast(CoreReadOnlyRepository, obj)
+        model = GitRepositoryAddReadOnly(
+            repository_id=obj.id,
+            repository_name=obj.name.value,
+            location=obj.location.value,
+            ref=obj.ref.value,
+            infrahub_branch_name=branch.name,
+            infrahub_branch_id=str(branch.get_uuid()),
+            internal_status=obj.internal_status.value,
+            created_by=authenticated_user,
+        )
+        await services.workflow.submit_workflow(
+            workflow=GIT_REPOSITORY_ADD_READ_ONLY,
+            context=context,
+            parameters={"model": model},
+        )
+
+    else:
+        obj = cast(CoreRepository, obj)
+        git_repo_add_model = GitRepositoryAdd(
+            repository_id=obj.id,
+            repository_name=obj.name.value,
+            location=obj.location.value,
+            default_branch_name=obj.default_branch.value,
+            infrahub_branch_name=branch.name,
+            infrahub_branch_id=str(branch.get_uuid()),
+            internal_status=obj.internal_status.value,
+            created_by=authenticated_user,
+        )
+
+        await services.workflow.submit_workflow(
+            workflow=GIT_REPOSITORY_ADD,
+            context=context,
+            parameters={"model": git_repo_add_model},
+        )
+    # TODO Validate that the creation of the repository went as expected

backend/infrahub/core/repositories/create_repository.py

@@ -54,6 +54,9 @@ async def mutate(
             mapping=fields_mapping,
             branch=graphql_context.branch,
             db=graphql_context.db,
+            account_session=graphql_context.active_account_session,
+            services=graphql_context.active_service,
+            context=graphql_context.get_context(),
         )
 
         dict_node = await new_node.to_graphql(db=graphql_context.db, fields={})