feat: Add Notification Client

docs/api_reference.rst

@@ -22,6 +22,7 @@ API Reference
    api_reference/work_item
    api_reference/test_plan
    api_reference/artifact
+   api_reference/notification
 
 Indices and tables
 ------------------

docs/api_reference/notification.rst

@@ -0,0 +1,14 @@
+.. _api_notification_page:
+
+nisystemlink.clients.notification
+==========================
+
+.. autoclass:: nisystemlink.clients.notification.NotificationClient
+   :exclude-members: __init__
+
+   .. automethod:: __init__
+   .. automethod:: apply_dynamic_notification_strategy
+
+.. automodule:: nisystemlink.clients.notification.models
+   :members:
+   :imported-members:

docs/getting_started.rst

@@ -506,3 +506,29 @@ Create, query and delete test plan templates.
 .. literalinclude:: ../examples/test_plan/test_plan_templates.py
    :language: python
    :linenos:
+
+Notification API
+----------------
+
+Overview
+~~~~~~~~
+
+The :class:`.NotificationClient` class is the primary entry point of the Notification API.
+
+When constructing a :class:`.NotificationClient`, you can pass an
+:class:`.HttpConfiguration` (like one retrieved from the
+:class:`.HttpConfigurationManager`), or let :class:`.NotificationClient` use the
+default connection. The default connection depends on your environment.
+
+With a :class:`.NotificationClient` object, you can:
+
+* Apply dynamic notification strategy using :meth:`~.NotificationClient.apply_dynamic_notification_strategy`
+
+Examples
+~~~~~~~~
+
+Apply a notification strategy
+
+.. literalinclude:: ../examples/notification/notification.py
+   :language: python
+   :linenos:

examples/notification/notification.py

@@ -0,0 +1,119 @@
+import uuid
+from datetime import datetime
+
+from nisystemlink.clients.alarm import AlarmClient
+from nisystemlink.clients.alarm.models._alarm import Alarm, AlarmSeverityLevel
+from nisystemlink.clients.alarm.models._create_or_update_alarm_request import (
+    CreateOrUpdateAlarmRequest,
+    SetAlarmTransition,
+)
+from nisystemlink.clients.core import HttpConfiguration
+from nisystemlink.clients.notification import NotificationClient
+from nisystemlink.clients.notification.models import (
+    DynamicNotificationConfiguration,
+    DynamicNotificationStrategy,
+    DynamicStrategyRequest,
+    SmtpAddressFields,
+    SmtpAddressGroup,
+    SmtpMessageTemplate,
+    SmtpMessageTemplateFields,
+)
+
+# Server configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
+server_configuration: HttpConfiguration | None = None
+
+# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
+# the following lines and provide your server URI and API key.
+# server_configuration = HttpConfiguration(
+#     server_uri="https://yourserver.yourcompany.com",
+#     api_key="",
+# )
+
+
+# Create request for applying strategy
+def create_notification_request_for_alarm(
+    alarm: Alarm,
+    address_group: SmtpAddressGroup,
+    message_template: SmtpMessageTemplate,
+) -> DynamicStrategyRequest:
+    """Creates and returns a dynamic strategy request."""
+    occurred_at = alarm.most_recent_transition_occurred_at
+
+    return DynamicStrategyRequest(
+        message_template_substitution_fields={
+            "alarm_id": alarm.alarm_id,
+            "alarm_condition": alarm.condition,
+            "alarm_description": alarm.description,
+            "alarm_severity": str(alarm.current_severity_level),
+            "alarm_occurred_at": occurred_at.isoformat() if occurred_at else "",
+        },
+        notification_strategy=DynamicNotificationStrategy(
+            notification_configurations=[
+                DynamicNotificationConfiguration(
+                    address_group=address_group,
+                    message_template=message_template,
+                )
+            ]
+        ),
+    )
+
+
+# Create clients for Notification and Alarm services
+notification_client = NotificationClient(configuration=server_configuration)
+alarm_client = AlarmClient(configuration=server_configuration)
+
+# Create a unique alarm ID for this example
+alarm_id = f"example_alarm_{uuid.uuid1().hex}"
+
+# Create an alarm with a SET transition
+create_alarm_request = CreateOrUpdateAlarmRequest(
+    alarm_id=alarm_id,
+    transition=SetAlarmTransition(
+        occurred_at=datetime.now(),
+        severity_level=AlarmSeverityLevel.HIGH,
+        value="85",
+        condition="Greater than 80",
+        short_text="Temperature is high",
+        detail_text="Temperature sensor reading is 85°C (higher than the configured threshold of 80°C)",
+    ),
+    description="Example alarm for notification",
+)
+id = alarm_client.create_or_update_alarm(create_alarm_request)
+print("Alarm created successfully")
+
+# Get the alarm by its instance ID (the unique occurrence identifier)
+retrieved_alarm = alarm_client.get_alarm(instance_id=id)
+
+# Define recipients to notify
+recipients = SmtpAddressFields(toAddresses=["sample1@example.com"])
+
+# Create address group
+address_group = SmtpAddressGroup(
+    display_name="Alarm Notification Recipients",
+    properties={"address group": "Alarm"},
+    fields=recipients,
+)
+
+# Create mail template for alarm creation notification
+alarm_creation_template = SmtpMessageTemplate(
+    display_name="Alarm Creation Template",
+    fields=SmtpMessageTemplateFields(
+        subject_template="Alarm Created: <alarm_id>",
+        body_template="An alarm with ID <alarm_id> has been created.\n"
+        "Condition: <alarm_condition>\n"
+        "Description: <alarm_description>\n"
+        "Current severity: <alarm_severity>\n"
+        "Occurred At: <alarm_occurred_at>",
+    ),
+)
+
+# Send notification for alarm creation
+notification_for_alarm_creation = create_notification_request_for_alarm(
+    alarm=retrieved_alarm,
+    address_group=address_group,
+    message_template=alarm_creation_template,
+)
+notification_client.apply_dynamic_notification_strategy(
+    request=notification_for_alarm_creation
+)
+print("Notification sent for alarm creation")

