add service obj

Author: tarekbadrsh

cli/cdevents/cli/artifact.py

@@ -1,6 +1,5 @@
 """Module for cli artifact commands."""
 from __future__ import annotations
-
 from typing import List
 import click
 

cli/cdevents/cli/branch.py

@@ -1,8 +1,6 @@
 """Module for cli branch commands."""
 from __future__ import annotations
-
 from typing import List
-
 import click
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args

cli/cdevents/cli/build.py

@@ -1,9 +1,6 @@
 """Module for cli build commands."""
 from __future__ import annotations
-
-import os
 from typing import List
-
 import click
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args

cli/cdevents/cli/repository.py

@@ -54,9 +54,9 @@ def created(
 ):
     print_function_args()
     repository = Repository(repository_type=RepositoryType.RepositoryCreatedEventV1, id=id, name=name, url=url)
-    pipelinerun_event = repository.create_event(data)
+    repository_event = repository.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(pipelinerun_event)
+    cdevents_command.run(repository_event)
 
 
 @click.command(help=add_disclaimer_text("Repository Modified CloudEvent."))
@@ -69,9 +69,9 @@ def modified(
 ):
     print_function_args()
     repository = Repository(repository_type=RepositoryType.RepositoryModifiedEventV1, id=id, name=name, url=url)
-    pipelinerun_event = repository.create_event(data)
+    repository_event = repository.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(pipelinerun_event)
+    cdevents_command.run(repository_event)
 
 
 @click.command(help=add_disclaimer_text("Repository Deleted CloudEvent."))
@@ -84,7 +84,7 @@ def deleted(
 ):
     print_function_args()
     repository = Repository(repository_type=RepositoryType.RepositoryDeletedEventV1, id=id, name=name, url=url)
-    pipelinerun_event = repository.create_event(data)
+    repository_event = repository.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(pipelinerun_event)
+    cdevents_command.run(repository_event)
 

cli/cdevents/cli/service.py

@@ -1,16 +1,12 @@
 """Module for cli service commands."""
 from __future__ import annotations
-
-import os
 from typing import List
-
 import click
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args
 from cdevents.cli.cdevents_command import CDeventsCommand
 
-from cdevents.core.events import Events
-from cdevents.core import event_type
+from cdevents.core.service import Service, ServiceType
 
 # pylint: disable=unused-argument
 def common_service_options(function):
@@ -57,11 +53,10 @@ def deployed(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_service_event(event_type.ServiceDeployedEventV1, envid, name, version, data)
+    service = Service(service_type=ServiceType.ServiceDeployedEventV1, envid=envid, name=name, version=version)
+    service_event = service.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
-
+    cdevents_command.run(service_event)
 
 @click.command(help=add_disclaimer_text("Service Upgraded CloudEvent."))
 @common_service_options
@@ -72,10 +67,10 @@ def upgraded(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_service_event(event_type.ServiceUpgradedEventV1, envid, name, version, data)
+    service = Service(service_type=ServiceType.ServiceUpgradedEventV1, envid=envid, name=name, version=version)
+    service_event = service.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(service_event)
 
 
 @click.command(help=add_disclaimer_text("Service Removed CloudEvent."))
@@ -87,10 +82,10 @@ def removed(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_service_event(event_type.ServiceRemovedEventV1, envid, name, version, data)
+    service = Service(service_type=ServiceType.ServiceRemovedEventV1, envid=envid, name=name, version=version)
+    service_event = service.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(service_event)
 
 
 @click.command(help=add_disclaimer_text("Service Rolledback CloudEvent."))
@@ -102,7 +97,7 @@ def rolledback(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_service_event(event_type.ServiceRolledbackEventV1, envid, name, version, data)
+    service = Service(service_type=ServiceType.ServiceRolledbackEventV1, envid=envid, name=name, version=version)
+    service_event = service.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(service_event)

core/cdevents/core/event_type.py

@@ -9,17 +9,6 @@
 ChangeMergedEventV1    :str = "cd.repository.change.merged.v1"
 ChangeAbandonedEventV1 :str = "cd.repository.change.abandoned.v1"
 
-# Repository events
-RepositoryCreatedEventV1  :str = "cd.repository.created.v1"
-RepositoryModifiedEventV1 :str = "cd.repository.modified.v1"
-RepositoryDeletedEventV1  :str = "cd.repository.deleted.v1"
-
-# Service Events
-ServiceDeployedEventV1   :str = "cd.service.deployed.v1"
-ServiceUpgradedEventV1   :str = "cd.service.upgraded.v1"
-ServiceRolledbackEventV1 :str = "cd.service.rolledback.v1"
-ServiceRemovedEventV1    :str = "cd.service.removed.v1"
-
 # TaskRun events
 TaskRunStartedEventV1  :str = "cd.taskrun.started.v1"
 TaskRunFinishedEventV1 :str = "cd.taskrun.finished.v1"

core/cdevents/core/events.py

@@ -23,22 +23,6 @@ def create_event(self, event_type: str, extensions:dict, data = {}) -> CloudEven
         return event
 
 
-
-    def create_service_event(self, event_type: str , envid: str, name: str, version: str, data = {}) -> CloudEvent:
-        """Create service event.
-        """
-            
-        extensions = {
-            "serviceenvid": envid,
-            "servicename": name,
-            "serviceversion": version,
-        }
-
-        event = self.create_event(event_type, extensions, data)
-
-        return event
-
-
     def create_taskrun_event(self, event_type: str, id: str, name: str, pipelineid: str, data = {}) -> CloudEvent:
         """Create taskrun event.
         """

core/cdevents/core/service.py

@@ -0,0 +1,39 @@
+"""service"""
+
+from enum import Enum
+from cdevents.core.events import Events
+
+class ServiceType(Enum):
+    ServiceDeployedEventV1   :str = "cd.service.deployed.v1"
+    ServiceUpgradedEventV1   :str = "cd.service.upgraded.v1"
+    ServiceRolledbackEventV1 :str = "cd.service.rolledback.v1"
+    ServiceRemovedEventV1    :str = "cd.service.removed.v1"
+
+
+class Service(Events):
+    """Service."""
+
+    def __init__(self, service_type: ServiceType, envid: str, name: str, version: str):
+        """Initializes class.
+        """
+        self._event_type = service_type
+        self._envid = envid
+        self._name = name
+        self._version = version
+        
+    def create_extensions(self):
+        """Create extensions.
+        """
+        extensions = {
+            "serviceenvid": self._envid,
+            "servicename": self._name,
+            "serviceversion": self._version,
+        }
+        return extensions
+    
+    def create_event(self, data: dict={}):
+        """Create Service event.
+        """
+        extensions = self.create_extensions()
+        event = super().create_event(event_type=self._event_type.value, extensions=extensions, data=data)
+        return event