feat(alerting): Add subscriptions alerts API

lago_python_client/models/alert.py

@@ -15,11 +15,11 @@ class AlertThresholdList(BaseModel):
 
 
 class Alert(BaseModel):
-    alert_type: str
-    code: str
+    alert_type: Optional[str]
+    code: Optional[str]
     name: Optional[str]
+    thresholds: Optional[AlertThresholdList]
     billable_metric_code: Optional[str]
-    thresholds: AlertThresholdList
 
 
 class AlertsList(BaseModel):
@@ -39,10 +39,13 @@ class AlertThresholdResponseList(BaseResponseModel):
 class AlertResponse(BaseResponseModel):
     lago_id: str
     lago_organization_id: str
-    external_subscription_id: str
+    external_subscription_id: Optional[str]
+    lago_wallet_id: Optional[str]
+    wallet_code: Optional[str]
     alert_type: str
     code: str
     name: Optional[str]
+    direction: Optional[str]
     previous_value: Optional[str]
     last_processed_at: Optional[str]
     thresholds: AlertThresholdResponseList

lago_python_client/subscriptions/alert_client.py

@@ -0,0 +1,28 @@
+from typing import ClassVar, Type
+
+from ..base_client import BaseClient
+from ..models.alert import AlertResponse
+
+from ..mixins import (
+    NestedCreateCommandMixin,
+    NestedUpdateCommandMixin,
+    NestedDestroyCommandMixin,
+    NestedFindCommandMixin,
+    NestedFindAllCommandMixin,
+)
+
+
+class SubscriptionAlertClient(
+    NestedCreateCommandMixin[AlertResponse],
+    NestedUpdateCommandMixin[AlertResponse],
+    NestedDestroyCommandMixin[AlertResponse],
+    NestedFindCommandMixin[AlertResponse],
+    NestedFindAllCommandMixin[AlertResponse],
+    BaseClient,
+):
+    API_RESOURCE: ClassVar[str] = "alerts"
+    RESPONSE_MODEL: ClassVar[Type[AlertResponse]] = AlertResponse
+    ROOT_NAME: ClassVar[str] = "alert"
+
+    def api_resource(self, subscription_id: str) -> tuple[str]:
+        return ("subscriptions", subscription_id, "alerts")

lago_python_client/subscriptions/clients.py

@@ -1,5 +1,6 @@
 from typing import Any, ClassVar, Mapping, Type
 
+from ..functools_ext import callable_cached_property
 from ..base_client import BaseClient
 from ..mixins import (
     CreateCommandMixin,
@@ -29,6 +30,8 @@
 )
 from ..services.json import to_json
 
