add receiver

Author: tarekbadrsh

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# VS Code
+.vscode

cli/cdevents/cli/artifact.py

@@ -53,7 +53,7 @@ def packaged(
     data: List[str] = None,
 ):
     print_function_args()
-    artifact_event = ArtifactPackagedEvent(id, name, version, data=data)
+    artifact_event = ArtifactPackagedEvent(id=id, name=name, version=version, data=data)
     cdevents_command = CDeventsCommand()
     cdevents_command.run(artifact_event)
 
@@ -67,6 +67,6 @@ def published(
     data: List[str] = None,
 ):
     print_function_args()
-    artifact_event = ArtifactPublishedEvent(id, name, version, data=data)
+    artifact_event = ArtifactPublishedEvent(id=id, name=name, version=version, data=data)
     cdevents_command = CDeventsCommand()
     cdevents_command.run(artifact_event)

cli/cdevents/cli/cdevents_command.py

@@ -1,7 +1,6 @@
 """Module for cli common command code."""
 import logging
 from abc import ABC
-import requests
 
 from cloudevents.http import CloudEvent
 

core/cdevents/core/artifact.py

@@ -1,25 +1,30 @@
 """artifact"""
 
-from enum import Enum
 from cdevents.core.event import Event
-
-class ArtifactType(Enum):
-    ArtifactPackagedEventV1: str = "cd.artifact.packaged.v1"
-    ArtifactPublishedEventV1: str = "cd.artifact.published.v1"
-
+from cdevents.core.event_type import EventType
 
 class ArtifactEvent(Event):
     """Artifact Event."""
 
-    def __init__(self, artifact_type: ArtifactType, id: str, name: str, version: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type = artifact_type
-        self._id = id
-        self._name = name
-        self._version = version
-        super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=data)
-    
+        self._event_type : EventType = kwargs['artifact_type']
+        if 'data' in kwargs:
+            self._data :dict = kwargs['data']
+        
+        if 'id' in kwargs and 'name' in kwargs and 'version' in kwargs:
+            self._id :str = kwargs['id']
+            self._name :str = kwargs['name']
+            self._version :str = kwargs['version']
+            super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=self._data)
+        
+        elif 'extensions' in kwargs:
+            self._id = kwargs['extensions'].get('artifactid')
+            self._name = kwargs['extensions'].get('artifactname')
+            self._version = kwargs['extensions'].get('artifactversion')
+            super().__init__(event_type=self._event_type.value,  extensions=self.create_extensions(), data=self._data)
+
     def create_extensions(self) -> dict:
         """Create extensions.
         """
@@ -32,18 +37,18 @@ def create_extensions(self) -> dict:
 
 class ArtifactPackagedEvent(ArtifactEvent):
 
