Merge pull request #4784 from opsmill/lgu-migrate-trigger-gen-def-run

Migrate TriggerGeneratorDefinitionRun to prefect

Author: LucasG0

backend/infrahub/generators/tasks.py

@@ -7,10 +7,12 @@
 from prefect import flow, task
 
 from infrahub import lock
-from infrahub.core.constants import GeneratorInstanceStatus
+from infrahub.core.constants import GeneratorInstanceStatus, InfrahubKind
 from infrahub.generators.models import RequestGeneratorRun
 from infrahub.git.base import extract_repo_file_information
 from infrahub.git.repository import get_initialized_repo
+from infrahub.message_bus import messages
+from infrahub.message_bus.types import ProposedChangeGeneratorDefinition
 from infrahub.services import InfrahubServices, services
 
 
@@ -103,3 +105,34 @@ async def _define_instance(model: RequestGeneratorRun, service: InfrahubServices
                 )
                 await instance.save()
     return instance
+
+
+@flow(name="generator_definition_run")
+async def run_generator_definition(branch: str) -> None:
+    service = services.service
+    generators = await service.client.filters(
+        kind=InfrahubKind.GENERATORDEFINITION, prefetch_relationships=True, populate_store=True, branch=branch
+    )
+
+    generator_definitions = [
+        ProposedChangeGeneratorDefinition(
+            definition_id=generator.id,
+            definition_name=generator.name.value,
+            class_name=generator.class_name.value,
+            file_path=generator.file_path.value,
+            query_name=generator.query.peer.name.value,
+            query_models=generator.query.peer.models.value,
+            repository_id=generator.repository.peer.id,
+            parameters=generator.parameters.value,
+            group_id=generator.targets.peer.id,
+            convert_query_response=generator.convert_query_response.value,
+        )
+        for generator in generators
+    ]
+
+    events = [
+        messages.RequestGeneratorDefinitionRun(branch=branch, generator_definition=generator_definition)
+        for generator_definition in generator_definitions
+    ]
+    for event in events:
+        await service.send(message=event)

backend/infrahub/message_bus/messages/__init__.py

@@ -41,7 +41,6 @@
 from .schema_migration_path import SchemaMigrationPath, SchemaMigrationPathResponse
 from .schema_validator_path import SchemaValidatorPath, SchemaValidatorPathResponse
 from .send_echo_request import SendEchoRequest, SendEchoRequestResponse
-from .trigger_generatordefinition_run import TriggerGeneratorDefinitionRun
 from .trigger_proposed_change_cancel import TriggerProposedChangeCancel
 from .trigger_webhook_actions import TriggerWebhookActions
 
@@ -87,7 +86,6 @@
     "request.repository.checks": RequestRepositoryChecks,
     "request.repository.user_checks": RequestRepositoryUserChecks,
     "send.echo.request": SendEchoRequest,
-    "trigger.generator_definition.run": TriggerGeneratorDefinitionRun,
     "trigger.proposed_change.cancel": TriggerProposedChangeCancel,
     "trigger.webhook.actions": TriggerWebhookActions,
 }

backend/infrahub/message_bus/operations/__init__.py

@@ -61,7 +61,6 @@
     "send.echo.request": send.echo.request,
     "schema.migration.path": schema.migration.path,
     "schema.validator.path": schema.validator.path,
-    "trigger.generator_definition.run": trigger.generator_definition.run,
     "trigger.proposed_change.cancel": trigger.proposed_change.cancel,
     "trigger.webhook.actions": trigger.webhook.actions,
 }

backend/infrahub/message_bus/operations/event/branch.py

@@ -15,6 +15,7 @@
     REQUEST_DIFF_REFRESH,
     REQUEST_DIFF_UPDATE,
     TRIGGER_ARTIFACT_DEFINITION_GENERATE,
+    TRIGGER_GENERATOR_DEFINITION_RUN,
 )
 
 log = get_logger()
@@ -56,7 +57,6 @@ async def merge(message: messages.EventBranchMerge, service: InfrahubServices) -
 
     events: List[InfrahubMessage] = [
         messages.RefreshRegistryBranches(),
-        messages.TriggerGeneratorDefinitionRun(branch=message.target_branch),
     ]
     component_registry = get_component_registry()
     default_branch = registry.get_branch_from_registry()
