usage alerts (account)

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.25.0 (August 19th, 2025)
+
+ENHANCEMENTS:
+* Add Usage Alerts (account): create/get/patch/delete/list; client-side validation; examples. Now accessible via alerts().usage.
+
 ## 0.24.0 (March 20th, 2025)
 
 ENHANCEMENTS:

examples/usage_alerts.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+#
+# Example of using the Usage Alerts API
+#
+
+import os
+import json
+from ns1 import NS1
+from ns1.config import Config
+
+# Create NS1 client
+config = {
+    "endpoint": "https://api.nsone.net",
+    "default_key": "test1",
+    "keys": {
+        "test1": {
+            "key": os.environ.get("NS1_APIKEY", "test1"),
+            "desc": "test key",
+        }
+    },
+}
+
+# Create a config from dictionary and create the client
+c = Config()
+c.loadFromDict(config)
+client = NS1(config=c)
+
+# If no real API key is set, we'll get appropriate errors
+# This is just an example to show the usage pattern
+if not os.environ.get("NS1_APIKEY"):
+    print("Using a mock endpoint - for real usage, set the NS1_APIKEY environment variable")
+
+
+# Usage Alerts API Examples
+def usage_alerts_example():
+    print("\n=== Usage Alerts Examples ===\n")
+
+    # List all usage alerts
+    print("Listing usage alerts:")
+    try:
+        alerts = client.alerts().usage.list(limit=10)
+        print(f"Total alerts: {alerts.get('total_results', 0)}")
+        for i, alert in enumerate(alerts.get("results", [])):
+            print(f"  {i+1}. {alert.get('name')} (id: {alert.get('id')})")
+    except Exception as e:
+        print(f"Error listing alerts: {e}")
+
+    # Create a usage alert
+    print("\nCreating a usage alert:")
+    try:
+        alert = client.alerts().usage.create(
+            name="Example query usage alert",
+            subtype="query_usage",
+            alert_at_percent=85,
+            notifier_list_ids=[],
+            zone_names=[],
+        )
+        alert_id = alert["id"]
+        print(f"Created alert: {alert['name']} (id: {alert_id})")
+        print(f"Alert details: {json.dumps(alert, indent=2)}")
+    except Exception as e:
+        print(f"Error creating alert: {e}")
+        return
+
+    # Update the alert
+    print("\nUpdating the alert threshold to 90%:")
+    try:
+        updated = client.alerts().usage.patch(alert_id, alert_at_percent=90)
+        print(f"Updated alert: {updated['name']}")
+        print(f"New threshold: {updated['data']['alert_at_percent']}%")
+    except Exception as e:
+        print(f"Error updating alert: {e}")
+
+    # Get alert details
+    print("\nGetting alert details:")
+    try:
+        details = client.alerts().usage.get(alert_id)
+        print(f"Alert details: {json.dumps(details, indent=2)}")
+    except Exception as e:
+        print(f"Error getting alert: {e}")
+
+    # Delete the alert
+    print("\nDeleting the alert:")
+    try:
+        client.alerts().usage.delete(alert_id)
+        print(f"Alert {alert_id} deleted successfully")
+    except Exception as e:
+        print(f"Error deleting alert: {e}")
+
+
+# Test validation failures
+def test_validation():
+    print("\n=== Validation Tests ===\n")
+
+    # Test invalid subtype
+    print("Testing invalid subtype:")
+    try:
+        client.alerts().usage.create(
+            name="Test alert", subtype="invalid_subtype", alert_at_percent=85
+        )
+    except ValueError as e:
+        print(f"Validation error (expected): {e}")
+
+    # Test threshold too low
+    print("\nTesting threshold too low (0):")
+    try:
+        client.alerts().usage.create(
+            name="Test alert", subtype="query_usage", alert_at_percent=0
+        )
+    except ValueError as e:
+        print(f"Validation error (expected): {e}")
+
+    # Test threshold too high
+    print("\nTesting threshold too high (101):")
+    try:
+        client.alerts().usage.create(
+            name="Test alert", subtype="query_usage", alert_at_percent=101
+        )
+    except ValueError as e:
+        print(f"Validation error (expected): {e}")
+
+
+if __name__ == "__main__":
+    print("Usage Alerts API Examples")
+    print("-" * 30)
+    print(
+        "Note: To run against the actual API, set the NS1_APIKEY environment variable"
+    )
+    print("Otherwise, this will run against a mock API endpoint")
+
+    # Run examples
+    usage_alerts_example()
+    test_validation()

ns1/__init__.py

@@ -5,7 +5,7 @@
 #
 from .config import Config
 
-version = "0.24.0"
+version = "0.25.0"
 
 
 class NS1:
@@ -242,6 +242,7 @@ def alerts(self):
 
         return ns1.rest.alerts.Alerts(self.config)
 
