feat(changes): add Change DTO definitions and wrappers

LICENSE

@@ -1,6 +1,6 @@
 BSD 2-Clause License
 
-Copyright (c) 2023, Oscar MARIE--TAILLEFER
+Copyright (c) 2023, SpikeeLabs
 
 Redistribution and use in source and binary forms, with or without
 modification, are permitted provided that the following conditions are met:

novu/api/__init__.py

@@ -2,6 +2,7 @@
 
 In this SDK, we choose to split the Novu API by business resource to simplify its complexity.
 """
+from novu.api.change import ChangeApi
 from novu.api.event import EventApi
 from novu.api.integration import IntegrationApi
 from novu.api.layout import LayoutApi
@@ -10,6 +11,7 @@
 from novu.api.topic import TopicApi
 
 __all__ = [
+    "ChangeApi",
     "EventApi",
     "IntegrationApi",
     "LayoutApi",

novu/api/change.py

@@ -0,0 +1,68 @@
+"""
+This module is used to define the ``ChangeApi``, a python wrapper to interact with ``Changes`` in Novu.
+"""
+from typing import Dict, Generator, List, Optional, Union
+
+from novu.api.base import Api
+from novu.constants import CHANGES_ENDPOINT
+from novu.dto.change import ChangeDto, PaginatedChangeDto
+
+
+class ChangeApi(Api):
+    """This class aims to handle all API methods around changes in API"""
+
+    def __init__(self, url: Optional[str] = None, api_key: Optional[str] = None) -> None:
+        super().__init__(url, api_key)
+
+        self._change_url = f"{self._url}{CHANGES_ENDPOINT}"
+
+    def list(self, page: Optional[int] = None, limit: Optional[int] = None) -> PaginatedChangeDto:
+        """List existing changes
+
+        Args:
+            page: Page to retrieve. Defaults to None.
+            limit: Size of the page to retrieve. Defaults to None.
+
+        Returns:
+            Paginated list of change
+        """
+        payload: Dict[str, Union[int, str]] = {}
+        if page:
+            payload["page"] = page
+        if limit:
+            payload["limit"] = limit
+
+        return PaginatedChangeDto.from_camel_case(self.handle_request("GET", f"{self._change_url}", payload=payload))
+
+    def count(self) -> int:
+        """Get the number of changes
+
+        Returns:
+            Number of changes
+        """
+        return self.handle_request("GET", f"{self._change_url}/count")["data"]
+
+    def apply(self, change_id: str) -> PaginatedChangeDto:
+        """Apply one change using its ID
+
+        Args:
+            change_ids: A change ID to apply.
+
+        Returns:
+            Paginated changes
+        """
+        return PaginatedChangeDto.from_camel_case(self.handle_request("POST", f"{self._change_url}/{change_id}/apply"))
+
+    def bulk_apply(self, change_ids: List[str]) -> Generator[ChangeDto, None, None]:
+        """Apply a list of changes using their IDs
+
+        Args:
+            change_ids: The list of IDs to apply
+
+        Yields:
+            Parsed Change from results list
+        """
+        results = self.handle_request("POST", f"{self._change_url}/bulk/apply", {"changeIds": change_ids})["data"]
+        for result in results:
+            for sub_result in result:
+                yield ChangeDto.from_camel_case(sub_result)

novu/constants.py

@@ -6,5 +6,6 @@
 NOTIFICATION_GROUPS_ENDPOINT = "/v1/notification-groups"
 SUBSCRIBERS_ENDPOINT = "/v1/subscribers"
 TOPICS_ENDPOINT = "/v1/topics"
+CHANGES_ENDPOINT = "/v1/changes"
 
 DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f%z"

novu/dto/__init__.py

@@ -4,6 +4,7 @@
 for developer purpose (instead of getting raw dict without any hint about what is in it).
 """
 
+from novu.dto.change import ChangeDetailDto, ChangeDto, PaginatedChangeDto
 from novu.dto.event import EventDto
 from novu.dto.integration import IntegrationChannelUsageDto, IntegrationDto
 from novu.dto.layout import LayoutDto, LayoutVariableDto, PaginatedLayoutDto
