Merge pull request #4885 from opsmill/lgu-migrate-req-prop-changed-data-integrity

Migrate RequestProposedChangeDataIntegrity to prefect

Author: LucasG0

backend/infrahub/message_bus/messages/__init__.py

@@ -17,7 +17,6 @@
 from .git_file_get import GitFileGet, GitFileGetResponse
 from .git_repository_connectivity import GitRepositoryConnectivity
 from .git_repository_importobjects import GitRepositoryImportObjects
-from .proposed_change.request_proposedchange_dataintegrity import RequestProposedChangeDataIntegrity
 from .proposed_change.request_proposedchange_refreshartifacts import RequestProposedChangeRefreshArtifacts
 from .proposed_change.request_proposedchange_repositorychecks import RequestProposedChangeRepositoryChecks
 from .proposed_change.request_proposedchange_rungenerators import RequestProposedChangeRunGenerators
@@ -61,7 +60,6 @@
     "refresh.webhook.configuration": RefreshWebhookConfiguration,
     "request.artifact_definition.check": RequestArtifactDefinitionCheck,
     "request.generator_definition.check": RequestGeneratorDefinitionCheck,
-    "request.proposed_change.data_integrity": RequestProposedChangeDataIntegrity,
     "request.proposed_change.pipeline": RequestProposedChangePipeline,
     "request.proposed_change.refresh_artifacts": RequestProposedChangeRefreshArtifacts,
     "request.proposed_change.repository_checks": RequestProposedChangeRepositoryChecks,

backend/infrahub/message_bus/operations/__init__.py

@@ -41,7 +41,6 @@
     "refresh.webhook.configuration": refresh.webhook.configuration,
     "request.generator_definition.check": requests.generator_definition.check,
     "request.artifact_definition.check": requests.artifact_definition.check,
-    "request.proposed_change.data_integrity": requests.proposed_change.data_integrity,
     "request.proposed_change.pipeline": requests.proposed_change.pipeline,
     "request.proposed_change.refresh_artifacts": requests.proposed_change.refresh_artifacts,
     "request.proposed_change.repository_checks": requests.proposed_change.repository_checks,

backend/infrahub/message_bus/operations/requests/proposed_change.py

@@ -32,8 +32,10 @@
     ProposedChangeRepository,
     ProposedChangeSubscriber,
 )
+from infrahub.proposed_change.models import RequestProposedChangeDataIntegrity
 from infrahub.pytest_plugin import InfrahubBackendPlugin
 from infrahub.services import InfrahubServices  # noqa: TCH001
+from infrahub.workflows.catalogue import REQUEST_PROPOSED_CHANGE_DATA_INTEGRITY
 
 if TYPE_CHECKING:
     from infrahub_sdk.node import InfrahubNode
@@ -71,23 +73,6 @@ def log_line(self) -> str:
         return "Doesn't require changes due to no relevant modified kinds or file changes in Git"
 
 