+
     def billing_usage(self):
         """
         Return a new raw REST interface to BillingUsage resources

ns1/alerting/__init__.py

@@ -0,0 +1,3 @@
+from .usage_alerts import UsageAlertsAPI, USAGE_SUBTYPES
+
+__all__ = ["UsageAlertsAPI", "USAGE_SUBTYPES"]

ns1/alerting/usage_alerts.py

@@ -0,0 +1,101 @@
+from typing import List, Optional, Dict, Any
+
+USAGE_SUBTYPES = {
+    "query_usage",
+    "record_usage",
+    "china_query_usage",
+    "rum_decision_usage",
+    "filter_chain_usage",
+    "monitor_usage",
+}
+
+
+def _validate(name: str, subtype: str, alert_at_percent: int) -> None:
+    if not name:
+        raise ValueError("name required")
+    if subtype not in USAGE_SUBTYPES:
+        raise ValueError("invalid subtype")
+    if not isinstance(alert_at_percent, int) or not (
+        1 <= alert_at_percent <= 100
+    ):
+        raise ValueError("data.alert_at_percent must be int in 1..100")
+
+
+class UsageAlertsAPI:
+    """
+    Account-scoped usage alerts. Triggers when usage ≥ alert_at_percent.
+
+    Server rules: 
+    - Always type='account' 
+    - data.alert_at_percent must be in 1..100
+    - PATCH must not include type/subtype
+    - zone_names/notifier_list_ids may be empty ([])
+    - Server ignores datafeed notifiers for usage alerts
+    """
+
+    def __init__(self, client) -> None:
+        self._c = client
+
+    def create(
+        self,
+        *,
+        name: str,
+        subtype: str,
+        alert_at_percent: int,
+        notifier_list_ids: Optional[List[str]] = None,
+        zone_names: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        _validate(name, subtype, alert_at_percent)
+        body = {
+            "name": name,
+            "type": "account",
+            "subtype": subtype,
+            "data": {"alert_at_percent": int(alert_at_percent)},
+            "notifier_list_ids": notifier_list_ids or [],
+            "zone_names": zone_names or [],
+        }
+        return self._c._post("/alerting/v1/alerts", json=body)
+
+    def get(self, alert_id: str) -> Dict[str, Any]:
+        return self._c._get(f"/alerting/v1/alerts/{alert_id}")
+
+    def patch(
+        self,
+        alert_id: str,
+        *,
+        name: Optional[str] = None,
+        alert_at_percent: Optional[int] = None,
+        notifier_list_ids: Optional[List[str]] = None,
+        zone_names: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        patch: Dict[str, Any] = {}
+        if name is not None:
+            patch["name"] = name
+        if alert_at_percent is not None:
+            if not isinstance(alert_at_percent, int) or not (
+                1 <= alert_at_percent <= 100
+            ):
+                raise ValueError("data.alert_at_percent must be int in 1..100")
+            patch["data"] = {"alert_at_percent": int(alert_at_percent)}
+        if notifier_list_ids is not None:
+            patch["notifier_list_ids"] = notifier_list_ids
+        if zone_names is not None:
+            patch["zone_names"] = zone_names
+        return self._c._patch(f"/alerting/v1/alerts/{alert_id}", json=patch)
+
+    def delete(self, alert_id: str) -> None:
+        self._c._delete(f"/alerting/v1/alerts/{alert_id}")
+
+    def list(
+        self,
+        *,
+        limit: int = 50,
+        next: Optional[str] = None,
+        order_descending: bool = False,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"limit": limit}
+        if next:
+            params["next"] = next
+        if order_descending:
+            params["order_descending"] = "true"
+        return self._c._get("/alerting/v1/alerts", params=params)

ns1/rest/alerts.py

@@ -4,6 +4,7 @@
 # License under The MIT License (MIT). See LICENSE in project root.
 #
 from . import resource
+from ns1.alerting import UsageAlertsAPI
 
 
 class Alerts(resource.BaseResource):
@@ -15,6 +16,62 @@ class Alerts(resource.BaseResource):
         "record_ids",
         "zone_names",
     ]
+
+    # Forward HTTP methods needed by UsageAlertsAPI
+    def _get(self, path, params=None):
+        """Forward GET requests to make_request"""
+        # Fix path to start with /alerting/v1/ if needed
+        if path.startswith('/'):
+            path = path[1:]  # Remove leading slash
+        if not path.startswith("alerting/v1/"):
+            # Alerting endpoints should have this prefix
+            path = f"{self.ROOT}/{path.split('/')[-1]}"
+        return self._make_request("GET", path, params=params)
+
+    def _post(self, path, json=None):
+        """Forward POST requests to make_request"""
+        if path.startswith('/'):
+            path = path[1:]  # Remove leading slash
+        if not path.startswith("alerting/v1/"):
+            path = f"{self.ROOT}"
+        return self._make_request("POST", path, body=json)
+
+    def _patch(self, path, json=None):
+        """Forward PATCH requests to make_request"""
+        if path.startswith('/'):
+            path = path[1:]  # Remove leading slash
+        if not path.startswith("alerting/v1/"):
+            parts = path.split('/')
+            path = f"{self.ROOT}/{parts[-1]}"
+        return self._make_request("PATCH", path, body=json)
+
+    def _delete(self, path):
+        """Forward DELETE requests to make_request"""
+        if path.startswith('/'):
+            path = path[1:]  # Remove leading slash
+        if not path.startswith("alerting/v1/"):
+            parts = path.split('/')
+            path = f"{self.ROOT}/{parts[-1]}"
+        return self._make_request("DELETE", path)
+
+    def __init__(self, config):
+        super(Alerts, self).__init__(config)
+        self._usage_api = None
+
+    @property
+    def usage(self):
+        """
+        Return interface to usage alerts operations
+
+        :return: :py:class:`ns1.alerting.UsageAlertsAPI`
+        """
+        if self._usage_api is None:
+            # The UsageAlertsAPI expects a client with HTTP methods (_get, _post, etc.)
+            # Since the NS1 object is not directly accessible here, we'll use self as the client
+            # The UsageAlertsAPI only needs HTTP methods (_get, _post, etc.)
+            # For tests, we'll later patch the _c attribute on the UsageAlertsAPI instance
+            self._usage_api = UsageAlertsAPI(self)
+        return self._usage_api
 
     def _buildBody(self, alid, **kwargs):
         body = {}