feat(blueprint): add blueprint api

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.blueprint import BlueprintApi
 from novu.api.change import ChangeApi
 from novu.api.environment import EnvironmentApi
 from novu.api.event import EventApi
@@ -19,6 +20,7 @@
 from novu.api.topic import TopicApi
 
 __all__ = [
+    "BlueprintApi",
     "ChangeApi",
     "EnvironmentApi",
     "EventApi",

novu/api/blueprint.py

@@ -0,0 +1,45 @@
+"""This module is used to define the ``BlueprintApi``,
+a python wrapper to interact with ``Blueprint`` in Novu."""
+from typing import Optional
+
+import requests
+
+from novu.api.base import Api
+from novu.constants import BLUEPRINTS_ENDPOINT
+from novu.dto.blueprint import BlueprintDto, GroupedBlueprintDto
+
+
+class BlueprintApi(Api):
+    """This class aims to handle all API methods around blueprints in Novu"""
+
+    def __init__(
+        self,
+        url: Optional[str] = None,
+        api_key: Optional[str] = None,
+        requests_timeout: Optional[int] = None,
+        session: Optional[requests.Session] = None,
+    ) -> None:
+        super().__init__(url=url, api_key=api_key, requests_timeout=requests_timeout, session=session)
+
+        self._blueprint_url = f"{self._url}{BLUEPRINTS_ENDPOINT}"
+
+    def get_by_id(self, template_id: str) -> BlueprintDto:
+        """Get a blueprint by ID
+
+        Args:
+            template_id: The ID of the blueprint
+
+        Returns:
+            The blueprint instance
+        """
+        return BlueprintDto.from_camel_case(self.handle_request("GET", f"{self._blueprint_url}/{template_id}")["data"])
+
+    def get_grouped_by_category(self) -> GroupedBlueprintDto:
+        """Get blueprints grouped by category
+
+        Returns:
+            Grouped blueprints instance
+        """
+        return GroupedBlueprintDto.from_camel_case(
+            self.handle_request("GET", f"{self._blueprint_url}/group-by-category")["data"]
+        )

novu/constants.py

@@ -15,5 +15,6 @@
 SUBSCRIBERS_ENDPOINT = "/v1/subscribers"
 TENANTS_ENDPOINT = "/v1/tenants"
 TOPICS_ENDPOINT = "/v1/topics"
+BLUEPRINTS_ENDPOINT = "/v1/blueprints"
 
 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.blueprint import BlueprintDto, GroupedBlueprintDto
 from novu.dto.change import ChangeDetailDto, ChangeDto, PaginatedChangeDto
 from novu.dto.environment import (
     EnvironmentApiKeyDto,
@@ -56,6 +57,7 @@
 from novu.dto.topic import PaginatedTopicDto, TopicDto, TriggerTopicDto
 
 __all__ = [
+    "BlueprintDto",
     "ChangeDetailDto",
     "ChangeDto",
     "EnvironmentApiKeyDto",
@@ -65,6 +67,7 @@
     "ExecutionDetailDto",
     "FeedDto",
     "FieldFilterPartDto",
+    "GroupedBlueprintDto",
     "IntegrationChannelUsageDto",
     "IntegrationDto",
     "LayoutDto",

novu/dto/blueprint.py

@@ -0,0 +1,97 @@
+"""This module is used to gather all DTO definitions related to the Blueprint resource in Novu"""
+import dataclasses
+from typing import List, Optional
+
+from novu.dto.base import CamelCaseDto, DtoIterableDescriptor
+
+
+@dataclasses.dataclass
+class BlueprintDto(CamelCaseDto["BlueprintDto"]):  # pylint: disable=R0902
+    """Blueprint definition."""
+
+    name: str
+    """Blueprint name."""
+
+    description: str
+    """Blueprint description."""
+
+    _id: Optional[str] = None
+    """Blueprint 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."""
+
+    _notification_group_id: Optional[str] = None
+    """Notification group ID in Novu internal storage system."""
+
+    _creator_id: Optional[str] = None
+    """Creator ID in Novu internal storage system."""
+
+    created_at: Optional[str] = None
+    """Creation date-time of the blueprint."""
+
+    updated_at: Optional[str] = None
+    """Last date-time when the blueprint was updated."""
+
+    deleted: Optional[bool] = None
+    """If the blueprint is deleted."""
+
+    deleted_at: Optional[str] = None
+    """Date-time of the removal of the blueprint."""
+
+    deleted_by: Optional[str] = None
+    """User who asked for the removal of the blueprint."""
+
+    critical: Optional[bool] = None
+    """If the blueprint is critical."""
+
+    _parent_id: Optional[str] = None
+    """Parent ID in Novu internal storage system."""
+
+    active: Optional[bool] = None
+    """Active status of the blueprint."""
+
+    blueprint_id: Optional[str] = None
+    """Blueprint ID in Novu internal storage system."""
+
+    draft: Optional[bool] = None
+    """Draft status of the blueprint."""
+
+    is_blueprint: Optional[bool] = None
+    """If the object is a blueprint."""
+
+    notification_group: Optional[object] = None
+    """Notification group object."""
+
+    preference_settings: Optional[object] = None
+    """Preference settings object."""
+
+    steps: Optional[List[object]] = None
+    """List of steps in the blueprint."""
+
+    tags: Optional[List[str]] = None
+    """List of tags associated with the blueprint."""
+
+    triggers: Optional[List[object]] = None
+    """List of triggers in the blueprint."""
+
+
+@dataclasses.dataclass
+class GroupedBlueprintDto(CamelCaseDto["GroupedBlueprintDto"]):
+    """Grouped blueprint definition."""
+
+    category: str
+    """Blueprint category."""
+
+    general: DtoIterableDescriptor[BlueprintDto] = DtoIterableDescriptor[BlueprintDto](
+        default_factory=list, item_cls=BlueprintDto
+    )
+    """List of general blueprints in the category."""
+
+    popular: DtoIterableDescriptor[BlueprintDto] = DtoIterableDescriptor[BlueprintDto](
+        default_factory=list, item_cls=BlueprintDto
+    )
+    """List of popular blueprints in the category."""

tests/api/test_blueprint.py

@@ -0,0 +1,93 @@
+from unittest import TestCase, mock
+
+from novu.api import BlueprintApi
+from novu.config import NovuConfig
+from novu.dto import BlueprintDto, GroupedBlueprintDto
+from tests.factories import MockResponse
+
+
+class BlueprintApiTests(TestCase):
+    api: BlueprintApi
+    blueprint_json = {
+        "name": "Sample Blueprint",
+        "description": "Sample Blueprint Description",
+        "_id": "63dafeda7779f59258e38450",
+        "_environmentId": "63dafed97779f59258e38445",
+        "_organizationId": "63dafed97779f59258e3843f",
+        "_notificationGroupId": "63dafed97779f59258e3844b",
+        "_creatorId": "63dafed4117f8c850991ec4a",
+        "blueprintId": "63dafeda7779f59258e38450",
+        "createdAt": "2023-06-23T12:00:00.000Z",
+        "updatedAt": "2023-06-23T12:00:00.000Z",
+        "deleted": False,
+        "deletedAt": None,
+        "deletedBy": None,
+    }
+
+    grouped_blueprint_json = {
+        "category": "example_category",
+        "general": [blueprint_json],
+        "popular": [blueprint_json],
+    }
+
+    response_get = {"data": blueprint_json}
+    response_grouped = {"data": grouped_blueprint_json}
+    expected_blueprint_dto = BlueprintDto(
+        name="Sample Blueprint",
+        description="Sample Blueprint Description",
+        _id="63dafeda7779f59258e38450",
+        _environment_id="63dafed97779f59258e38445",
+        _organization_id="63dafed97779f59258e3843f",
+        _notification_group_id="63dafed97779f59258e3844b",
+        _creator_id="63dafed4117f8c850991ec4a",
+        blueprint_id="63dafeda7779f59258e38450",
+        created_at="2023-06-23T12:00:00.000Z",
+        updated_at="2023-06-23T12:00:00.000Z",
+        deleted=False,
+        deleted_at=None,
+        deleted_by=None,
+    )
+    expected_grouped_blueprint_dto = GroupedBlueprintDto(
+        category="example_category",
+        general=[expected_blueprint_dto],
+        popular=[expected_blueprint_dto],
+    )
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        NovuConfig.configure("sample.novu.com", "api-key")
+        cls.api = BlueprintApi()
+
+    @mock.patch("requests.request")
+    def test_get_blueprint_by_id(self, mock_request: mock.MagicMock) -> None:
+        mock_request.return_value = MockResponse(200, self.response_get)
+
+        result = self.api.get_by_id("63dafeda7779f59258e38450")
+        self.assertIsInstance(result, BlueprintDto)
+        self.assertEqual(result, self.expected_blueprint_dto)
+
+        mock_request.assert_called_once_with(
+            method="GET",
+            url="sample.novu.com/v1/blueprints/63dafeda7779f59258e38450",
+            headers={"Authorization": "ApiKey api-key"},
+            json=None,
+            params=None,
+            timeout=5,
+        )
+
+    @mock.patch("requests.request")
+    def test_get_grouped_blueprints(self, mock_request: mock.MagicMock) -> None:
+        mock_request.return_value = MockResponse(200, self.response_grouped)
+
+        result = self.api.get_grouped_by_category()
+        self.assertIsInstance(result, GroupedBlueprintDto)
+        self.assertEqual(result, self.expected_grouped_blueprint_dto)
+
+        mock_request.assert_called_once_with(
+            method="GET",
+            url="sample.novu.com/v1/blueprints/group-by-category",
+            headers={"Authorization": "ApiKey api-key"},
+            json=None,
+            params=None,
+            timeout=5,
+        )