Merge pull request #2 from unity-sds/413-submit-ogc

Add SRL Initiator for EDRGen and New OGC Action Type

Author: drewm-swe

README.md

@@ -34,7 +34,9 @@ def execute(self):
             },
             "note": "",
         }
-        response = httpx.post(url, auth=auth, headers=headers, json=body)
+        response = httpx.post(
+            url, auth=auth, headers=headers, json=body, verify=False
+        )  # nosec
         if response.status_code in (200, 201):
             success = True
             resp = response.json()

src/unity_initiator/actions/submit_dag_by_id.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, verify=False)  # nosec
+        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/actions/submit_ogc_process_execution.py

@@ -25,7 +25,7 @@ initiator_config:
 # Configuration for matching payload (e.g. url) against a set of compiled regular expressions
 # and mapping any matches to a set of evaluators.
 regex_config:
-  regexes: list(compiled_regex(), required=True, min=1)
+  regexes: list(str, required=True, min=1)
   evaluators: list(include("evaluator_config"), required=True, min=1)
 
 # Configuration of actions that submit to evaluators.
@@ -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)
+  actions: list(include("action_config"), required=True, min=1, max=1)

src/unity_initiator/resources/routers_schema.yaml

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import re
 
 from .evaluator import Evaluator
 from .utils.conf_utils import YamlConf, YamlConfEncoder
@@ -14,6 +15,15 @@ class Router:
     def __init__(self, config_file):
         self._config_file = config_file
         self._config = YamlConf(self._config_file)
+        self._compile_regexes()
+
+    def _compile_regexes(self):
+        """Compile all regex strings in the configuration."""
+        for url_cfg in (
+            self._config.get("initiator_config").get("payload_type").get("url", [])
+        ):
+            if "regexes" in url_cfg:
+                url_cfg["regexes"] = [re.compile(regex) for regex in url_cfg["regexes"]]
 
     def get_evaluators_by_url(self, url):
         found_match = False

src/unity_initiator/router.py

@@ -5,7 +5,7 @@
 
 | Name | Version |
 |------|---------|
-| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | ~> 1.4.6 |
+| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | ~> 1.8.2 |
 | <a name="requirement_archive"></a> [archive](#requirement\_archive) | >=2.4.2 |
 | <a name="requirement_aws"></a> [aws](#requirement\_aws) | >=5.50.0 |
 | <a name="requirement_local"></a> [local](#requirement\_local) | >=2.5.1 |