nisystemlink/clients/notification/__init__.py

@@ -0,0 +1,3 @@
+from ._notification_client import NotificationClient
+
+# flake8: noqa

nisystemlink/clients/notification/_notification_client.py

@@ -0,0 +1,49 @@
+"""Implementation of Notification Client"""
+
+from nisystemlink.clients import core
+from nisystemlink.clients.core._uplink._base_client import BaseClient
+from nisystemlink.clients.core._uplink._methods import post
+from uplink import retry
+
+from . import models
+
+
+@retry(
+    when=retry.when.status(408, 429, 502, 503, 504),
+    stop=retry.stop.after_attempt(1),
+    on_exception=retry.CONNECTION_ERROR,
+)
+class NotificationClient(BaseClient):
+    def __init__(self, configuration: core.HttpConfiguration | None = None):
+        """Initialize an instance.
+
+        Args:
+            configuration: Defines the web server to connect to and information about
+                how to connect. If not provided, the
+                :class:`HttpConfigurationManager <nisystemlink.clients.core.HttpConfigurationManager>`
+                is used to obtain the configuration.
+
+        Raises:
+            ApiException: if unable to communicate with the `/ninotification` service.
+        """
+        if configuration is None:
+            configuration = core.HttpConfigurationManager.get_configuration()
+
+        super().__init__(configuration, base_path="/ninotification/v1/")
+
+    @post("apply-dynamic-strategy")
+    def apply_dynamic_notification_strategy(
+        self, request: models.DynamicStrategyRequest
+    ) -> None:
+        """Applies the notification strategy from the given request.
+
+        Args:
+            request: Request with message template substitution fields and notification strategies.
+
+        Returns:
+            None.
+
+        Raises:
+            ApiException: if unable to communicate with the `/ninotification` service or provided invalid arguments.
+        """
+        ...

nisystemlink/clients/notification/models/__init__.py

@@ -0,0 +1,9 @@
+from ._address_group import AddressGroup
+from ._dynamic_strategy_request import DynamicStrategyRequest
+from ._message_template import MessageTemplate
+from ._dynamic_notification_configuration import DynamicNotificationConfiguration
+from ._dynamic_notification_strategy import DynamicNotificationStrategy
+from ._smtp_address_group import SmtpAddressFields, SmtpAddressGroup
+from ._smtp_message_template import SmtpMessageTemplateFields, SmtpMessageTemplate
+
+# flake8: noqa

nisystemlink/clients/notification/models/_address_group.py

