Migrate GitRepositoryAdd to prefect

backend/infrahub/git/models.py

@@ -33,3 +33,15 @@ class RequestArtifactGenerate(BaseModel):
     query: str = Field(..., description="The name of the query to use when collecting data")
     timeout: int = Field(..., description="Timeout for requests used to generate this artifact")
     variables: dict = Field(..., description="Input variables when generating the artifact")
+
+
+class GitRepositoryAdd(BaseModel):
+    """Clone and sync an external repository after creation."""
+
+    location: str = Field(..., description="The external URL of the repository")
+    repository_id: str = Field(..., description="The unique ID of the Repository")
+    repository_name: str = Field(..., description="The name of the repository")
+    created_by: Optional[str] = Field(default=None, description="The user ID of the user that created the repository")
+    default_branch_name: Optional[str] = Field(None, description="Default branch for this repository")
+    infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")
+    internal_status: str = Field(..., description="Administrative status of the repository")

backend/infrahub/git/tasks.py

@@ -6,18 +6,44 @@
 from infrahub.core.protocols import CoreRepository
 from infrahub.core.registry import registry
 from infrahub.exceptions import RepositoryError
+from infrahub.git.repository import InfrahubRepository, get_initialized_repo
 from infrahub.services import services
 
 from ..log import get_logger
 from ..tasks.artifact import define_artifact
 from ..workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE, REQUEST_ARTIFACT_GENERATE
 from ..workflows.utils import add_branch_tag
-from .models import RequestArtifactDefinitionGenerate, RequestArtifactGenerate
-from .repository import InfrahubRepository, get_initialized_repo
+from .models import GitRepositoryAdd, RequestArtifactDefinitionGenerate, RequestArtifactGenerate
 
 log = get_logger()
 
 
+@flow(name="git-repository-add-read-write")
+async def add_git_repository(model: GitRepositoryAdd) -> None:
+    service = services.service
+    async with service.git_report(
+        related_node=model.repository_id,
+        title=f"Initial import of the repository in branch: {model.infrahub_branch_name}",
+        created_by=model.created_by,
+    ) as git_report:
+        async with lock.registry.get(name=model.repository_name, namespace="repository"):
+            repo = await InfrahubRepository.new(
+                id=model.repository_id,
+                name=model.repository_name,
+                location=model.location,
+                client=service.client,
+                task_report=git_report,
+                infrahub_branch_name=model.infrahub_branch_name,
+                internal_status=model.internal_status,
+                default_branch_name=model.default_branch_name,
+            )
+            await repo.import_objects_from_files(
+                infrahub_branch_name=model.infrahub_branch_name, git_branch_name=model.default_branch_name
+            )
+            if model.internal_status == RepositoryInternalStatus.ACTIVE.value:
+                await repo.sync()
+
+
 @flow(name="git-repositories-branch-create")
 async def create_branch(branch: str, branch_id: str) -> None:
     """Request to the creation of git branches in available repositories."""

backend/infrahub/graphql/mutations/repository.py

@@ -15,6 +15,8 @@
 from infrahub.message_bus import messages
 from infrahub.message_bus.messages.git_repository_connectivity import GitRepositoryConnectivityResponse
 
+from ...git.models import GitRepositoryAdd
+from ...workflows.catalogue import GIT_REPOSITORY_ADD
 from .main import InfrahubMutationMixin, InfrahubMutationOptions
 
 if TYPE_CHECKING:
@@ -99,7 +101,7 @@ async def mutate_create(
             )
         else:
             obj = cast(CoreRepository, obj)
