feat: Add Client for Alarm API (#173)

Author: RSam-NI

docs/api_reference.rst

@@ -8,6 +8,7 @@ API Reference
    :caption: Table of Contents
 
    api_reference/core
+   api_reference/alarm
    api_reference/tag
    api_reference/product
    api_reference/testmonitor

docs/api_reference/alarm.rst

@@ -0,0 +1,19 @@
+.. _api_alarm_page:
+
+nisystemlink.clients.alarm
+==========================
+
+.. autoclass:: nisystemlink.clients.alarm.AlarmClient
+   :exclude-members: __init__
+
+   .. automethod:: __init__
+   .. automethod:: create_or_update_alarm
+   .. automethod:: get_alarm
+   .. automethod:: delete_alarm
+   .. automethod:: delete_alarms
+   .. automethod:: acknowledge_alarms
+   .. automethod:: query_alarms
+
+.. automodule:: nisystemlink.clients.alarm.models
+   :members:
+   :imported-members:

docs/getting_started.rst

@@ -3,6 +3,55 @@
 Getting Started
 ===============
 
+Alarm API
+---------
+
+Overview
+~~~~~~~~
+
+The :class:`.AlarmClient` class is the primary entry point of the Alarm API.
+
+When constructing an :class:`.AlarmClient`, you can pass an
+:class:`.HttpConfiguration` (like one retrieved from the
+:class:`.HttpConfigurationManager`), or let :class:`.AlarmClient` use the
+default connection. The default connection depends on your environment.
+
+With an :class:`.AlarmClient` object, you can:
+
+* Create and update alarm instances using :meth:`~.AlarmClient.create_or_update_alarm`
+
+  * Alarms have two key identifiers:
+  
+    * ``alarm_id``: A user-defined identifier for the alarm type
+    * ``instance_id``: A server-generated unique identifier for each alarm occurrence
+  
+  * Create alarm transitions (SET, CLEAR) to track alarm state changes
+
+* Query alarms with :meth:`~.AlarmClient.query_alarms`
+
+  * Filter alarms using Dynamic LINQ expressions
+  * Control which transitions are returned (most recent only or all)
+  * Sort and paginate results
+
+* Get a specific alarm by its instance_id using :meth:`~.AlarmClient.get_alarm`
+
+* Acknowledge alarms by its instance_id using :meth:`~.AlarmClient.acknowledge_alarms`
+
+  * Optionally force-clear alarms when acknowledging
+
+* Delete alarms using :meth:`~.AlarmClient.delete_alarm` or 
+  :meth:`~.AlarmClient.delete_alarms`
+
+Examples
+~~~~~~~~
+
+Create, query, acknowledge, and delete alarms
+
+.. literalinclude:: ../examples/alarm/alarm.py
+   :language: python
+   :linenos:
+
+
 Tag API
 -------
 

examples/alarm/alarm.py

@@ -0,0 +1,113 @@
+import uuid
+from datetime import datetime
+
+from nisystemlink.clients.alarm import AlarmClient
+from nisystemlink.clients.alarm.models import (
+    AlarmOrderBy,
+    AlarmSeverityLevel,
+    ClearAlarmTransition,
+    CreateOrUpdateAlarmRequest,
+    QueryAlarmsWithFilterRequest,
+    SetAlarmTransition,
+    TransitionInclusionOption,
+)
+from nisystemlink.clients.core import ApiException, HttpConfiguration
+
+# Setup the server configuration to point to your instance of SystemLink Enterprise
+server_configuration = HttpConfiguration(
+    server_uri="https://yourserver.yourcompany.com",
+    api_key="YourAPIKeyGeneratedFromSystemLink",
+)
+client = AlarmClient(configuration=server_configuration)
+
+# Create a unique alarm ID for this example
+# alarm_id is a user-defined identifier for this alarm
+alarm_id = f"example_alarm_{uuid.uuid1().hex}"
+
+# Create an alarm with a SET transition
+create_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)",
+    ),
+)
+# Returns instance_id - a server-generated unique identifier for this specific alarm occurrence
+id = client.create_or_update_alarm(create_request)
+
+# Get the alarm by its instance ID (the unique occurrence identifier)
+alarm = client.get_alarm(instance_id=id)
+print(f"Retrieved alarm: {alarm.alarm_id}, Condition: {alarm.condition}")
+
+# Update the alarm with a higher severity (same alarm_id, updates the same instance)
+update_request = CreateOrUpdateAlarmRequest(
+    alarm_id=alarm_id,
+    transition=SetAlarmTransition(
+        occurred_at=datetime.now(),
+        severity_level=AlarmSeverityLevel.CRITICAL,
+        value="95",
+        condition="Greater than 90",
+        short_text="Temperature is critical",
+        detail_text="Temperature sensor reading is 95°C (higher than the configured threshold of 90°C)",
+    ),
+)
+client.create_or_update_alarm(update_request)
+
+# Query alarms with a filter (can filter by alarm_id to find all instances)
+# Include all transitions to see the full alarm history
+query_request = QueryAlarmsWithFilterRequest(
+    filter="alarmId=@0",
+    substitutions=[alarm_id],
+    order_by=AlarmOrderBy.UPDATED_AT,
+    order_by_descending=True,
+    transition_inclusion_option=TransitionInclusionOption.ALL,
+    return_count=True,
+)
+query_response = client.query_alarms(query_request)
+
+# Display query results
+print(f"Total alarms found: {query_response.total_count}")
+for alarm in query_response.alarms:
+    print(f"  Alarm ID: {alarm.alarm_id}, Transitions: {len(alarm.transitions)}")
+    for transition in alarm.transitions:
+        print(f"- {transition.transition_type}: {transition.condition}")
+
+# Acknowledge the alarm
+client.acknowledge_alarms(instance_ids=[id])
+
+# Clear the alarm with 409 conflict handling - Method 1: Manual exception handling
+# A 409 Conflict response indicates that the requested transition would not change the alarm's state.
+# This allows stateless applications to simply attempt state transitions without first checking
+# the current state. For example, a monitoring system can repeatedly try to CLEAR an alarm
+# when conditions return to normal, and the API will return 409 if already cleared.
+clear_request = CreateOrUpdateAlarmRequest(
+    alarm_id=alarm_id,
+    transition=ClearAlarmTransition(
+        occurred_at=datetime.now(),
+        condition="Temperature returned to normal",
+    ),
+)
+try:
+    client.create_or_update_alarm(clear_request)
+    print("Alarm cleared successfully")
+except ApiException as e:
+    if e.http_status_code == 409:
+        print("Alarm is already in the requested state (409 Conflict)")
+    else:
+        raise
+
+# Clear the alarm with 409 conflict handling - Method 2: Using ignore_conflict parameter
+# This approach is cleaner for stateless applications that don't care about 409 errors.
+# Returns None if the alarm is already in the requested state.
+result = client.create_or_update_alarm(clear_request, ignore_conflict=True)
+if result is None:
+    print("No state change needed (alarm already in requested state)")
+else:
+    print(f"Alarm cleared successfully: {result}")
+
+# Delete the alarm by its instance ID
+client.delete_alarm(instance_id=id)