-    def __init__(self, id: str, name: str, version: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = ArtifactType.ArtifactPackagedEventV1
+        self._event_type: str = EventType.ArtifactPackagedEventV1
 
-        super().__init__(artifact_type=self._event_type, id=id, name=name, version=version, data=data)
+        super().__init__(artifact_type=self._event_type, **kwargs)
 
 class ArtifactPublishedEvent(ArtifactEvent):
 
-    def __init__(self, id: str, name: str, version: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = ArtifactType.ArtifactPublishedEventV1
+        self._event_type: str = EventType.ArtifactPublishedEventV1
 
-        super().__init__(artifact_type=self._event_type, id=id, name=name, version=version, data=data)
+        super().__init__(artifact_type=self._event_type, **kwargs)

core/cdevents/core/branch.py

@@ -1,25 +1,30 @@
 """branch"""
 
-from enum import Enum
 from cdevents.core.event import Event
-
-class BranchType(Enum):
-    BranchCreatedEventV1: str = "cd.repository.branch.created.v1"
-    BranchDeletedEventV1: str = "cd.repository.branch.deleted.v1"
-
+from cdevents.core.event_type import EventType
 
 class BranchEvent(Event):
     """Branch Event."""
 
-    def __init__(self, branch_type: BranchType, id: str, name: str, repoid: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type = branch_type
-        self._id = id
-        self._name = name
-        self._repoid = repoid
-        super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=data)
-    
+        self._event_type : EventType = kwargs['branch_type']
+      
+        if 'data' in kwargs:
+            self._data :dict = kwargs['data']
+        
+        if 'id' in kwargs and 'name' in kwargs and 'repoid' in kwargs:
+            self._id :str = kwargs['id']
+            self._name :str = kwargs['name']
+            self._repoid :str = kwargs['repoid']
+            super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=self._data)
+        elif 'extensions' in kwargs:
+            self._id = kwargs['extensions'].get('branchid')
+            self._name = kwargs['extensions'].get('branchname')
+            self._repoid = kwargs['extensions'].get('branchrepositoryid')
+            super().__init__(event_type=self._event_type.value,  extensions=self.create_extensions(), data=self._data)
+
     def create_extensions(self) -> dict:
         """Create extensions.
         """
@@ -32,18 +37,18 @@ def create_extensions(self) -> dict:
 
 class BranchCreatedEvent(BranchEvent):
 
-    def __init__(self, id: str, name: str, repoid: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = BranchType.BranchCreatedEventV1
+        self._event_type: str = EventType.BranchCreatedEventV1
 
-        super().__init__(branch_type=self._event_type, id=id, name=name, repoid=repoid, data=data)
+        super().__init__(branch_type=self._event_type, **kwargs)
 
 class BranchDeletedEvent(BranchEvent):
 
-    def __init__(self, id: str, name: str, repoid: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = BranchType.BranchDeletedEventV1
+        self._event_type: str = EventType.BranchDeletedEventV1
 
-        super().__init__(branch_type=self._event_type, id=id, name=name, repoid=repoid, data=data)
+        super().__init__(branch_type=self._event_type, **kwargs)

core/cdevents/core/build.py

@@ -1,26 +1,30 @@
 """build"""
 
-from enum import Enum
-from cdevents.core.event import Event
-
-class BuildType(Enum):
-    BuildStartedEventV1  :str = "cd.build.started.v1"
-    BuildQueuedEventV1   :str = "cd.build.queued.v1"
-    BuildFinishedEventV1 :str = "cd.build.finished.v1"
-
+from cdevents.core.event import Event 
+from cdevents.core.event_type import EventType
 
 class BuildEvent(Event):
     """Build Event."""
 
-    def __init__(self, build_type: BuildType, id: str, name: str, artifact: str, data: dict = {}): 
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type = build_type
-        self._id = id
-        self._name = name
-        self._artifact = artifact
-        super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=data)
-    
+        self._event_type : EventType = kwargs['build_type']
+        if 'data' in kwargs:
+            self._data :dict = kwargs['data']
+
+        if 'id' in kwargs and 'name' in kwargs and 'artifact' in kwargs:
+            self._id :str = kwargs['id']
+            self._name :str = kwargs['name']
+            self._artifact :str = kwargs['artifact']
+            super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=self._data)
+
+        elif 'extensions' in kwargs:
+            self._id = kwargs['extensions'].get('buildid'),
+            self._name = kwargs['extensions'].get('buildname')
+            self._artifact = kwargs['extensions'].get('buildartifactid')
+            super().__init__(event_type=self._event_type.value,  extensions=self.create_extensions(), attrs=kwargs['attrs'], data=self._data)
+
     def create_extensions(self) -> dict:
         """Create extensions.
         """
@@ -32,28 +36,28 @@ def create_extensions(self) -> dict:
         return extensions
 
 class BuildStartedEvent(BuildEvent):
-    
-    def __init__(self, id: str, name: str, artifact: str, data: dict = {}):
+    """Build Started Event."""
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = BuildType.BuildStartedEventV1
+        self._event_type: str = EventType.BuildStartedEventV1
 
-        super().__init__(build_type=self._event_type, id=id, name=name, artifact=artifact, data=data)
-
-class BuildQueuedEvent(BuildEvent):
+        super().__init__(build_type=self._event_type, **kwargs)
 
-    def __init__(self, id: str, name: str, artifact: str, data: dict = {}):
+class BuildQueuedEvent(BuildEvent):
+    """Build Queued Event."""
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = BuildType.BuildQueuedEventV1
+        self._event_type: str = EventType.BuildQueuedEventV1
 
-        super().__init__(build_type=self._event_type, id=id, name=name, artifact=artifact, data=data)
+        super().__init__(build_type=self._event_type, **kwargs)
 
 class BuildFinishedEvent(BuildEvent):
-    
-    def __init__(self, id: str, name: str, artifact: str, data: dict = {}):
+    """Build Finished Event."""
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = BuildType.BuildFinishedEventV1
+        self._event_type: str = EventType.BuildFinishedEventV1
 
-        super().__init__(build_type=self._event_type, id=id, name=name, artifact=artifact, data=data)
+        super().__init__(build_type=self._event_type, **kwargs)

core/cdevents/core/env.py

@@ -1,25 +1,29 @@
 """env"""
 
-from enum import Enum
 from cdevents.core.event import Event
-
-class EnvType(Enum):
-    EnvironmentCreatedEventV1  :str = "cd.environment.created.v1"
-    EnvironmentModifiedEventV1 :str = "cd.environment.modified.v1"
-    EnvironmentDeletedEventV1  :str = "cd.environment.deleted.v1"
-
+from cdevents.core.event_type import EventType
 
 class EnvEvent(Event):
     """Env Event."""
 
-    def __init__(self, env_type: EnvType, id: str, name: str, repo: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type = env_type
-        self._id = id
-        self._name = name
-        self._repo = repo
-        super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=data)
+        self._event_type : EventType = kwargs['env_type']
+        if 'data' in kwargs:
+            self._data :dict = kwargs['data']
+
+        if 'id' in kwargs and 'name' in kwargs and 'repo' in kwargs:
+            self._id :str = kwargs['id']
+            self._name :str = kwargs['name']
+            self._repo :str = kwargs['repo']
+            super().__init__(event_type=self._event_type.value, extensions=self.create_extensions(), data=self._data)
+        
+        elif 'extensions' in kwargs:
+            self._id = kwargs['extensions'].get('envId')
+            self._name = kwargs['extensions'].get('envname')
+            self._repo = kwargs['extensions'].get('envrepourl')
+            super().__init__(event_type=self._event_type.value,  extensions=self.create_extensions(), data=self._data)
 
     def create_extensions(self) -> dict:
         """Create extensions.
@@ -33,29 +37,29 @@ def create_extensions(self) -> dict:
 
 class EnvEventCreatedEvent(EnvEvent):
 
-    def __init__(self, id: str, name: str, repo: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = EnvType.EnvironmentCreatedEventV1
+        self._event_type: str = EventType.EnvironmentCreatedEventV1
 
-        super().__init__(env_type=self._event_type, id=id, name=name, repo=repo, data=data)
+        super().__init__(env_type=self._event_type, **kwargs) 
 
 class EnvEventModifiedEvent(EnvEvent):
 
-    def __init__(self, id: str, name: str, repo: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = EnvType.EnvironmentModifiedEventV1
+        self._event_type: str = EventType.EnvironmentModifiedEventV1
 
-        super().__init__(env_type=self._event_type, id=id, name=name, repo=repo, data=data)
+        super().__init__(env_type=self._event_type, **kwargs)
 
 
 class EnvEventDeletedEvent(EnvEvent):
 
-    def __init__(self, id: str, name: str, repo: str, data: dict = {}):
+    def __init__(self, **kwargs):
         """Initializes class.
         """
-        self._event_type: str = EnvType.EnvironmentDeletedEventV1
+        self._event_type: str = EventType.EnvironmentDeletedEventV1
 
-        super().__init__(env_type=self._event_type, id=id, name=name, repo=repo, data=data)
+        super().__init__(env_type=self._event_type, **kwargs)
 

core/cdevents/core/event.py

@@ -6,29 +6,26 @@
 class Event(CloudEvent):
     """Event."""
 
-    def __init__(self, event_type: str, extensions: dict, data = {}):
+    def __init__(self, event_type: str, extensions: dict, attrs=None, data = {}):
         """Initializes class.
         """
-        self._event_type = event_type
-        self._extensions = extensions
-        self._data = data
+        if attrs:
+            super().__init__(attributes=attrs, data=data)
+        else:
+            self._event_type = event_type
+            self._extensions = extensions
+            self._data = data
 
-        self._attributes = {
-            "type": self._event_type,
-            "source": "cde-cli",
-            "extensions": self._extensions,
-        }
-        super().__init__(self._attributes, dict(self._data))
+            self._attributes = {
+                "type": self._event_type,
+                "source": "cde-cli",
+                "extensions": self._extensions,
+            }
+            super().__init__(self._attributes, dict(self._data))
 
     @abstractmethod
     def create_extensions(self) -> dict:
         """Create extensions.
         """
         extensions = {}
         return extensions
-
-    # @abstractmethod
-    # def event_from_json(json_obj: dict) -> Events:
-    #     """Create event from json.
-    #     """
-    #     pass