@@ -69,6 +69,11 @@ async def merge(message: messages.EventBranchMerge, service: InfrahubServices) -
         parameters={"branch": message.target_branch},
     )
 
+    await service.workflow.submit_workflow(
+        workflow=TRIGGER_GENERATOR_DEFINITION_RUN,
+        parameters={"branch": message.target_branch},
+    )
+
     for diff_root in branch_diff_roots:
         if (
             diff_root.base_branch_name != diff_root.diff_branch_name

backend/infrahub/message_bus/operations/trigger/__init__.py

@@ -1,3 +1,3 @@
-from . import generator_definition, proposed_change, webhook
+from . import proposed_change, webhook
 
-__all__ = ["generator_definition", "proposed_change", "webhook"]
+__all__ = ["proposed_change", "webhook"]

backend/infrahub/message_bus/operations/trigger/generator_definition.py

@@ -62,6 +62,13 @@
     function="generate_artifact_definition",
 )
 
+TRIGGER_GENERATOR_DEFINITION_RUN = WorkflowDefinition(
+    name="generator_definition_run",
+    type=WorkflowType.INTERNAL,
+    module="infrahub.generators.tasks",
+    function="run_generator_definition",
+)
+
 IPAM_RECONCILIATION = WorkflowDefinition(
     name="ipam_reconciliation",
     type=WorkflowType.INTERNAL,
@@ -160,4 +167,5 @@
     REQUEST_GENERATOR_RUN,
     REQUEST_DIFF_UPDATE,
     REQUEST_DIFF_REFRESH,
+    TRIGGER_GENERATOR_DEFINITION_RUN,
 ]

backend/infrahub/workflows/catalogue.py

@@ -13,7 +13,12 @@
 from infrahub.message_bus.operations.event.branch import delete, merge, rebased
 from infrahub.services import InfrahubServices, services
 from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
-from infrahub.workflows.catalogue import REQUEST_DIFF_REFRESH, REQUEST_DIFF_UPDATE, TRIGGER_ARTIFACT_DEFINITION_GENERATE
+from infrahub.workflows.catalogue import (
+    REQUEST_DIFF_REFRESH,
+    REQUEST_DIFF_UPDATE,
+    TRIGGER_ARTIFACT_DEFINITION_GENERATE,
+    TRIGGER_GENERATOR_DEFINITION_RUN,
+)
 from tests.adapters.message_bus import BusRecorder
 
 
@@ -101,6 +106,10 @@ async def test_merged(default_branch: Branch, init_service: InfrahubServices, pr
 
         expected_calls = [
             call(workflow=TRIGGER_ARTIFACT_DEFINITION_GENERATE, parameters={"branch": message.target_branch}),
+            call(
+                workflow=TRIGGER_GENERATOR_DEFINITION_RUN,
+                parameters={"branch": target_branch_name},
+            ),
             call(
                 workflow=REQUEST_DIFF_UPDATE,
                 parameters={"model": RequestDiffUpdate(branch_name=tracked_diff_roots[0].diff_branch_name)},
@@ -118,9 +127,8 @@ async def test_merged(default_branch: Branch, init_service: InfrahubServices, pr
     )
     diff_repo.get_empty_roots.assert_awaited_once_with(base_branch_names=[target_branch_name])
 
-    assert len(service.message_bus.messages) == 2
+    assert len(service.message_bus.messages) == 1
     assert service.message_bus.messages[0] == messages.RefreshRegistryBranches()
-    assert service.message_bus.messages[1] == messages.TriggerGeneratorDefinitionRun(branch=target_branch_name)
 
 
 async def test_rebased(default_branch: Branch, prefect_test_fixture):

backend/tests/unit/message_bus/operations/event/test_branch.py

@@ -1,4 +1,5 @@
 New permissions system in UI:
+
 - Implemented CRUD views for managing accounts, groups, roles, and permissions
 - Updated all components to support new permission system
 - Added dynamic message display according to user access levels

changelog/+permissions.added.md

@@ -1,4 +1,5 @@
 Reworked branch selector:
+
 - Redesigned the UI
 - Added filter for branch
 - Improved accessibility & keyboard navigation