Feat(alerts): add batch alert actions (#364)

* add batch alert actions

* rename batch actions

* unify alert model with another PR

Author: annvelents

lago_python_client/models/__init__.py

@@ -1,4 +1,14 @@
 from .activity_log import ActivityLog as ActivityLog, ActivityLogResponse as ActivityLogResponse
+from .alert import (
+    Alert as Alert,
+    AlertsList as AlertsList,
+    AlertThreshold as AlertThreshold,
+    AlertThresholdList as AlertThresholdList,
+    AlertResponse as AlertResponse,
+    AlertsResponseList as AlertsResponseList,
+    AlertThresholdResponse as AlertThresholdResponse,
+    AlertThresholdResponseList as AlertThresholdResponseList,
+)
 from .api_log import ApiLog as ApiLog, ApiLogResponse as ApiLogResponse
 from .applied_coupon import AppliedCoupon as AppliedCoupon
 from .billable_metric import (

lago_python_client/models/alert.py

@@ -0,0 +1,54 @@
+from typing import List, Optional
+
+from ..base_model import BaseModel, BaseResponseModel
+from .billable_metric import BillableMetricResponse
+
+
+class AlertThreshold(BaseModel):
+    code: Optional[str]
+    value: str
+    recurring: Optional[bool]
+
+
+class AlertThresholdList(BaseModel):
+    __root__: List[AlertThreshold]
+
+
+class Alert(BaseModel):
+    alert_type: str
+    code: str
+    name: Optional[str]
+    billable_metric_code: Optional[str]
+    thresholds: AlertThresholdList
+
+
+class AlertsList(BaseModel):
+    alerts: List[Alert]
+
+
+class AlertThresholdResponse(BaseResponseModel):
+    code: Optional[str]
+    value: str
+    recurring: bool
+
+
+class AlertThresholdResponseList(BaseResponseModel):
+    __root__: List[AlertThresholdResponse]
+
+
+class AlertResponse(BaseResponseModel):
+    lago_id: str
+    lago_organization_id: str
+    external_subscription_id: str
+    alert_type: str
+    code: str
+    name: Optional[str]
+    previous_value: Optional[str]
+    last_processed_at: Optional[str]
+    thresholds: AlertThresholdResponseList
+    created_at: str
+    billable_metric: Optional[BillableMetricResponse]
+
+
+class AlertsResponseList(BaseResponseModel):
+    alerts: List[AlertResponse]

lago_python_client/subscriptions/clients.py

@@ -8,20 +8,24 @@
     FindCommandMixin,
     UpdateCommandMixin,
 )
+from ..models.alert import AlertResponse, AlertsList
 from ..models.fixed_charge import FixedChargeResponse
 from ..models.lifetime_usage import LifetimeUsageResponse
 from ..models.subscription import SubscriptionResponse
 from ..services.request import (
     make_headers,
     make_url,
+    send_delete_request,
     send_get_request,
+    send_post_request,
     send_put_request,
 )
 from ..services.response import (
     Response,
     get_response_data,
     prepare_index_response,
     prepare_object_response,
+    verify_response,
 )
 from ..services.json import to_json
 
@@ -71,6 +75,32 @@ def update_lifetime_usage(
             data=get_response_data(response=api_response, key="lifetime_usage"),
         )
 
+    def create_alerts(self, external_id: str, input_object: AlertsList) -> Mapping[str, Any]:
+        api_response: Response = send_post_request(
+            url=make_url(
+                origin=self.base_url,
+                path_parts=(self.API_RESOURCE, external_id, "alerts"),
+            ),
+            content=to_json(input_object.dict()),
+            headers=make_headers(api_key=self.api_key),
+        )
+
+        return prepare_index_response(
+            api_resource="alerts",
+            response_model=AlertResponse,
+            data=get_response_data(response=api_response),
+        )
+
+    def delete_alerts(self, external_id: str) -> None:
+        api_response: Response = send_delete_request(
+            url=make_url(
+                origin=self.base_url,
+                path_parts=(self.API_RESOURCE, external_id, "alerts"),
+            ),
+            headers=make_headers(api_key=self.api_key),
+        )
+        verify_response(api_response)
+
     def find_all_fixed_charges(self, external_id: str) -> Mapping[str, Any]:
         api_response: Response = send_get_request(
             url=make_url(

tests/fixtures/subscription_alerts.json

@@ -0,0 +1,40 @@
+{
+  "alerts": [
+    {
+      "lago_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+      "lago_organization_id": "2b902b90-2b90-2b90-2b90-2b902b902b90",
+      "external_subscription_id": "sub_1234",
+      "alert_type": "current_usage_amount",
+      "code": "alert1",
+      "name": "First Alert",
+      "previous_value": "0.0",
+      "last_processed_at": null,
+      "thresholds": [
+        {
+          "code": "warn",
+          "value": "1000.0",
+          "recurring": false
+        }
+      ],
+      "created_at": "2025-03-20T10:00:00Z"
+    },
+    {
+      "lago_id": "3c903c90-3c90-3c90-3c90-3c903c903c90",
+      "lago_organization_id": "2b902b90-2b90-2b90-2b90-2b902b902b90",
+      "external_subscription_id": "sub_1234",
+      "alert_type": "billable_metric_current_usage_amount",
+      "code": "alert2",
+      "name": "Second Alert",
+      "previous_value": "0.0",
+      "last_processed_at": null,
+      "thresholds": [
+        {
+          "code": null,
+          "value": "2000.0",
+          "recurring": false
+        }
+      ],
+      "created_at": "2025-03-20T10:00:00Z"
+    }
+  ]
+}

tests/test_subscription_client.py

@@ -6,6 +6,7 @@
 from lago_python_client.client import Client
 from lago_python_client.exceptions import LagoApiError
 from lago_python_client.models import Subscription
+from lago_python_client.models.alert import Alert, AlertThreshold, AlertsList
 
 
 def create_subscription():
@@ -400,3 +401,95 @@ def test_fixed_charge_response_model_parsing():
     assert response.properties.graduated_ranges[0].from_value == 0
     assert response.properties.graduated_ranges[0].per_unit_amount == "100"
     assert response.properties.graduated_ranges[1].to_value is None
+
+
+def mock_subscription_alerts_response():
+    this_dir = os.path.dirname(os.path.abspath(__file__))
+    my_data_path = os.path.join(this_dir, "fixtures/subscription_alerts.json")
+
+    with open(my_data_path, "rb") as alerts_response:
+        return alerts_response.read()
+
+
+def test_valid_create_alerts_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    external_id = "sub_1234"
+
+    httpx_mock.add_response(
+        method="POST",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/alerts",
+        content=mock_subscription_alerts_response(),
+    )
+
+    input_object = AlertsList(
+        alerts=[
+            Alert(
+                alert_type="current_usage_amount",
+                code="alert1",
+                name="First Alert",
+                thresholds=[AlertThreshold(code="warn", value="1000")],
+            ),
+            Alert(
+                alert_type="billable_metric_current_usage_amount",
+                code="alert2",
+                billable_metric_code="storage",
+                thresholds=[AlertThreshold(value="2000")],
+            ),
+        ]
+    )
+
+    response = client.subscriptions.create_alerts(external_id, input_object)
+
+    assert len(response["alerts"]) == 2
+    assert response["alerts"][0].lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+    assert response["alerts"][0].code == "alert1"
+    assert response["alerts"][0].alert_type == "current_usage_amount"
+    assert response["alerts"][1].lago_id == "3c903c90-3c90-3c90-3c90-3c903c903c90"
+    assert response["alerts"][1].code == "alert2"
+
+
+def test_invalid_create_alerts_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+    external_id = "sub_1234"
+
+    httpx_mock.add_response(
+        method="POST",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/alerts",
+        status_code=422,
+        content=b"",
+    )
+
+    input_object = AlertsList(alerts=[])
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.create_alerts(external_id, input_object)
+
+
+def test_valid_delete_alerts_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    external_id = "sub_1234"
+
+    httpx_mock.add_response(
+        method="DELETE",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/alerts",
+        status_code=200,
+        content=b"",
+    )
+
+    result = client.subscriptions.delete_alerts(external_id)
+    assert result is None
+
+
+def test_invalid_delete_alerts_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+    external_id = "invalid_sub"
+
+    httpx_mock.add_response(
+        method="DELETE",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/alerts",
+        status_code=404,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.subscriptions.delete_alerts(external_id)