feat: New initiator action - OGC process execution

Author: Drew Meyers

src/unity_initiator/actions/submit_ogc_process_execution.py

@@ -0,0 +1,50 @@
+import httpx
+
+from ..utils.logger import logger
+from .base import Action
+
+__all__ = ["SubmitOgcProcessExecution"]
+
+
+class SubmitOgcProcessExecution(Action):
+    def __init__(self, payload, payload_info, params):
+        super().__init__(payload, payload_info, params)
+        logger.info("instantiated %s", __class__.__name__)
+
+    def execute(self):
+        logger.debug("executing execute in %s", __class__.__name__)
+        url = f"{self._params['ogc_processes_base_api_endpoint']}/processes/{self._params['process_id']}/execution"
+        logger.info("url: %s", url)
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+        # body = {
+        #     "inputs": self._params["execution_inputs"],
+        #     "outputs": self._params["execution_outputs"],
+        #     "subscriber": self._params["execution_subscriber"],
+        # }
+        body = {
+            "inputs": {
+                "payload": self._payload,
+                "payload_info": self._payload_info,
+                "on_success": self._params["on_success"],
+            },
+            "outputs": None,
+            "subscriber": None,
+        }
+        response = httpx.post(url, headers=headers, json=body)
+        if response.status_code in (200, 201):
+            success = True
+            resp = response.json()
+            logger.info(
+                "Successfully triggered the execution of the OGC Process %s: %s",
+                self._params["process_id"],
+                resp,
+            )
+        else:
+            success = False
+            resp = response.text
+            logger.info(
+                "Failed to trigger the execution of the OGC Process %s: %s",
+                self._params["process_id"],
+                resp,
+            )
+        return {"success": success, "response": resp}

src/unity_initiator/resources/routers_schema.yaml

@@ -36,7 +36,7 @@ evaluator_config:
 # Currently only 2 types of actions are supported:
 # 1. submit payload to an SNS topic
 # 2. submit payload to an airflow DAG
-action_config: any(include("submit_dag_by_id_action"), include("submit_to_sns_topic_action"))
+action_config: any(include("submit_dag_by_id_action"), include("submit_to_sns_topic_action"), include("submit_ogc_process_execution_action"))
 
 # Configuration for submitting a payload to an airflow DAG.
 submit_dag_by_id_action:
@@ -58,6 +58,17 @@ submit_to_sns_topic_action:
     topic_arn: str(required=False)
     on_success: include("on_success_actions", required=False)
 
+# Configuration for submitting a OGC process execution.
+submit_ogc_process_execution_action:
+  name: str(equals="submit_ogc_process_execution")
+  params:
+    process_id: str()
+    ogc_processes_base_api_endpoint: str(required=False)
+    execution_inputs: map(required=False)
+    execution_outputs: map(required=False)
+    execution_subscriber: map(required=False)
+    on_success: include("on_success_actions", required=False)
+
 # Configuration to pass onto the evaluator to use when evaluation is a success.
 on_success_actions:
   actions: list(include("action_config"), required=True, min=1, max=1)

tests/resources/test_router.yaml

@@ -129,3 +129,18 @@ initiator_config:
                           #airflow_base_api_endpoint: xxx
                           #airflow_username: <ARN to username entry in AWS Secrets Manager>
                           #airflow_password: <ARN to password entry in Secrets Manager>
+
+      - regexes:
+          - '(?<=/)(?P<filename>hello_world\.txt)$'
+        evaluators:
+          - name: eval_hello_world_readiness
+            actions:
+              - name: submit_ogc_process_execution
+                params:
+                  process_id: eval_hello_world_readiness
+                  ogc_processes_base_api_endpoint: ${ogc_processes_base_api_endpoint}
+                  on_success:
+                    actions:
+                      - name: submit_ogc_process_execution
+                        params:
+                          process_id: hello_world