Read from configuration

Author: tarekbadrsh

cli/cdevents/cli/artifact.py

@@ -5,23 +5,13 @@
 from typing import List
 
 import click
-import requests
-from cloudevents.http import CloudEvent, to_structured
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args
-
+from cdevents.cli.cdevents_command import CDeventsCommand
 
 # pylint: disable=unused-argument
 def common_artifact_options(function):
     """Decorator for common cli options for artifact."""
-    function = click.option(
-        "--cde_sink",
-        "-c",
-        required=False,
-        type=str,
-        default=lambda: os.environ.get("CDE_SINK", "http://localhost:8080"),
-        help="CDE_SINK",
-    )(function)
     function = click.option(
         "--id",
         "-i",
@@ -58,50 +48,36 @@ def common_artifact_options(function):
 @click.command(help=add_disclaimer_text("Artifact Packaged CloudEvent."))
 @common_artifact_options
 def packaged(
-    cde_sink: str,
     id: str,
     name: str = None,
     version: str = None,
     data: List[str] = None,
 ):
     print_function_args()
-    attributes = {
-        "type": "cd.artifact.packaged.v1",
-        "source": "cde-cli",
-        "extensions": {
-            "artifactid": id,
-            "artifactname": name,
-            "artifactversion": version,
-        },
+    artifact_packaged_eventV1  = "cd.artifact.packaged.v1"
+    extensions = {
+        "artifactid": id,
+        "artifactname": name,
+        "artifactversion": version,
     }
-    event = CloudEvent(attributes, dict(data))
-    headers, body = to_structured(event)
-
-    # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
+    cdevents_command = CDeventsCommand()
+    cdevents_command.run(artifact_packaged_eventV1, extensions, data)
 
 
 @click.command(help=add_disclaimer_text("Artifact Published CloudEvent."))
 @common_artifact_options
 def published(
-    cde_sink: str,
     id: str,
     name: str = None,
     version: str = None,
     data: List[str] = None,
 ):
     print_function_args()
-    attributes = {
-        "type": "cd.artifact.published.v1",
-        "source": "cde-cli",
-        "extensions": {
-            "artifactid": id,
-            "artifactname": name,
-            "artifactversion": version,
-        },
+    artifact_published_eventV1  = "cd.artifact.published.v1"
+    extensions = {
+        "artifactid": id,
+        "artifactname": name,
+        "artifactversion": version,
     }
-    event = CloudEvent(attributes, dict(data))
-    headers, body = to_structured(event)
-
-    # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
+    cdevents_command = CDeventsCommand()
+    cdevents_command.run(artifact_published_eventV1, extensions, data)

cli/cdevents/cli/branch.py

@@ -1,6 +1,7 @@
 """Module for cli branch commands."""
 from __future__ import annotations
 
+import logging
 import os
 from typing import List
 
@@ -10,6 +11,7 @@
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args
 
+_log = logging.getLogger(__name__)
 
 # pylint: disable=unused-argument
 def common_branch_options(function):
@@ -78,8 +80,8 @@ def created(
     headers, body = to_structured(event)
 
     # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
-
+    result = requests.post(cde_sink, headers=headers, data=body)
+    _log.debug(f"Response with state code {result.status_code}")
 
 @click.command(help=add_disclaimer_text("Branch Deleted CloudEvent."))
 @common_branch_options
@@ -104,4 +106,5 @@ def deleted(
     headers, body = to_structured(event)
 
     # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
+    result = requests.post(cde_sink, headers=headers, data=body)
+    _log.debug(f"Response with state code {result.status_code}")

cli/cdevents/cli/build.py

@@ -1,6 +1,7 @@
 """Module for cli build commands."""
 from __future__ import annotations
 
+import logging
 import os
 from typing import List
 
@@ -10,6 +11,8 @@
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args
 
+_log = logging.getLogger(__name__)
+
 
 # pylint: disable=unused-argument
 def common_build_options(function):
@@ -79,7 +82,8 @@ def started(
     headers, body = to_structured(event)
 
     # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
+    result = requests.post(cde_sink, headers=headers, data=body)
+    _log.debug(f"Response with state code {result.status_code}")
 
 
 @click.command(help=add_disclaimer_text("Build Finished CloudEvent."))
@@ -105,7 +109,8 @@ def finished(
     headers, body = to_structured(event)
 
     # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
+    result = requests.post(cde_sink, headers=headers, data=body)
+    _log.debug(f"Response with state code {result.status_code}")
 
 
 @click.command(help=add_disclaimer_text("PipelineRun Queued CloudEvent."))
@@ -131,4 +136,5 @@ def queued(
     headers, body = to_structured(event)
 
     # send and print event
-    requests.post(cde_sink, headers=headers, data=body)
+    result = requests.post(cde_sink, headers=headers, data=body)
+    _log.debug(f"Response with state code {result.status_code}")

cli/cdevents/cli/cdevents_command.py

@@ -0,0 +1,44 @@
+"""Module for cli common command code."""
+import logging
+from abc import ABC
+import requests
+
+from cdevents.cli.configuration_handler import ConfigurationHandler
+from cloudevents.http import CloudEvent, to_structured
+from cdevents.cli.configuration_handler import new_default_configuration_handler
+
+class CDeventsCommand(ABC):
+    """Abstract base class for all Hostlog commands."""
+
+    def __init__(self, config_handler: ConfigurationHandler = None):
+        """Initializes base class.
+
+        Args:
+            config_handler (ConfigurationHandler): the configuration handler.
+        """
+        self._log = logging.getLogger(__name__)
+        self._config_handler = config_handler
+        if config_handler is None:
+            self._config_handler = new_default_configuration_handler()
+
+    def run(self, type, extensions, data):
+        """run command.
+        """
+        attributes = {
+            "type": type,
+            "source": self.config_handler.source.name,
+            "extensions": extensions,
+        }
+        event = CloudEvent(attributes, dict(data))
+        headers, body = to_structured(event)
+        cde_link = self.config_handler.client.host
+
+        # send and print event
+        result = requests.post(cde_link, headers=headers, data=body)
+        self._log.info(f"Response with state code {result.status_code}")
+
+
+    @property
+    def config_handler(self) -> ConfigurationHandler:
+        """Property for configuration handler."""
+        return self._config_handler

cli/cdevents/cli/configuration_handler.py

@@ -0,0 +1,122 @@
+"""Module for configuration provider."""
+from __future__ import annotations
+
+import copy
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Union
+
+from cdevents.cli.configuration_reader import ConfigurationReader
+from cdevents.cli.constants import DEAFULT_CONFIGURATION_FILE
+from cdevents.cli.utils import DictUtils
+
+
+def get_default_configuration_file() -> str:
+    """Returns the default configuration file path."""
+    return DEAFULT_CONFIGURATION_FILE
+
+def new_default_configuration_handler() -> ConfigurationHandler:
+    """Returnes a configuration handler with the default configuration file"""
+    config_handler: ConfigurationHandler = ConfigurationHandler.create_new(
+        get_default_configuration_file()
+    )
+
+    return config_handler
+
+
+def new_configuration_handler_with_override(
+    client_host, source_name
+) -> ConfigurationHandler:
+    """Returnes a configuration handler where args override configuration file."""
+    args_as_config = ConfigurationHandler.create_override_config(
+        client_host=client_host,
+        source_name=source_name
+    )
+
+    config_handler: ConfigurationHandler = ConfigurationHandler.create_new(
+        get_default_configuration_file(), args_as_config
+    )
+
+    return config_handler
+
+
+class ConfigurationHandler:
+    """Class for providing configuration."""
+
+    def __init__(self, configuration: dict):
+        """Initializes the configuration. 
+
+        Args:
+            configuration (dict): The configuration.
+        """
+        self.client = _ClientConfig(**configuration["client"])
+        self.source = _SourceConfig(**configuration["source"])
+
+    @staticmethod
+    def create_override_config(
+        client_host: str = None,
+        source_name: str = None,
+    ) -> dict:
+        """Create a dict that can be used for overriding the default configuration.
+
+        Args:
+            host (str, optional): client host address. Defaults to None.
+            name (str, optional): source name. Defaults to None.
+
+        Returns:
+            dict: the dict to ovverride configuration with.
+        """
+        override_dict: dict = {}
+        if client_host:
+            DictUtils.merge_dicts({"client": {"host": client_host}}, override_dict)
+        if source_name:
+            DictUtils.merge_dicts({"source": {"name": source_name}}, override_dict)
+
+        return override_dict
+
+    @staticmethod
+    def create_new(
+        configuration_files: Union[list[str], str] = None, override_config: dict = None
+    ) -> ConfigurationHandler:
+        """Reads default configurationfile plus any additional configuration files provided.
+
+        Additional configuration files will be merged with the default configfuration file.
+        Ovveride configuration will ovverid any configuration from files.
+        A configuration provider is returned.
+
+        Args:
+            configuration_files (Union[list[str], str], optional): yaml or json configration file. \
+                Defaults to None.
+            override_config (dict):override configuration from cli args. Defaults to None.
+
+        Returns:
+            ConfigurationHandler: the handler.
+        """
+        file_list = []
+        if isinstance(configuration_files, str) or isinstance(configuration_files, Path):
+            file_list.append(configuration_files)
+        elif isinstance(configuration_files, list):
+            file_list.extend(configuration_files)
+
+        reader = ConfigurationReader()
+        for _file in file_list:
+            reader.read_configuration(_file)
+
+        # merge dicts
+        if override_config:
+            _config = copy.deepcopy(override_config)
+            DictUtils.merge_dicts(reader.configuration["configuration"], _config)
+        else:
+            _config = reader.configuration["configuration"]
+
+        return ConfigurationHandler(_config)
+
+
+@dataclass
+class _ClientConfig:
+    host: str
+
+@dataclass
+class _SourceConfig:
+    name: str