feat(writer): allow CRON trigger block to be run - WF-881 (#1250)

* fix(writer): allow CRON trigger block to be run - WF-881

* feat(writer): add support for CRON blueprints and triggers in BlueprintRunner - AB-881

* refactor(writer): remove trigger_type parameter from run_blueprint_via_api and streamline trigger selection logic - AB-881

Author: madeindjs

src/writer/blueprints.py

@@ -69,6 +69,10 @@ def __init__(self, session: writer.core.WriterSession):
     def api_blueprints(self):
         return self._gather_api_blueprints()
 
+    @property
+    def cron_blueprints(self):
+        return self._gather_cron_blueprints()
+
     @contextmanager
     def _get_executor(self) -> Generator[ThreadPoolExecutor, None, None]:
         """Return the application's thread pool executor.
@@ -137,6 +141,17 @@ def is_blueprint_api_available(
         """
         return blueprint_id in self.api_blueprints
 
+    def is_blueprint_cron_available(
+        self, blueprint_id: str
+    ):
+        """
+        Checks if a blueprint with the given key is available for Cron execution.
+
+        :param blueprint_id: The blueprint identifier.
+        :return: True if the blueprint is available for Cron execution, False otherwise.
+        """
+        return blueprint_id in self.cron_blueprints
+
     def get_blueprint_api_trigger(
         self, blueprint_id: str
     ):
@@ -152,48 +167,79 @@ def get_blueprint_api_trigger(
             )
         return self.api_blueprints[blueprint_id]
 
-    def _gather_api_blueprints(self):
+    def get_blueprint_cron_trigger(
+        self, blueprint_id: str
+    ):
         """
-        Gathers all blueprints that have an API trigger.
+        Retrieves the Cron trigger for a given blueprint key.
+
+        :param blueprint_key: The blueprint identifier.
+        :return: The Cron trigger component.
+        """
+        if not self.is_blueprint_cron_available(blueprint_id):
+            raise ValueError(
+                f'Cron trigger not found for blueprint "{blueprint_id}".'
+            )
+        return self.cron_blueprints[blueprint_id]
 
-        :return: A set of blueprint keys that have an API trigger.
+    def _gather_blueprints_by_trigger(self, trigger_type: str):
+        """
+        Gathers all blueprints that have a trigger of the specified type.
+
+        :param trigger_type: The trigger component type (e.g., "blueprints_apitrigger").
+        :return: A dict mapping blueprint IDs to their trigger IDs.
         """
         triggers = [
             c for c in self.session.session_component_tree.components.values()
-            if c.type == "blueprints_apitrigger"
-            ]
-        api_blueprints = {}
+            if c.type == trigger_type
+        ]
+        blueprints = {}
 
         for trigger in triggers:
             parent_blueprint_id = \
                 self.session.session_component_tree.get_parent(trigger.id)[0]
             parent_blueprint = \
                 self.session.session_component_tree.get_component(
                     parent_blueprint_id
-                    )
+                )
 
             if (
                 parent_blueprint
                 and
                 parent_blueprint.type == "blueprints_blueprint"
             ):
                 # Store the blueprint key against its trigger ID
-                api_blueprints[parent_blueprint_id] = \
-                    trigger.id
+                blueprints[parent_blueprint_id] = trigger.id
 