nisystemlink/clients/alarm/__init__.py

@@ -0,0 +1,3 @@
+from nisystemlink.clients.alarm._alarm_client import AlarmClient
+
+# flake8: noqa

nisystemlink/clients/alarm/_alarm_client.py

@@ -0,0 +1,191 @@
+"""Implementation of Alarm Client"""
+
+from typing import List, Literal, overload
+
+from nisystemlink.clients import core
+from nisystemlink.clients.core._uplink._base_client import BaseClient
+from nisystemlink.clients.core._uplink._methods import delete, get, post
+from uplink import Field, Path, retry
+
+from . import models
+
+
+@retry(
+    when=retry.when.status(408, 429, 502, 503, 504),
+    stop=retry.stop.after_attempt(5),
+    on_exception=retry.CONNECTION_ERROR,
+)
+class AlarmClient(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 `/nialarm` Service.
+        """
+        if configuration is None:
+            configuration = core.HttpConfigurationManager.get_configuration()
+
+        super().__init__(configuration, base_path="/nialarm/v1/")
+
+    @post(
+        "acknowledge-instances-by-instance-id",
+        args=[Field("instanceIds"), Field("forceClear")],
+    )
+    def acknowledge_alarms(
+        self, instance_ids: List[str], *, force_clear: bool = False
+    ) -> models.AcknowledgeAlarmsResponse:
+        """Acknowledges one or more alarm instances by their instance IDs.
+
+        Args:
+            instance_ids: List of instance IDs (unique occurrence identifiers) of the alarms to acknowledge.
+                 These are the server-generated IDs returned when creating/updating alarms,
+                 not the user-defined alarm_id.
+            force_clear: Whether or not the affected alarms should have their clear field set to true.
+                         Defaults to False.
+
+        Returns:
+            A response containing the instance IDs that were successfully acknowledged,
+            the instance IDs that failed, and error details for failures.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nialarm` Service or provided invalid arguments.
+        """
+        ...
+
+    @overload
+    def create_or_update_alarm(  # noqa: E704
+        self,
+        request: models.CreateOrUpdateAlarmRequest,
+        *,
+        ignore_conflict: Literal[False] = False,
+    ) -> str: ...
+
+    @overload
+    def create_or_update_alarm(  # noqa: E704
+        self,
+        request: models.CreateOrUpdateAlarmRequest,
+        *,
+        ignore_conflict: Literal[True],
+    ) -> str | None: ...
+
+    def create_or_update_alarm(
+        self,
+        request: models.CreateOrUpdateAlarmRequest,
+        *,
+        ignore_conflict: bool = False,
+    ) -> str | None:
+        """Creates or updates an instance, or occurrence, of an alarm.
+
+        Creates or updates an alarm based on the requested transition and the state
+        of the current active alarm with the given alarm_id (specified in the request).
+        Multiple calls with the same alarm_id will update the same alarm instance.
+
+        Args:
+            request: The request containing alarm_id (user-defined identifier),
+                    transition details, and other alarm properties.
+            ignore_conflict: If True, 409 Conflict errors will be ignored and None will be returned.
+                           If False (default), 409 errors will raise an ApiException.
+                           Setting this to True is useful for stateless applications that want to
+                           attempt state transitions without checking the current alarm state first.
+
+        Returns:
+            The instance_id (unique occurrence identifier) of the created or modified alarm.
+            Use this ID for operations like get_alarm(), delete_alarm(), or acknowledge.
+            Returns None if ignore_conflict is True and a 409 Conflict occurs.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nialarm` Service or provided invalid arguments.
+                A 409 Conflict error occurs when the request does not represent a valid transition
+                for an existing alarm, such as attempting to clear an alarm which is already clear,
+                or attempting to set an alarm which is already set at the given severity level.
+                This error can be suppressed by setting ignore_conflict=True.
+        """
+        try:
+            return self._create_or_update_alarm(request)
+        except core.ApiException as e:
+            if ignore_conflict and e.http_status_code == 409:
+                return None
+            raise
+
+    @post("instances", return_key="instanceId")
+    def _create_or_update_alarm(
+        self, request: models.CreateOrUpdateAlarmRequest
+    ) -> str:
+        """Internal implementation of create_or_update_alarm."""
+        ...
+
+    @get("instances/{instance_id}", args=[Path("instance_id")])
+    def get_alarm(self, instance_id: str) -> models.Alarm:
+        """Gets an alarm by its instance_id.
+
+        Args:
+            instance_id: The unique instance ID (occurrence identifier) of the alarm to retrieve.
+                This is the server-generated ID returned from create_or_update_alarm(),
+                not the user-defined alarm_id.
+
+        Returns:
+            The alarm with the specified instance_id.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nialarm` Service or provided invalid arguments.
+        """
+        ...
+
+    @delete("instances/{instance_id}", args=[Path("instance_id")])
+    def delete_alarm(self, instance_id: str) -> None:
+        """Deletes an alarm by its instance_id.
+
+        Args:
+            instance_id: The unique instance ID (occurrence identifier) of the alarm to delete.
+                This is the server-generated ID returned from create_or_update_alarm(),
+                not the user-defined alarm_id.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nialarm` Service or provided invalid arguments.
+        """
+        ...
+
+    @post("delete-instances-by-instance-id", args=[Field("instanceIds")])
+    def delete_alarms(self, instance_ids: List[str]) -> models.DeleteAlarmsResponse:
+        """Deletes multiple alarm instances by their instance IDs.
+
+        Args:
+            instance_ids: List of instance IDs (unique occurrence identifiers) of the alarms to delete.
+                 These are the server-generated IDs returned when creating/updating alarms,
+                 not the user-defined alarm_id.
+
+        Returns:
+            A response containing lists of successfully deleted and failed instance IDs,
+            along with error information for failures.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nialarm` Service or provided invalid arguments.
+        """
+        ...
+
+    @post("query-instances-with-filter")
+    def query_alarms(
+        self, request: models.QueryAlarmsWithFilterRequest
+    ) -> models.QueryAlarmsWithFilterResponse:
+        """Queries for instances, or occurrences, of alarms using Dynamic LINQ.
+
+        Specifying an empty JSON object in the request body will result in all alarms being returned.
+
+        Args:
+            request: The request containing filter information and query options.
+
+        Returns:
+            A response containing the list of alarms that match the query, along with
+            optional total count and continuation token for pagination.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nialarm` Service or provided invalid arguments.
+        """
+        ...