+from .alert_client import SubscriptionAlertClient
+
 
 class SubscriptionClient(
     CreateCommandMixin[SubscriptionResponse],
@@ -42,6 +45,10 @@ class SubscriptionClient(
     RESPONSE_MODEL: ClassVar[Type[SubscriptionResponse]] = SubscriptionResponse
     ROOT_NAME: ClassVar[str] = "subscription"
 
+    @callable_cached_property
+    def alerts(self) -> SubscriptionAlertClient:
+        return SubscriptionAlertClient(self.base_url, self.api_key)
+
     def lifetime_usage(self, resource_id: str) -> LifetimeUsageResponse:
         api_response: Response = send_get_request(
             url=make_url(

tests/fixtures/subscription_alert.json

@@ -0,0 +1,45 @@
+{
+  "alert": {
+    "lago_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+    "lago_organization_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+    "external_subscription_id": "subscription_id",
+    "lago_wallet_id": null,
+    "wallet_code": null,
+    "alert_type": "billable_metric_current_usage_amount",
+    "code": "storage_threshold_alert",
+    "name": "Storage Usage Alert",
+    "billable_metric": {
+      "lago_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+      "name": "Storage",
+      "code": "storage",
+      "description": "GB of storage used in my application",
+      "recurring": false,
+      "rounding_function": "round",
+      "rounding_precision": 2,
+      "created_at": "2022-09-14T16:35:31Z",
+      "expression": "round((ended_at - started_at) * units)",
+      "field_name": "gb",
+      "aggregation_type": "sum_agg",
+      "weighted_interval": "seconds",
+      "filters": [
+        {
+          "key": "region",
+          "values": [
+            "us-east-1"
+          ]
+        }
+      ]
+    },
+    "previous_value": 1000,
+    "direction": "increasing",
+    "thresholds": [
+      {
+        "code": "warn",
+        "recurring": false,
+        "value": "99.0"
+      }
+    ],
+    "created_at": "2025-03-20T10:00:00Z",
+    "last_processed_at": "2025-05-19T10:04:21Z"
+  }
+}

tests/fixtures/subscription_alert_index.json

@@ -0,0 +1,54 @@
+{
+  "alerts": [
+    {
+      "lago_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+      "lago_organization_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+      "external_subscription_id": "subscription_id",
+      "lago_wallet_id": null,
+      "wallet_code": null,
+      "alert_type": "billable_metric_current_usage_amount",
+      "code": "storage_threshold_alert",
+      "name": "Storage Usage Alert",
+      "billable_metric": {
+        "lago_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+        "name": "Storage",
+        "code": "storage",
+        "description": "GB of storage used in my application",
+        "recurring": false,
+        "rounding_function": "round",
+        "rounding_precision": 2,
+        "created_at": "2022-09-14T16:35:31Z",
+        "expression": "round((ended_at - started_at) * units)",
+        "field_name": "gb",
+        "aggregation_type": "sum_agg",
+        "weighted_interval": "seconds",
+        "filters": [
+          {
+            "key": "region",
+            "values": [
+              "us-east-1"
+            ]
+          }
+        ]
+      },
+      "previous_value": 1000,
+      "direction": "increasing",
+      "thresholds": [
+        {
+          "code": "warn",
+          "recurring": false,
+          "value": "99.0"
+        }
+      ],
+      "created_at": "2025-03-20T10:00:00Z",
+      "last_processed_at": "2025-05-19T10:04:21Z"
+    }
+  ],
+  "meta": {
+    "current_page": 1,
+    "next_page": 2,
+    "prev_page": null,
+    "total_pages": 4,
+    "total_count": 70
+  }
+}

tests/test_subscription_alert_client.py

@@ -0,0 +1,192 @@
+import pytest
+from pytest_httpx import HTTPXMock
+
+from lago_python_client.client import Client
+from lago_python_client.exceptions import LagoApiError
+from lago_python_client.models import (
+    Alert,
+    AlertThreshold,
+    AlertThresholdList,
+)
+
+from .utils.mixin import mock_response
+
+
+def alert_object():
+    threshold = AlertThreshold(code="warn", value=10000, recurring=False)
+
+    return Alert(
+        alert_type="billable_metric_current_usage_amount",
+        code="storage_threshold_alert",
+        name="Storage Usage Alert",
+        billable_metric_code="storage",
+        thresholds=AlertThresholdList(__root__=[threshold]),
+    )
+
+
+def test_valid_create_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+
+    httpx_mock.add_response(
+        method="POST",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts",
+        content=mock_response("subscription_alert"),
+    )
+    response = client.subscriptions.alerts.create("subscription_id", alert_object())
+
+    assert response.lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+    assert response.lago_organization_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+    assert response.external_subscription_id == "subscription_id"
+    assert response.lago_wallet_id is None
+    assert response.wallet_code is None
+    assert response.code == "storage_threshold_alert"
+    assert response.name == "Storage Usage Alert"
+    assert response.alert_type == "billable_metric_current_usage_amount"
+    assert response.previous_value == "1000"
+    assert response.billable_metric.code == "storage"
+    assert response.thresholds == AlertThresholdList(
+        __root__=[AlertThreshold(code="warn", value=99.0, recurring=False)]
+    )
+
+
+def test_invalid_create_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+
+    httpx_mock.add_response(
+        method="POST",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts",
+        status_code=401,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.alerts.create("subscription_id", alert_object())
+
+
+def test_valid_update_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    code = "alert-code"
+
+    httpx_mock.add_response(
+        method="PUT",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts/" + code,
+        content=mock_response("subscription_alert"),
+    )
+    response = client.subscriptions.alerts.update("subscription_id", code, alert_object())
+
+    assert response.lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+
+
+def test_invalid_update_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+    code = "invalid"
+
+    httpx_mock.add_response(
+        method="PUT",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts/" + code,
+        status_code=401,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.alerts.update("subscription_id", code, alert_object())
+
+
+def test_valid_find_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    code = "alert-code"
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts/" + code,
+        content=mock_response("subscription_alert"),
+    )
+    response = client.subscriptions.alerts.find("subscription_id", code)
+
+    assert response.lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+
+
+def test_invalid_find_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+    code = "invalid"
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts/" + code,
+        status_code=404,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.alerts.find("subscription_id", code)
+
+
+def test_valid_destroy_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    code = "alert-code"
+
+    httpx_mock.add_response(
+        method="DELETE",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts/" + code,
+        content=mock_response("subscription_alert"),
+    )
+    response = client.subscriptions.alerts.destroy("subscription_id", code)
+
+    assert response.lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+
+
+def test_invalid_destroy_subscription_alert_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+    code = "invalid"
+
+    httpx_mock.add_response(
+        method="DELETE",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts/" + code,
+        status_code=404,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.alerts.destroy("subscription_id", code)
+
+
+def test_valid_find_all_subscription_alerts_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts",
+        content=mock_response("subscription_alert_index"),
+    )
+    response = client.subscriptions.alerts.find_all("subscription_id")
+
+    assert response["alerts"][0].lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+    assert response["meta"]["current_page"] == 1
+
+
+def test_valid_find_all_subscription_alerts_request_with_options(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts?page=1&per_page=2",
+        content=mock_response("subscription_alert_index"),
+    )
+    response = client.subscriptions.alerts.find_all("subscription_id", options={"per_page": 2, "page": 1})
+
+    assert response["alerts"][0].lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+    assert response["meta"]["current_page"] == 1
+
+
+def test_invalid_find_all_subscription_alerts_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/subscription_id/alerts",
+        status_code=404,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.alerts.find_all("subscription_id")