-        return api_blueprints
+        return blueprints
+
+    def _gather_api_blueprints(self):
+        """
+        Gathers all blueprints that have an API trigger.
+
+        :return: A dict mapping blueprint IDs to their API trigger IDs.
+        """
+        return self._gather_blueprints_by_trigger("blueprints_apitrigger")
+
+    def _gather_cron_blueprints(self):
+        """
+        Gathers all blueprints that have a Cron trigger.
+
+        :return: A dict mapping blueprint IDs to their Cron trigger IDs.
+        """
+        return self._gather_blueprints_by_trigger("blueprints_crontrigger")
 
     def run_blueprint_via_api(
         self,
         blueprint_id: str,
-        trigger_type: Literal["API", "Cron"],
         branch_id: Optional[str] = None,
         execution_environment: Optional[Dict[str, Any]] = None
     ):
         """
         Executes a blueprint by its key via the API.
 
         :param blueprint_id: The blueprint identifier.
+        :param branch_id: Optional branch ID to start execution from.
         :param execution_environment: The execution environment for
         the blueprint.
         :return: The result of the blueprint execution.
@@ -202,9 +248,24 @@ def run_blueprint_via_api(
             execution_environment = {}
 
         trigger_id = branch_id
-        if trigger_id is None:
+        if trigger_id is not None:
+            # Determine trigger type from the component
+            component = self.session.session_component_tree.get_component(trigger_id)
+            if component and component.type == "blueprints_apitrigger":
+                trigger_type = "API"
+            elif component and component.type == "blueprints_crontrigger":
+                trigger_type = "Cron"
+            else:
+                trigger_type = "Branch"
+        elif self.is_blueprint_api_available(blueprint_id):
+            # Prioritize API trigger over Cron if both exist
             trigger_id = self.get_blueprint_api_trigger(blueprint_id)
-
+            trigger_type = "API"
+        elif self.is_blueprint_cron_available(blueprint_id):
+            trigger_id = self.get_blueprint_cron_trigger(blueprint_id)
+            trigger_type = "Cron"
+        else:
+            raise ValueError(f'No trigger found for blueprint "{blueprint_id}".')
         return self.run_branch(
             trigger_id,
             None,

src/writer/core.py

@@ -1283,15 +1283,11 @@ def run_blueprint_via_api(
         blueprint_id = payload.pop("blueprint_id", None)
         if not blueprint_id:
             raise ValueError("Missing blueprint_id in payload")
-        trigger_type = payload.pop("trigger_type", None)
-        if not trigger_type:
-            raise ValueError("Missing trigger_type in payload")
         execution_environment = EventHandler._get_blueprint_execution_environment(
             payload, context, session, vault
         )
         return blueprint_runner.run_blueprint_via_api(
             blueprint_id=blueprint_id,
-            trigger_type=trigger_type,
             branch_id=payload.pop("branch_id", None),
             execution_environment=execution_environment,
         )

src/writer/serve.py

@@ -385,12 +385,13 @@ async def _get_payload_as_json(request: Request):
             raise HTTPException(status_code=400, detail="Cannot parse the payload.")
         return payload
 
-    def has_api_trigger(app_runner: AppRunner, blueprint_id: str) -> bool:
-        # Check if blueprint has at least one API trigger component
+    def is_blueprint_triggerable(app_runner: AppRunner, blueprint_id: str) -> bool:
+        """Check if blueprint has at least one supported trigger (API or Cron)."""
         if not app_runner.bmc_components:
             return False
+        supported_triggers = ("blueprints_apitrigger", "blueprints_crontrigger")
         return any(
-            comp["type"] == "blueprints_apitrigger" and comp.get("parentId") == blueprint_id
+            comp["type"] in supported_triggers and comp.get("parentId") == blueprint_id
             for comp in app_runner.bmc_components.values()
         )
 
@@ -409,7 +410,7 @@ async def get_blueprints(request: Request):
             }
             for comp in app_runner.bmc_components.values()
             if comp["type"] == "blueprints_blueprint"
-            and has_api_trigger(app_runner, comp["id"])
+            and is_blueprint_triggerable(app_runner, comp["id"])
         ]
 
         return JSONResponse(content=blueprints)
@@ -520,9 +521,9 @@ async def event_logic(queue: asyncio.Queue):
                     }))
                     return
 
-                if not branch_id and not has_api_trigger(app_runner, blueprint_id):
+                if not branch_id and not is_blueprint_triggerable(app_runner, blueprint_id):
                     await queue.put(await format_event("error", {
-                        "msg": f"Blueprint '{blueprint_id}' lacks an API trigger.",
+                        "msg": f"Blueprint '{blueprint_id}' lacks a supported trigger (API or Cron).",
                         "finished_at": int(time.time())
                     }))
                     return
@@ -538,39 +539,21 @@ async def event_logic(queue: asyncio.Queue):
 
                 await queue.put(await format_event("status", {"status": "executing", "msg": (f"Executing branch: {branch_id}..." if branch_id else f"Executing blueprint: {blueprint_id}...")}))
 
-                if branch_id:
-                    task = asyncio.create_task(
-                        app_runner.handle_event(
-                            session_id,
-                            WriterEvent(
-                                type="wf-run-blueprint-via-api",
-                                isSafe=True,
-                                handler="run_blueprint_via_api",
-                                payload={
-                                    "blueprint_id": blueprint_id,
-                                    "trigger_type": "Cron",
-                                    "branch_id": branch_id,
-                                    **(payload or {})
-                                },
-                            )
-                        )
-                    )
-                else:
-                    task = asyncio.create_task(
-                        app_runner.handle_event(
-                            session_id,
-                            WriterEvent(
-                                type="wf-run-blueprint-via-api",
-                                isSafe=True,
-                                handler="run_blueprint_via_api",
-                                payload={
-                                    "blueprint_id": blueprint_id,
-                                    "trigger_type": "API",
-                                    **(payload or {})
-                                },
-                            )
+                task = asyncio.create_task(
+                    app_runner.handle_event(
+                        session_id,
+                        WriterEvent(
+                            type="wf-run-blueprint-via-api",
+                            isSafe=True,
+                            handler="run_blueprint_via_api",
+                            payload={
+                                "blueprint_id": blueprint_id,
+                                "branch_id": branch_id,
+                                **(payload or {})
+                            },
                         )
                     )
+                )
 
                 await queue.put(await format_event("status", {"status": "running", "msg": ("Branch is running. Awaiting output..." if branch_id else "Blueprint is running. Awaiting output...")}))