-@flow(name="proposed-changed-data-integrity")
-async def data_integrity(message: messages.RequestProposedChangeDataIntegrity, service: InfrahubServices) -> None:
-    """Triggers a data integrity validation check on the provided proposed change to start."""
-    async with service.task_report(
-        related_node=message.proposed_change,
-        title="Data Integrity",
-    ):
-        log.info(f"Got a request to process data integrity defined in proposed_change: {message.proposed_change}")
-
-        destination_branch = await registry.get_branch(db=service.database, branch=message.destination_branch)
-        source_branch = await registry.get_branch(db=service.database, branch=message.source_branch)
-        component_registry = get_component_registry()
-        async with service.database.start_transaction() as dbt:
-            diff_coordinator = await component_registry.get_component(DiffCoordinator, db=dbt, branch=source_branch)
-            await diff_coordinator.update_branch_diff(base_branch=destination_branch, diff_branch=source_branch)
-
-
 @flow(name="proposed-changed-pipeline")
 async def pipeline(message: messages.RequestProposedChangePipeline, service: InfrahubServices) -> None:
     async with service.task_report(
@@ -164,14 +149,15 @@ async def pipeline(message: messages.RequestProposedChangePipeline, service: Inf
             branch=message.source_branch
         ):
             await task_report.info("Adding Data Integrity job", proposed_change=message.proposed_change)
-            events.append(
-                messages.RequestProposedChangeDataIntegrity(
-                    proposed_change=message.proposed_change,
-                    source_branch=message.source_branch,
-                    source_branch_sync_with_git=message.source_branch_sync_with_git,
-                    destination_branch=message.destination_branch,
-                    branch_diff=branch_diff,
-                )
+            model = RequestProposedChangeDataIntegrity(
+                proposed_change=message.proposed_change,
+                source_branch=message.source_branch,
+                source_branch_sync_with_git=message.source_branch_sync_with_git,
+                destination_branch=message.destination_branch,
+                branch_diff=branch_diff,
+            )
+            await service.workflow.submit_workflow(
+                workflow=REQUEST_PROPOSED_CHANGE_DATA_INTEGRITY, parameters={"model": model}
             )
 
         if message.check_type in [CheckType.REPOSITORY, CheckType.USER]:

backend/infrahub/message_bus/messages/proposed_change/request_proposedchange_dataintegrity.py renamed to backend/infrahub/proposed_change/models.py

@@ -1,4 +1,4 @@
-from .base_with_diff import BaseProposedChangeWithDiffMessage
+from infrahub.message_bus.messages.proposed_change.base_with_diff import BaseProposedChangeWithDiffMessage
 
 
 class RequestProposedChangeDataIntegrity(BaseProposedChangeWithDiffMessage):

backend/infrahub/proposed_change/tasks.py

@@ -4,10 +4,14 @@
 from prefect import flow, task
 from prefect.logging import get_run_logger
 
+from infrahub.core import registry
 from infrahub.core.constants import ProposedChangeState
-from infrahub.services import (
-    services,
+from infrahub.core.diff.coordinator import DiffCoordinator
+from infrahub.dependencies.registry import get_component_registry
+from infrahub.proposed_change.models import (
+    RequestProposedChangeDataIntegrity,  # noqa: TCH001. as symbol is required by prefect flow
 )
+from infrahub.services import services
 
 
 @flow(name="proposed-changes-cancel-branch", description="Cancel all Proposed change associated with a branch.")
@@ -39,3 +43,19 @@ async def cancel_proposed_change(proposed_change: CoreProposedChange) -> None:
     proposed_change = await service.client.get(kind=CoreProposedChange, id=proposed_change.id)
     proposed_change.state.value = ProposedChangeState.CANCELED.value
     await proposed_change.save()
+
+
+@flow(
+    name="proposed-changed-data-integrity",
+    flow_run_name="Triggers data integrity check",
+)
+async def run_proposed_change_data_integrity_check(model: RequestProposedChangeDataIntegrity) -> None:
+    """Triggers a data integrity validation check on the provided proposed change to start."""
+
+    service = services.service
+    destination_branch = await registry.get_branch(db=service.database, branch=model.destination_branch)
+    source_branch = await registry.get_branch(db=service.database, branch=model.source_branch)
+    component_registry = get_component_registry()
+    async with service.database.start_transaction() as dbt:
+        diff_coordinator = await component_registry.get_component(DiffCoordinator, db=dbt, branch=source_branch)
+        await diff_coordinator.update_branch_diff(base_branch=destination_branch, diff_branch=source_branch)

backend/infrahub/workflows/catalogue.py

@@ -1,5 +1,3 @@
-from infrahub.core.constants import BranchSupportType
-
 from .constants import WorkflowTag, WorkflowType
 from .models import WorkerPoolDefinition, WorkflowDefinition
 
@@ -19,15 +17,13 @@
     type=WorkflowType.USER,
     module="infrahub.transformations.tasks",
     function="transform_render_jinja2_template",
-    branch_support=BranchSupportType.AWARE,
 )
 
 TRANSFORM_PYTHON_RENDER = WorkflowDefinition(
     name="transform_render_python",
     type=WorkflowType.USER,
     module="infrahub.transformations.tasks",
     function="transform_python",
-    branch_support=BranchSupportType.AWARE,
 )
 
 ANONYMOUS_TELEMETRY_SEND = WorkflowDefinition(
@@ -43,7 +39,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.core.migrations.schema.tasks",
     function="schema_apply_migrations",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -52,7 +47,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.core.validators.tasks",
     function="schema_validate_migrations",
-    branch_support=BranchSupportType.AWARE,
 )
 
 TRIGGER_ARTIFACT_DEFINITION_GENERATE = WorkflowDefinition(
@@ -74,7 +68,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.core.ipam.tasks",
     function="ipam_reconciliation",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -90,7 +83,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.generators.tasks",
     function="request_generator_definition_run",
-    branch_support=BranchSupportType.AWARE,
 )
 
 REQUEST_ARTIFACT_GENERATE = WorkflowDefinition(
@@ -134,7 +126,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.git.tasks",
     function="create_branch",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -143,7 +134,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.git.tasks",
     function="add_git_repository",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -152,7 +142,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.git.tasks",
     function="add_git_repository_read_only",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -168,7 +157,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.git.tasks",
     function="merge_git_repository",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -177,7 +165,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.core.branch.tasks",
     function="rebase_branch",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -186,7 +173,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.core.branch.tasks",
     function="merge_branch",
-    branch_support=BranchSupportType.AWARE,
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
@@ -195,15 +181,13 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.core.branch.tasks",
     function="delete_branch",
-    branch_support=BranchSupportType.AWARE,
 )
 
 BRANCH_VALIDATE = WorkflowDefinition(
     name="branch-validate",
     type=WorkflowType.INTERNAL,
     module="infrahub.core.branch.tasks",
     function="validate_branch",
-    branch_support=BranchSupportType.AWARE,
 )
 
 BRANCH_CANCEL_PROPOSED_CHANGES = WorkflowDefinition(
@@ -218,7 +202,6 @@
     type=WorkflowType.INTERNAL,
     module="infrahub.groups.tasks",
     function="update_graphql_query_group",
-    branch_support=BranchSupportType.AWARE,
 )
 
 PROCESS_COMPUTED_MACRO = WorkflowDefinition(
@@ -242,6 +225,13 @@
     function="process_transform",
 )
 
+REQUEST_PROPOSED_CHANGE_DATA_INTEGRITY = WorkflowDefinition(
+    name="proposed-changed-data-integrity",
+    type=WorkflowType.INTERNAL,
+    module="infrahub.proposed_change.tasks",
+    function="run_proposed_change_data_integrity_check",
+)
+
 
 worker_pools = [INFRAHUB_WORKER_POOL]
 
@@ -276,4 +266,5 @@
     PROCESS_COMPUTED_MACRO,
     COMPUTED_ATTRIBUTE_SETUP,
     UPDATE_COMPUTED_ATTRIBUTE_TRANSFORM,
+    REQUEST_PROPOSED_CHANGE_DATA_INTEGRITY,
 ]

backend/infrahub/workflows/models.py

@@ -10,7 +10,6 @@
 from typing_extensions import Self
 
 from infrahub import __version__
-from infrahub.core.constants import BranchSupportType
 
 from .constants import TAG_NAMESPACE, WorkflowTag, WorkflowType
 
@@ -40,7 +39,6 @@ class WorkflowDefinition(BaseModel):
     module: str
     function: str
     cron: str | None = None
-    branch_support: BranchSupportType = BranchSupportType.AGNOSTIC
     tags: list[WorkflowTag] = Field(default_factory=list)
 
     @property

backend/tests/integration/message_bus/operations/request/test_proposed_change.py

@@ -13,6 +13,7 @@
 from infrahub.message_bus.types import ProposedChangeBranchDiff
 from infrahub.server import app, app_initialization
 from infrahub.services import InfrahubServices, services
+from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
 from tests.adapters.log import FakeLogger
 from tests.adapters.message_bus import BusRecorder
 from tests.helpers.file_repo import FileRepo
@@ -97,7 +98,7 @@ async def prepare_proposed_change(
     config = Config(api_token=admin_token, requester=test_client.async_request)
     client = InfrahubClient(config=config)
 
-    service = InfrahubServices(message_bus=bus, client=client)
+    service = InfrahubServices(message_bus=bus, client=client, workflow=WorkflowLocalExecution())
     services.prepare(service=service)
 
     repo = await InfrahubRepository.new(id=obj.id, name=file_repo.name, location=file_repo.path, client=client)
@@ -154,7 +155,6 @@ async def test_run_pipeline_validate_requested_jobs(
     ]
 
     assert sorted(bus_post_data_changes.seen_routing_keys) == [
-        "request.proposed_change.data_integrity",
         "request.proposed_change.run_generators",
         "request.proposed_change.run_tests",
         "request.proposed_change.schema_integrity",

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

@@ -536,24 +536,6 @@ For more detailed explanations on how to use these events within Infrahub, see t
 ### Request Proposed Change
 <!-- vale on -->
 
-<!-- vale off -->
-#### Event request.proposed_change.data_integrity
-<!-- vale on -->
-
-**Description**: Sent trigger data integrity checks for a proposed change
-
-**Priority**: 3
-
-<!-- vale off -->
-| Key | Description | Type | Default Value |
-|-----|-------------|------|---------------|
-| **meta** | Meta properties for the message | N/A | None |
-| **proposed_change** | The unique ID of the Proposed Change | string | None |
-| **source_branch** | The source branch of the proposed change | string | None |
-| **source_branch_sync_with_git** | Indicates if the source branch should sync with git | boolean | None |
-| **destination_branch** | The destination branch of the proposed change | string | None |
-| **branch_diff** | The calculated diff between the two branches | N/A | None |
-<!-- vale on -->
 <!-- vale off -->
 #### Event request.proposed_change.pipeline
 <!-- vale on -->
@@ -1298,25 +1280,6 @@ For more detailed explanations on how to use these events within Infrahub, see t
 ### Request Proposed Change
 <!-- vale on -->
 
-<!-- vale off -->
-#### Event request.proposed_change.data_integrity
-<!-- vale on -->
-
-**Description**: Sent trigger data integrity checks for a proposed change
-
-**Priority**: 3
-
-
-<!-- vale off -->
-| Key | Description | Type | Default Value |
-|-----|-------------|------|---------------|
-| **meta** | Meta properties for the message | N/A | None |
-| **proposed_change** | The unique ID of the Proposed Change | string | None |
-| **source_branch** | The source branch of the proposed change | string | None |
-| **source_branch_sync_with_git** | Indicates if the source branch should sync with git | boolean | None |
-| **destination_branch** | The destination branch of the proposed change | string | None |
-| **branch_diff** | The calculated diff between the two branches | N/A | None |
-<!-- vale on -->
 <!-- vale off -->
 #### Event request.proposed_change.pipeline
 <!-- vale on -->