Using pydantic-yaml and validations through pydantic.validators.

Author: Bryan Sieber

poetry.lock

@@ -15,9 +15,9 @@ prometheus-client = "^0.13.1"
 python-bugzilla = "^3.2.0"
 atlassian-python-api = "^3.20.1"
 dockerflow = "2022.1.0"
-PyYAML = "^6.0"
 types-PyYAML = "^6.0.4"
 Jinja2 = "^3.0.3"
+pydantic-yaml = {extras = ["pyyaml"], version = "^0.6.1"}
 
 
 [tool.poetry.dev-dependencies]

pyproject.toml

@@ -7,7 +7,7 @@
 from fastapi.responses import JSONResponse
 
 from src.app import environment
-from src.jbi.services import jbi_service_health_map
+from src.jbi.service import jbi_service_health_map
 
 api_router = APIRouter(tags=["Monitor"])
 

src/app/monitor.py

@@ -1,83 +1,33 @@
-"""Parsing and validating the YAML configuration occurs within this module"""
-import importlib
+"""
+Parsing and validating the YAML configuration occurs within this module
+"""
 import logging
-from inspect import signature
-from types import ModuleType
-from typing import Any, Dict, Optional
+from functools import lru_cache
 
-import yaml
-from yaml import Loader
+from pydantic import ValidationError
 
 from src.app import environment
+from src.jbi.model import Actions
 
 settings = environment.get_settings()
 jbi_logger = logging.getLogger("src.jbi")
 
 
 class ConfigError(Exception):
-    """Errors used to determine when the Configuration is invalid"""
-
-
-class ProcessError(Exception):
-    """Error when an exception occurs during processing"""
+    """Error when an exception occurs during processing config"""
 
 
+@lru_cache
 def get_yaml_configurations(
     jbi_config_file: str = f"config/config.{settings.env}.yaml",
-) -> Dict[str, Dict]:
+) -> Actions:
     """Convert and validate YAML configuration to python dict"""
 
     with open(jbi_config_file, encoding="utf-8") as file:
         try:
-            file_data = file.read()
-            data = yaml.load(file_data, Loader)
-            validated_action_dict = process_actions(
-                action_configuration=data.get("actions")
-            )
-            return validated_action_dict
-        except (ValueError, ConfigError, yaml.YAMLError) as exception:
+            yaml_data = file.read()
+            actions: Actions = Actions.parse_raw(yaml_data)
+            return actions
+        except ValidationError as exception:
             jbi_logger.exception(exception)
-            raise ProcessError("Errors exist.") from exception
-
-
-def process_actions(action_configuration) -> Dict[str, Dict]:
-    """Validate `actions` section of the YAML config"""
-    requested_actions = {}
-    for yaml_action_key, inner_action_dict in action_configuration.items():
-        inner_action_dict.setdefault("action", "src.jbi.whiteboard_actions.default")
-        inner_action_dict.setdefault("enabled", False)
-        inner_action_dict.setdefault("parameters", {})
-        validate_action_yaml_jbi_naming(
-            yaml_action_key=yaml_action_key, action_dict=inner_action_dict
-        )
-        validate_action_yaml_module(action_dict=inner_action_dict)
-        requested_actions[yaml_action_key] = inner_action_dict
-    return requested_actions
-
-
-def validate_action_yaml_jbi_naming(yaml_action_key, action_dict):
-    """Validate yaml_action_key == parameters.whiteboard_tag"""
-    wb_tag = action_dict["parameters"].get("whiteboard_tag")
-    if yaml_action_key != wb_tag:
-        raise ConfigError(
-            f"Expected action key '{wb_tag}', found `{yaml_action_key}."
-            "(from the `parameters.whiteboard_tag` field)."
-        )
-
-
-def validate_action_yaml_module(action_dict: Dict[str, Any]):
-    """Validate action: exists, has init function, and has expected params"""
-    try:
-        action: str = action_dict.get("action")  # type: ignore
-        action_parameters: Optional[Dict[str, Any]] = action_dict.get("parameters")
-        action_module: ModuleType = importlib.import_module(action)
-        if not action_module:
-            raise TypeError("Module is not found.")
-        if not hasattr(action_module, "init"):
-            raise TypeError("Module is missing `init` method.")
-
-        signature(action_module.init).bind(**action_parameters)  # type: ignore
-    except ImportError as exception:
-        raise ConfigError(f"Unknown action `{action}`.") from exception
-    except (TypeError, AttributeError) as exception:
-        raise ConfigError("Action is not properly setup.") from exception
+            raise ConfigError("Errors exist.") from exception