@@ -22,12 +23,15 @@
 from novu.dto.topic import PaginatedTopicDto, TopicDto, TriggerTopicDto
 
 __all__ = [
+    "ChangeDetailDto",
+    "ChangeDto",
     "EventDto",
     "IntegrationChannelUsageDto",
     "IntegrationDto",
     "LayoutDto",
     "LayoutVariableDto",
     "NotificationGroupDto",
+    "PaginatedChangeDto",
     "PaginatedLayoutDto",
     "PaginatedNotificationGroupDto",
     "PaginatedSubscriberDto",

novu/dto/change.py

@@ -0,0 +1,67 @@
+"""This module is used to gather all DTO definitions related to the Change resource in Novu"""
+import dataclasses
+from typing import Any, List, Optional
+
+from novu.dto.base import CamelCaseDto, DtoIterableDescriptor
+from novu.enums import ChangeKind
+
+
+@dataclasses.dataclass
+class ChangeDetailDto(CamelCaseDto["ChangeDetailDto"]):
+    """Definition of a change detail"""
+
+    op: Optional[str] = None  # pylint: disable=C0103
+    """The operation code of the change"""
+
+    path: Optional[List[str]] = None
+    """The path of the modification"""
+
+    val: Optional[Any] = None
+    """Value of the change, struct depend of the context"""
+
+
+@dataclasses.dataclass
+class ChangeDto(CamelCaseDto["ChangeDto"]):  # pylint: disable=R0902
+    """Definition of a change"""
+
+    _id: Optional[str] = None
+    """Change ID in Novu internal storage system"""
+
+    _environment_id: Optional[str] = None
+    """Environment ID in Novu internal storage system"""
+
+    _organization_id: Optional[str] = None
+    """Organization ID in Novu internal storage system"""
+
+    _entity_id: Optional[str] = None
+    """Entity ID in Novu internal storage system"""
+
+    _parent_id: Optional[str] = None
+    """Parent ID in Novu internal storage system"""
+
+    enabled: Optional[bool] = None
+    """If the change was enabled."""
+
+    type: Optional[ChangeKind] = None
+    """Kind of change applied."""
+
+    change: DtoIterableDescriptor[ChangeDetailDto] = DtoIterableDescriptor[ChangeDetailDto](
+        default_factory=list, item_cls=ChangeDetailDto
+    )
+    """Details about a change"""
+
+    created_at: Optional[str] = None
+    """Date-time of the integration initial configuration"""
+
+    updated_at: Optional[str] = None
+    """Date-time of the last update of the integration"""
+
+
+@dataclasses.dataclass
+class PaginatedChangeDto(CamelCaseDto["PaginatedChangeDto"]):
+    """Definition of paginated changes"""
+
+    page: int = 0
+    total_count: int = 0
+    page_size: int = 0
+    data: DtoIterableDescriptor[ChangeDto] = DtoIterableDescriptor[ChangeDto](default_factory=list, item_cls=ChangeDto)

novu/enums/__init__.py

@@ -1,5 +1,6 @@
 """This module is used to gather all enumerations defined by Novu in Python format to be reused by developers."""
 
+from novu.enums.change import ChangeKind
 from novu.enums.channel import Channel
 from novu.enums.event import EventStatus
 from novu.enums.provider import (
@@ -14,6 +15,7 @@
 from novu.enums.template import TemplateVariableTypeEnum
 
 __all__ = [
+    "ChangeKind",
     "Channel",
     "EventStatus",
     "ChatProviderIdEnum",

novu/enums/change.py

@@ -0,0 +1,12 @@
+"""This module is used to gather enumerations related to the Change resource in Novu"""
+from enum import Enum
+
+
+class ChangeKind(Enum):
+    """This enumeration define all kinds of change in Novu"""
+
+    FEED = "Feed"
+    MESSAGE_TEMPLATE = "MessageTemplate"
+    LAYOUT = "Layout"
+    NOTIFICATION_TEMPLATE = "NotificationTemplate"
+    NOTIFICATION_GROUP = "NotificationGroup"