add artifact extension

Author: tarekbadrsh

cli/cdevents/cli/cdevents_command.py

@@ -28,19 +28,6 @@ def __init__(self, config_handler: ConfigurationHandler = None):
     def run(self, event: CloudEvent):
         """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}")
-
         e = EventSender(cde_link=self.config_handler.client.host)
         e.send(event)
 

cli/cdevents/cli/utils.py

@@ -115,11 +115,7 @@ def _os_name() -> str:
 
 
 # TODO: Update disclaimer text
-DISCLAIMER_TEXT = """!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
-    N.B. Due to data privacy regulations, all locally generated cdevents data files
-    ,such as hdf5 etc, must be deleted after a logging session is finished.
-    Exluded are log files generated by Vigem on the way to be stored in Oden
-    !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"""
+DISCLAIMER_TEXT = ""
 
 
 def add_disclaimer_text(help_text: str) -> str:

core/cdevents/core/event_sender.py

@@ -1,4 +1,3 @@
-
 """Core events sender."""
 
 import requests

core/cdevents/core/extensions/artifact_extension.py

@@ -0,0 +1,28 @@
+"""ArtifactExtension represents the extension for extension context."""
+
+from cloudevents.http import CloudEvent
+
+ArtifactIdExtension      = "artifactid"
+ArtifactNameExtension    = "artifactname"
+ArtifactVersionExtension = "artifactversion"
+
+
+class ArtifactExtension():
+    """Artifact Extension."""
+    def __init__(self) -> None:
+        pass
+
+    def read_transformer():
+        """Read transformer."""
+        pass
+
+    def write_transformer(event: CloudEvent, extensions: dict) -> CloudEvent:
+        """Write transformer."""
+        if event._attributes["extensions"].get(ArtifactIdExtension):
+            event._attributes["extensions"].set(ArtifactIdExtension, extensions.get(ArtifactIdExtension))
+        if event._attributes["extensions"].get(ArtifactNameExtension):
+            event._attributes["extensions"].set(ArtifactNameExtension, extensions.get(ArtifactNameExtension))
+        if event._attributes["extensions"].get(ArtifactVersionExtension):
+            event._attributes["extensions"].set(ArtifactVersionExtension, extensions.get(ArtifactVersionExtension))
+        return event
+

core/cdevents/core/extensions/service_extension.py

@@ -0,0 +1,28 @@
+"""ServiceExtension represents the extension for extension context."""
+
+from cloudevents.http import CloudEvent
+
+ServiceEnvIdExtension   = "serviceenvid"
+ServiceNameExtension    = "servicename"
+ServiceVersionExtension = "serviceversion"
+
+
+class ServiceExtension():
+    """Service Extension."""
+    def __init__(self) -> None:
+        pass
+
+    def read_transformer():
+        """Read transformer."""
+        pass
+
+    def write_transformer(event: CloudEvent, extensions: dict) -> CloudEvent:
+        """Write transformer."""
+        if event._attributes["extensions"].get(ServiceEnvIdExtension):
+            event._attributes["extensions"].set(ServiceEnvIdExtension, extensions.get(ServiceEnvIdExtension))
+        if event._attributes["extensions"].get(ServiceNameExtension):
+            event._attributes["extensions"].set(ServiceNameExtension, extensions.get(ServiceNameExtension))
+        if event._attributes["extensions"].get(ServiceVersionExtension):
+            event._attributes["extensions"].set(ServiceVersionExtension, extensions.get(ServiceVersionExtension))
+        return event
+