-            message = messages.GitRepositoryAdd(
+            git_repo_add_model = GitRepositoryAdd(
                 repository_id=obj.id,
                 repository_name=obj.name.value,
                 location=obj.location.value,
@@ -109,8 +111,10 @@ async def mutate_create(
                 created_by=authenticated_user,
             )
 
-        if context.service:
-            await context.service.send(message=message)
+            if context.service:
+                context.service.workflow.submit_workflow(
+                    workflow=GIT_REPOSITORY_ADD, parameters={"model": git_repo_add_model}
+                )
 
         # TODO Validate that the creation of the repository went as expected
 

backend/infrahub/message_bus/messages/__init__.py

@@ -15,7 +15,6 @@
 from .finalize_validator_execution import FinalizeValidatorExecution
 from .git_diff_namesonly import GitDiffNamesOnly, GitDiffNamesOnlyResponse
 from .git_file_get import GitFileGet, GitFileGetResponse
-from .git_repository_add import GitRepositoryAdd
 from .git_repository_connectivity import GitRepositoryConnectivity
 from .git_repository_importobjects import GitRepositoryImportObjects
 from .git_repository_merge import GitRepositoryMerge
@@ -61,7 +60,6 @@
     "finalize.validator.execution": FinalizeValidatorExecution,
     "git.diff.names_only": GitDiffNamesOnly,
     "git.file.get": GitFileGet,
-    "git.repository.add": GitRepositoryAdd,
     "git.repository.connectivity": GitRepositoryConnectivity,
     "git.repository.merge": GitRepositoryMerge,
     "git.repository.add_read_only": GitRepositoryAddReadOnly,

backend/infrahub/message_bus/operations/__init__.py

@@ -35,7 +35,6 @@
     "finalize.validator.execution": finalize.validator.execution,
     "git.diff.names_only": git.diff.names_only,
     "git.file.get": git.file.get,
-    "git.repository.add": git.repository.add,
     "git.repository.add_read_only": git.repository.add_read_only,
     "git.repository.connectivity": git.repository.connectivity,
     "git.repository.import_objects": git.repository.import_objects,

backend/infrahub/message_bus/operations/git/repository.py

@@ -15,37 +15,6 @@
 log = get_logger()
 
 
-@flow(name="git-repository-add-read-write")
-async def add(message: messages.GitRepositoryAdd, service: InfrahubServices) -> None:
-    log.info(
-        "Cloning and importing repository",
-        repository=message.repository_name,
-        location=message.location,
-        internal_status=message.internal_status,
-    )
-    async with service.git_report(
-        related_node=message.repository_id,
-        title=f"Initial import of the repository in branch: {message.infrahub_branch_name}",
-        created_by=message.created_by,
-    ) as git_report:
-        async with lock.registry.get(name=message.repository_name, namespace="repository"):
-            repo = await InfrahubRepository.new(
-                id=message.repository_id,
-                name=message.repository_name,
-                location=message.location,
-                client=service.client,
-                task_report=git_report,
-                infrahub_branch_name=message.infrahub_branch_name,
-                internal_status=message.internal_status,
-                default_branch_name=message.default_branch_name,
-            )
-            await repo.import_objects_from_files(
-                infrahub_branch_name=message.infrahub_branch_name, git_branch_name=message.default_branch_name
-            )
-            if message.internal_status == RepositoryInternalStatus.ACTIVE.value:
-                await repo.sync()
-
-
 @flow(name="git-repository-add-read-only")
 async def add_read_only(message: messages.GitRepositoryAddReadOnly, service: InfrahubServices) -> None:
     log.info(

backend/infrahub/workflows/catalogue.py

@@ -122,6 +122,16 @@
     branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
+
+GIT_REPOSITORY_ADD = WorkflowDefinition(
+    name="git-repository-add-read-write",
+    type=WorkflowType.INTERNAL,
+    module="infrahub.git.tasks",
+    function="add_git_repository",
+    branch_support=BranchSupportType.AWARE,
+    tags=[WorkflowTag.DATABASE_CHANGE],
+)
+
 BRANCH_REBASE = WorkflowDefinition(
     name="branch-rebase",
     type=WorkflowType.INTERNAL,
@@ -160,4 +170,5 @@
     REQUEST_GENERATOR_RUN,
     REQUEST_DIFF_UPDATE,
     REQUEST_DIFF_REFRESH,
+    GIT_REPOSITORY_ADD,
 ]

backend/tests/unit/git/test_git_rpc.py

@@ -10,11 +10,13 @@
 from infrahub.core.constants import InfrahubKind, RepositoryInternalStatus
 from infrahub.exceptions import RepositoryError
 from infrahub.git import InfrahubRepository
+from infrahub.git.models import GitRepositoryAdd
 from infrahub.git.repository import InfrahubReadOnlyRepository
+from infrahub.git.tasks import add_git_repository
 from infrahub.lock import InfrahubLockRegistry
 from infrahub.message_bus import Meta, messages
 from infrahub.message_bus.operations import git
-from infrahub.services import InfrahubServices
+from infrahub.services import InfrahubServices, services
 from tests.helpers.test_client import dummy_async_request
 
 # pylint: disable=redefined-outer-name
@@ -49,28 +51,22 @@ def setup_method(self):
         self.default_branch_name = "default-branch"
         self.client = AsyncMock(spec=InfrahubClient)
         self.git_report = AsyncContextManagerMock()
+        self.original_services = services.service
+        services.service = InfrahubServices(client=self.client)
+        services.service.git_report = self.git_report
 
-        self.services = InfrahubServices(client=self.client)
-        self.services.git_report = self.git_report
-        lock_patcher = patch("infrahub.message_bus.operations.git.repository.lock")
-        self.mock_infra_lock = lock_patcher.start()
-        self.mock_infra_lock.registry = AsyncMock(spec=InfrahubLockRegistry)
-        repo_class_patcher = patch(
-            "infrahub.message_bus.operations.git.repository.InfrahubRepository", spec=InfrahubRepository
-        )
-        self.mock_repo_class = repo_class_patcher.start()
         self.mock_repo = AsyncMock(spec=InfrahubRepository)
         self.mock_repo.default_branch = self.default_branch_name
         self.mock_repo.infrahub_branch_name = self.default_branch_name
         self.mock_repo.internal_status = "active"
-        self.mock_repo_class.new.return_value = self.mock_repo
 
     def teardown_method(self):
         patch.stopall()
+        services.service = self.original_services
 
     async def test_git_rpc_create_successful(self, git_upstream_repo_01: dict[str, str]):
         repo_id = str(UUIDT())
-        message = messages.GitRepositoryAdd(
+        model = GitRepositoryAdd(
             repository_id=repo_id,
             repository_name=git_upstream_repo_01["name"],
             location=git_upstream_repo_01["path"],
@@ -79,25 +75,33 @@ async def test_git_rpc_create_successful(self, git_upstream_repo_01: dict[str, s
             internal_status="active",
         )
 
-        await git.repository.add(message=message, service=self.services)
-
-        self.mock_infra_lock.registry.get.assert_called_once_with(
-            name=git_upstream_repo_01["name"], namespace="repository"
-        )
-        self.mock_repo_class.new.assert_awaited_once_with(
-            id=repo_id,
-            name=git_upstream_repo_01["name"],
-            location=git_upstream_repo_01["path"],
-            client=self.client,
-            task_report=self.git_report,
-            infrahub_branch_name=self.default_branch_name,
-            internal_status="active",
-            default_branch_name=self.default_branch_name,
-        )
-        self.mock_repo.import_objects_from_files.assert_awaited_once_with(
-            infrahub_branch_name=self.default_branch_name, git_branch_name=self.default_branch_name
-        )
-        self.mock_repo.sync.assert_awaited_once_with()
+        with (
+            patch("infrahub.git.tasks.lock") as mock_infra_lock,
+            patch("infrahub.git.tasks.InfrahubRepository", spec=InfrahubRepository) as mock_repo_class,
+        ):
+            mock_infra_lock.registry = AsyncMock(spec=InfrahubLockRegistry)
+            mock_repo_class.new.return_value = self.mock_repo
+
+            await add_git_repository(model=model)
+
+            mock_infra_lock.registry.get.assert_called_once_with(
+                name=git_upstream_repo_01["name"], namespace="repository"
+            )
+
+            mock_repo_class.new.assert_awaited_once_with(
+                id=repo_id,
+                name=git_upstream_repo_01["name"],
+                location=git_upstream_repo_01["path"],
+                client=self.client,
+                task_report=self.git_report,
+                infrahub_branch_name=self.default_branch_name,
+                internal_status="active",
+                default_branch_name=self.default_branch_name,
+            )
+            self.mock_repo.import_objects_from_files.assert_awaited_once_with(
+                infrahub_branch_name=self.default_branch_name, git_branch_name=self.default_branch_name
+            )
+            self.mock_repo.sync.assert_awaited_once_with()
 
 
 async def test_git_rpc_merge(

docs/docs/reference/message-bus-events.mdx

@@ -356,26 +356,6 @@ For more detailed explanations on how to use these events within Infrahub, see t
 ### Git Repository
 <!-- vale on -->
 
-<!-- vale off -->
-#### Event git.repository.add
-<!-- vale on -->
-
-**Description**: Clone and sync an external repository after creation.
-
-**Priority**: 3
-
-<!-- vale off -->
-| Key | Description | Type | Default Value |
-|-----|-------------|------|---------------|
-| **meta** | Meta properties for the message | N/A | None |
-| **location** | The external URL of the repository | string | None |
-| **repository_id** | The unique ID of the Repository | string | None |
-| **repository_name** | The name of the repository | string | None |
-| **created_by** | The user ID of the user that created the repository | N/A | None |
-| **default_branch_name** | Default branch for this repository | N/A | None |
-| **infrahub_branch_name** | Infrahub branch on which to sync the remote repository | string | None |
-| **internal_status** | Administrative status of the repository | string | None |
-<!-- vale on -->
 <!-- vale off -->
 #### Event git.repository.connectivity
 <!-- vale on -->
@@ -1278,27 +1258,6 @@ For more detailed explanations on how to use these events within Infrahub, see t
 ### Git Repository
 <!-- vale on -->
 
-<!-- vale off -->
-#### Event git.repository.add
-<!-- vale on -->
-
-**Description**: Clone and sync an external repository after creation.
-
-**Priority**: 3
-
-
-<!-- vale off -->
-| Key | Description | Type | Default Value |
-|-----|-------------|------|---------------|
-| **meta** | Meta properties for the message | N/A | None |
-| **location** | The external URL of the repository | string | None |
-| **repository_id** | The unique ID of the Repository | string | None |
-| **repository_name** | The name of the repository | string | None |
-| **created_by** | The user ID of the user that created the repository | N/A | None |
-| **default_branch_name** | Default branch for this repository | N/A | None |
-| **infrahub_branch_name** | Infrahub branch on which to sync the remote repository | string | None |
-| **internal_status** | Administrative status of the repository | string | None |
-<!-- vale on -->
 <!-- vale off -->
 #### Event git.repository.connectivity
 <!-- vale on -->