src/jbi/configuration.py

@@ -0,0 +1,64 @@
+"""
+Python Module for Pydantic Models and validation
+"""
+import importlib
+from inspect import signature
+from types import ModuleType
+from typing import Any, Dict, Optional
+
+from pydantic import Extra, ValidationError, root_validator, validator
+from pydantic_yaml import YamlModel
+
+
+class ActionConfig(YamlModel, extra=Extra.allow):
+    """
+    ActionConfig is the inner model of `actions` in yaml
+    """
+
+    action: str = "src.jbi.whiteboard_actions.default"
+    enabled: bool = False
+    parameters: dict = {}
+
+    @root_validator
+    def validate_action_config(
+        cls, values
+    ):  # pylint: disable=no-self-argument, no-self-use
+        """Validate action: exists, has init function, and has expected params"""
+        try:
+            action: str = values["action"]  # type: ignore
+            action_parameters: Optional[Dict[str, Any]] = values["parameters"]
+            action_module: ModuleType = importlib.import_module(action)
+            if not action_module:
+                raise TypeError("Module is not found.")
+            if not hasattr(action_module, "init"):
+                raise TypeError("Module is missing `init` method.")
+
+            signature(action_module.init).bind(**action_parameters)  # type: ignore
+        except ImportError as exception:
+            raise ValidationError(f"unknown action `{action}`.") from exception
+        except (TypeError, AttributeError) as exception:
+            raise ValidationError("action is not properly setup.") from exception
+        return values
+
+
+class Actions(YamlModel):
+    """
+    Actions is the overall model from parsing the yaml config `actions` sections
+    """
+
+    actions: Dict[str, ActionConfig]
+
+    @validator("actions")
+    def validate_action_yaml_jbi_naming(
+        cls, actions
+    ):  # pylint: disable=no-self-argument, no-self-use
+        """
+        Validate that the inner actions are named as expected
+        """
+        if not actions:
+            raise ValidationError("no actions configured")
+        for name, action in actions.items():
+            if name != action.parameters["whiteboard_tag"]:
+                raise ValidationError("action name must match whiteboard tag")
+
+        return actions

src/jbi/model.py

@@ -36,7 +36,7 @@ def get_whiteboard_tag(
     whiteboard_tag: Optional[str] = None,
 ):
     """API for viewing whiteboard_tags"""
-    data = configuration.get_yaml_configurations()
+    data = configuration.get_yaml_configurations().dict()["actions"]
     if whiteboard_tag:
         wb_val = data.get(whiteboard_tag)
         if wb_val:
@@ -47,7 +47,7 @@ def get_whiteboard_tag(
 @api_router.get("/actions/")
 def get_actions_by_type(action_type: Optional[str] = None):
     """API for viewing actions"""
-    configured_actions = configuration.get_yaml_configurations()
+    configured_actions = configuration.get_yaml_configurations().dict()["actions"]
     if action_type:
         data = [
             a["action"]
@@ -62,7 +62,7 @@ def get_actions_by_type(action_type: Optional[str] = None):
 @api_router.get("/powered_by_jbi", response_class=HTMLResponse)
 def powered_by_jbi(request: Request, enabled: Optional[bool] = None):
     """API for `Powered By` endpoint"""
-    data = configuration.get_yaml_configurations()
+    data = configuration.get_yaml_configurations().dict()["actions"]
     context = {
         "request": request,
         "title": "Powered by JBI",

src/jbi/router.py

@@ -20,7 +20,7 @@ <h1>{{title}}</h1>
                 </thead>
                 <tbody class="list">
                     {% for key, value in data.items() %}
-                        {% if enable_query is none or value.enabled is enable_query %}
+                        {% if enable_query == none or value.enabled == enable_query %}
                         <tr>
                             <td class="identifier">{{key}}</td>
                             <td class="action">{{value.action}}</td>

src/jbi/services.py renamed to src/jbi/service.py

@@ -8,7 +8,7 @@
 
 
 def test_mock_jbi_files():
-    with pytest.raises(configuration.ProcessError):
+    with pytest.raises(configuration.ConfigError):
         configuration.get_yaml_configurations(
             jbi_config_file="tests/unit/jbi/test-config.yaml"
         )