@@ -0,0 +1,13 @@
+from typing import Dict, List
+
+from ._base_notification_metadata import BaseNotificationMetadata
+
+
+class AddressGroup(BaseNotificationMetadata):
+    """Model defining notification recipients for generic service."""
+
+    interpreting_service_name: str
+    """Name of the interpreting service."""
+
+    fields: Dict[str, List[str]]
+    """Address group's fields. Requires at least one valid recipient."""

nisystemlink/clients/notification/models/_base_notification_metadata.py

@@ -0,0 +1,23 @@
+from typing import Dict, List
+
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+
+
+class BaseNotificationMetadata(JsonModel):
+    id: str | None = None
+    """Identifier of the notification metadata"""
+
+    display_name: str | None = None
+    """Display name of the object.
+
+    Example: "name"
+    """
+
+    properties: Dict[str, str] | None = None
+    """Additional properties for the base metadata.
+
+    Example: { "property": "value" }
+    """
+
+    referencing_notification_strategies: List[str] | None = None
+    """List of notification strategies referencing the notification metadata"""

nisystemlink/clients/notification/models/_dynamic_notification_configuration.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+from nisystemlink.clients.notification.models._address_group import AddressGroup
+from nisystemlink.clients.notification.models._message_template import MessageTemplate
+from nisystemlink.clients.notification.models._smtp_address_group import (
+    SmtpAddressGroup,
+)
+from nisystemlink.clients.notification.models._smtp_message_template import (
+    SmtpMessageTemplate,
+)
+from pydantic import Field, model_validator
+
+
+class DynamicNotificationConfiguration(JsonModel):
+    """Model for notification configuration defining address groups and message template for the notification.
+
+    Requires at least one of addressGroupId or addressGroup, and one of messageTemplateId or messageTemplate.
+    """
+
+    address_group_id: str | None = None
+    """ID referencing the associated address group."""
+
+    message_template_id: str | None = None
+    """ID referencing the associated message template."""
+
+    address_group: AddressGroup | SmtpAddressGroup | None = Field(default=None)
+    """Address group defining notification recipients."""
+
+    message_template: MessageTemplate | SmtpMessageTemplate | None = Field(default=None)
+    """Message template defining notification content structure"""
+
+    @model_validator(mode="after")
+    def validate_required_pairs(self) -> DynamicNotificationConfiguration:
+        """Validator to check at least one of address_group_id or address_group, and
+        one of message_template_id or message_template is present.
+        """
+        if self.address_group_id is None and self.address_group is None:
+            raise ValueError(
+                "One of either AddressGroupId or AddressGroup is required."
+            )
+
+        if self.message_template_id is None and self.message_template is None:
+            raise ValueError(
+                "One of either MessageTemplateId or MessageTemplate is required."
+            )
+
+        return self

nisystemlink/clients/notification/models/_dynamic_notification_strategy.py

@@ -0,0 +1,16 @@
+from typing import Annotated, List
+
+
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+from pydantic import Field
+
+from ._dynamic_notification_configuration import DynamicNotificationConfiguration
+
+
+class DynamicNotificationStrategy(JsonModel):
+    """Model for the notification strategy to be applied."""
+
+    notification_configurations: Annotated[
+        List[DynamicNotificationConfiguration], Field(min_length=1)
+    ]
+    """Notification configurations associated with this strategy."""

nisystemlink/clients/notification/models/_dynamic_strategy_request.py

@@ -0,0 +1,18 @@
+from typing import Dict
+
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+
+from ._dynamic_notification_strategy import DynamicNotificationStrategy
+
+
+class DynamicStrategyRequest(JsonModel):
+    """Request model for applying a dynamic notification strategy."""
+
+    message_template_substitution_fields: Dict[str, str] | None = None
+    """Defines the fields used for substituting values in the message template.
+
+    Example: { "replacement": "value" }
+    """
+
+    notification_strategy: DynamicNotificationStrategy
+    """Notification strategy containing configurations for address groups and message templates."""

nisystemlink/clients/notification/models/_message_template.py

@@ -0,0 +1,13 @@
+from typing import Dict
+
+from ._base_notification_metadata import BaseNotificationMetadata
+
+
+class MessageTemplate(BaseNotificationMetadata):
+    """Model defining the notification content structure."""
+
+    interpreting_service_name: str
+    """Name of the interpreting service"""
+
+    fields: Dict[str, str]
+    """Template fields for message."""