Convert all operations to prefect flow

Author: dgarros

backend/infrahub/message_bus/operations/check/artifact.py

@@ -1,5 +1,7 @@
 from typing import Union
 
+from prefect import flow
+
 from infrahub.core.constants import InfrahubKind, ValidatorConclusion
 from infrahub.core.timestamp import Timestamp
 from infrahub.git.repository import InfrahubReadOnlyRepository, InfrahubRepository
@@ -12,6 +14,7 @@
 log = get_logger()
 
 
+@flow(name="git-repository-check-artifact-create")
 async def create(message: messages.CheckArtifactCreate, service: InfrahubServices) -> None:
     log.debug("Creating artifact", message=message)
     validator = await service.client.get(

backend/infrahub/message_bus/operations/check/generator.py

@@ -3,6 +3,7 @@
 from infrahub_sdk import InfrahubNode
 from infrahub_sdk.exceptions import ModuleImportError
 from infrahub_sdk.schema import InfrahubGeneratorDefinitionConfig
+from prefect import flow
 
 from infrahub import lock
 from infrahub.core.constants import GeneratorInstanceStatus, InfrahubKind, ValidatorConclusion
@@ -16,6 +17,7 @@
 # pylint: disable=duplicate-code
 
 
+@flow(name="git-repository-check-generator-run")
 async def run(message: messages.CheckGeneratorRun, service: InfrahubServices) -> None:
     repository = await get_initialized_repo(
         repository_id=message.repository_id,

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

@@ -1,6 +1,7 @@
 from typing import List
 
 from infrahub_sdk import UUIDT
+from prefect import flow
 
 from infrahub import lock
 from infrahub.core.constants import InfrahubKind
@@ -16,6 +17,7 @@
 log = get_logger()
 
 
+@flow(name="git-repository-check-definition")
 async def check_definition(message: messages.CheckRepositoryCheckDefinition, service: InfrahubServices) -> None:
     definition = await service.client.get(
         kind=InfrahubKind.CHECKDEFINITION, id=message.check_definition_id, branch=message.branch_name
@@ -141,6 +143,7 @@ async def check_definition(message: messages.CheckRepositoryCheckDefinition, ser
             await service.send(message=event)
 
 
+@flow(name="git-repository-check-merge-conflict")
 async def merge_conflicts(message: messages.CheckRepositoryMergeConflicts, service: InfrahubServices) -> None:
     """Runs a check to see if there are merge conflicts between two branches."""
     log.info(
@@ -222,6 +225,7 @@ async def merge_conflicts(message: messages.CheckRepositoryMergeConflicts, servi
     )
 
 
+@flow(name="git-repository-user-check")
 async def user_check(message: messages.CheckRepositoryUserCheck, service: InfrahubServices) -> None:
     validator = await service.client.get(kind=InfrahubKind.USERVALIDATOR, id=message.validator_id)
     await validator.checks.fetch()

backend/infrahub/message_bus/operations/finalize/validator.py

@@ -1,3 +1,5 @@
+from prefect import flow
+
 from infrahub import config
 from infrahub.core.timestamp import Timestamp
 from infrahub.log import get_logger
@@ -8,6 +10,7 @@
 log = get_logger()
 
 
+@flow(name="validator-finalize-execution")
 async def execution(message: messages.FinalizeValidatorExecution, service: InfrahubServices) -> None:
     """Monitors the status of checks associated with a validator and finalizes the conclusion of the validator
 

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

@@ -1,3 +1,5 @@
+from prefect import flow
+
 from infrahub import lock
 from infrahub.git.repository import InfrahubRepository
 from infrahub.log import get_logger
@@ -7,6 +9,7 @@
 log = get_logger()
 
 
+@flow(name="git-repository-branch-create")
 async def create(message: messages.GitBranchCreate, service: InfrahubServices) -> None:
     log.info("creating branch in repository", branch=message.branch, repository=message.repository_name)
     repo = await InfrahubRepository.init(id=message.repository_id, name=message.repository_name, client=service.client)

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

@@ -1,3 +1,5 @@
+from prefect import flow
+
 from infrahub.git.repository import get_initialized_repo
 from infrahub.log import get_logger
 from infrahub.message_bus import messages
@@ -7,6 +9,7 @@
 log = get_logger()
 
 
+@flow(name="git-repository-diff-files-names-only")
 async def names_only(message: messages.GitDiffNamesOnly, service: InfrahubServices) -> None:
     log.info(
         "Collecting modifications between commits",

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

@@ -1,3 +1,5 @@
+from prefect import flow
+
 from infrahub.git.repository import get_initialized_repo
 from infrahub.log import get_logger
 from infrahub.message_bus import messages
@@ -7,6 +9,7 @@
 log = get_logger()
 
 
+@flow(name="git-repository-get-file")
 async def get(message: messages.GitFileGet, service: InfrahubServices) -> None:
     log.info("Collecting file from repository", repository=message.repository_name, file=message.file)
 

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

@@ -1,3 +1,5 @@
+from prefect import flow
+
 from infrahub import lock
 from infrahub.core.constants import InfrahubKind, RepositoryInternalStatus
 from infrahub.exceptions import RepositoryError
@@ -13,6 +15,7 @@
 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",
@@ -43,6 +46,7 @@ async def add(message: messages.GitRepositoryAdd, service: InfrahubServices) ->
                 await repo.sync()
 
 
+@flow(name="git-repository-add-read-only")
 async def add_read_only(message: messages.GitRepositoryAddReadOnly, service: InfrahubServices) -> None:
     log.info(
         "Cloning and importing read-only repository", repository=message.repository_name, location=message.location
@@ -67,6 +71,7 @@ async def add_read_only(message: messages.GitRepositoryAddReadOnly, service: Inf
                 await repo.sync_from_remote()
 
 
+@flow(name="git-repository-check-connectivity")
 async def connectivity(message: messages.GitRepositoryConnectivity, service: InfrahubServices) -> None:
     response_data = GitRepositoryConnectivityResponseData(message="Successfully accessed repository", success=True)
 
@@ -83,6 +88,7 @@ async def connectivity(message: messages.GitRepositoryConnectivity, service: Inf
         await service.reply(message=response, initiator=message)
 
 
+@flow(name="git-repository-import-object")
 async def import_objects(message: messages.GitRepositoryImportObjects, service: InfrahubServices) -> None:
     async with service.git_report(
         related_node=message.repository_id,

backend/infrahub/message_bus/operations/refresh/registry.py

@@ -1,3 +1,5 @@
+from prefect import flow
+
 from infrahub import lock
 from infrahub.core.registry import registry
 from infrahub.message_bus import messages
@@ -6,6 +8,7 @@
 from infrahub.worker import WORKER_IDENTITY
 
 
+@flow(name="registry-branch-refresh")
 async def branches(message: messages.RefreshRegistryBranches, service: InfrahubServices) -> None:
     if message.meta and message.meta.initiator_id == WORKER_IDENTITY:
         service.log.info("Ignoring refresh registry refresh request originating from self", worker=WORKER_IDENTITY)
@@ -17,6 +20,7 @@ async def branches(message: messages.RefreshRegistryBranches, service: InfrahubS
     await service.component.refresh_schema_hash()
 
 
+@flow(name="registry-branch-rebase")
 async def rebased_branch(message: messages.RefreshRegistryRebasedBranch, service: InfrahubServices) -> None:
     if message.meta and message.meta.initiator_id == WORKER_IDENTITY:
         service.log.info(

backend/infrahub/message_bus/operations/refresh/webhook.py

@@ -1,10 +1,12 @@
 import ujson
+from prefect import flow
 
 from infrahub.core.constants import InfrahubKind
 from infrahub.message_bus import messages
 from infrahub.services import InfrahubServices
 
 
+@flow(name="registry-webhook-config-refresh")
 async def configuration(
     message: messages.RefreshWebhookConfiguration,  # pylint: disable=unused-argument
     service: InfrahubServices,