Add flow decorator to all tasks

Author: dgarros

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

@@ -1,5 +1,7 @@
 from typing import List
 
+from prefect import flow
+
 from infrahub.core import registry
 from infrahub.core.diff.model.path import BranchTrackingId
 from infrahub.core.diff.repository.repository import DiffRepository
@@ -11,6 +13,7 @@
 log = get_logger()
 
 
+@flow(name="event-branch-create")
 async def create(message: messages.EventBranchCreate, service: InfrahubServices) -> None:
     log.info("run_message", branch=message.branch)
 
@@ -23,6 +26,7 @@ async def create(message: messages.EventBranchCreate, service: InfrahubServices)
         await service.send(message=event)
 
 
+@flow(name="event-branch-delete")
 async def delete(message: messages.EventBranchDelete, service: InfrahubServices) -> None:
     log.info("Branch was deleted", branch=message.branch)
 
@@ -36,6 +40,7 @@ async def delete(message: messages.EventBranchDelete, service: InfrahubServices)
         await service.send(message=event)
 
 
+@flow(name="branch-event-merge")
 async def merge(message: messages.EventBranchMerge, service: InfrahubServices) -> None:
     log.info("Branch merged", source_branch=message.source_branch, target_branch=message.target_branch)
 
@@ -64,6 +69,7 @@ async def merge(message: messages.EventBranchMerge, service: InfrahubServices) -
         await service.send(message=event)
 
 
+@flow(name="event-branch-rebased")
 async def rebased(message: messages.EventBranchRebased, service: InfrahubServices) -> None:
     log.info("Branch rebased", branch=message.branch)
 

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

@@ -1,5 +1,7 @@
 from typing import List
 
+from prefect import flow
+
 from infrahub.core.constants import InfrahubKind
 from infrahub.log import get_logger
 from infrahub.message_bus import InfrahubMessage, messages
@@ -8,6 +10,7 @@
 log = get_logger()
 
 
+@flow(name="event-node-mutated")
 async def mutated(
     message: messages.EventNodeMutated,
     service: InfrahubServices,

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

@@ -1,10 +1,13 @@
+from prefect import flow
+
 from infrahub.log import get_logger
 from infrahub.message_bus import messages
 from infrahub.services import InfrahubServices
 
 log = get_logger()
 
 
+@flow(name="event-schema-update")
 async def update(message: messages.EventSchemaUpdate, service: InfrahubServices) -> None:
     log.info("run_message", branch=message.branch)
 

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

@@ -1,7 +1,10 @@
+from prefect import flow
+
 from infrahub.message_bus import messages
 from infrahub.services import InfrahubServices
 
 
+@flow(name="event-worker-newprimary-api")
 async def new_primary_api(message: messages.EventWorkerNewPrimaryAPI, service: InfrahubServices) -> None:
     service.log.info("api_worker promoted to primary", worker_id=message.worker_id)
 

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

@@ -104,6 +104,7 @@ async def import_objects(message: messages.GitRepositoryImportObjects, service:
         await repo.import_objects_from_files(infrahub_branch_name=message.infrahub_branch_name, commit=message.commit)
 
 
+@flow(name="git-repository-pull-read-only")
 async def pull_read_only(message: messages.GitRepositoryPullReadOnly, service: InfrahubServices) -> None:
     if not message.ref and not message.commit:
         log.warning(
@@ -154,6 +155,7 @@ async def pull_read_only(message: messages.GitRepositoryPullReadOnly, service: I
             await repo.sync_from_remote(commit=message.commit)
 
 
+@flow(name="git-repository-merge")
 async def merge(message: messages.GitRepositoryMerge, service: InfrahubServices) -> None:
     log.info(
         "Merging repository branch",

backend/infrahub/message_bus/operations/schema/migration.py

@@ -1,4 +1,4 @@
-from prefect import task
+from prefect import flow, task
 from prefect.logging import get_run_logger
 from prefect.runtime import task_run
 
@@ -15,6 +15,7 @@
 log = get_logger()
 
 
+@flow(name="schema-migration-path")
 async def path(message: SchemaMigrationPath, service: InfrahubServices) -> None:
     async with service.database.start_session() as db:
         node_kind = None

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

@@ -1,4 +1,4 @@
-from prefect import task
+from prefect import flow, task
 from prefect.runtime import task_run
 
 from infrahub.core.validators.aggregated_checker import AggregatedConstraintChecker
@@ -16,6 +16,7 @@
 log = get_logger()
 
 
+@flow(name="schema-validator-path")
 async def path(message: SchemaValidatorPath, service: InfrahubServices) -> None:
     async with service.database.start_session() as db:
         log.info(

backend/infrahub/message_bus/operations/send/echo.py

@@ -1,8 +1,11 @@
+from prefect import flow
+
 from infrahub.message_bus import messages
 from infrahub.message_bus.messages.send_echo_request import SendEchoRequestResponse, SendEchoRequestResponseData
 from infrahub.services import InfrahubServices
 
 
+@flow(name="echo-request")
 async def request(message: messages.SendEchoRequest, service: InfrahubServices) -> None:
     service.log.info(f"Received message: {message.message}")
     if message.reply_requested:

backend/infrahub/message_bus/operations/send/telemetry.py

@@ -96,6 +96,7 @@ async def gather_anonymous_telemetry_data(service: InfrahubServices) -> dict:
     return data
 
 
+@flow(name="telemetry-push-legacy")
 async def push(
     message: messages.SendTelemetryPush,  # pylint: disable=unused-argument
     service: InfrahubServices,

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

@@ -11,6 +11,7 @@
 from infrahub.webhook import CustomWebhook, StandardWebhook, TransformWebhook, Webhook
 
 
+@flow(name="event-send-webhook")
 async def event(message: messages.SendWebhookEvent, service: InfrahubServices) -> None:
     async with service.task_report(
         related_node=message.webhook_id,