feat(alerting): Add batch actions for wallet alerts (#372)

* Add batch actions for wallet alerts

* Fix linters

Author: vankiru

lago_python_client/customers/wallet_client.py

@@ -10,8 +10,8 @@
     NestedUpdateCommandMixin,
 )
 from ..models.wallet import WalletResponse
-from .wallets.metadata_client import CustomerWalletMetadataClient
 from .wallets.alert_client import CustomerWalletAlertClient
+from .wallets.metadata_client import CustomerWalletMetadataClient
 
 
 class CustomerWalletClient(

lago_python_client/customers/wallets/alert_client.py

@@ -1,14 +1,26 @@
 from typing import ClassVar, Type
 
 from ...base_client import BaseClient
-from ...models.alert import AlertResponse
-
 from ...mixins import (
     NestedCreateCommandMixin,
-    NestedUpdateCommandMixin,
     NestedDestroyCommandMixin,
-    NestedFindCommandMixin,
     NestedFindAllCommandMixin,
+    NestedFindCommandMixin,
+    NestedUpdateCommandMixin,
+)
+from ...models.alert import AlertResponse, AlertsList
+from ...services.json import to_json
+from ...services.request import (
+    make_headers,
+    make_url,
+    send_delete_request,
+    send_post_request,
+)
+from ...services.response import (
+    Response,
+    get_response_data,
+    prepare_index_response,
+    verify_response,
 )
 
 
@@ -26,3 +38,27 @@ class CustomerWalletAlertClient(
 
     def api_resource(self, customer_id: str, wallet_code: str) -> tuple[str]:
         return ("customers", customer_id, "wallets", wallet_code, "alerts")
+
+    def create_batch(self, customer_id: str, wallet_code: str, input_object: AlertsList) -> None:
+        api_response: Response = send_post_request(
+            url=make_url(origin=self.base_url, path_parts=self.api_resource(customer_id, wallet_code)),
+            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 destroy_all(self, customer_id: str, wallet_code: str) -> None:
+        api_response: Response = send_delete_request(
+            url=make_url(
+                origin=self.base_url,
+                path_parts=self.api_resource(customer_id, wallet_code),
+            ),
+            headers=make_headers(api_key=self.api_key),
+        )
+
+        verify_response(api_response)

tests/fixtures/wallet_alert_index.json

@@ -22,6 +22,29 @@
       "billable_metric": null,
       "created_at": "2025-03-20T10:00:00Z",
       "last_processed_at": "2025-05-19T10:04:21Z"
+    },
+    {
+      "lago_id": "1a921a92-1a92-1a92-1a92-1a921a921a92",
+      "lago_organization_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+      "external_subscription_id": null,
+      "external_customer_id": "cus_0987654321",
+      "lago_wallet_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
+      "wallet_code": "wallet_code",
+      "code": "wallet_credits_alert",
+      "name": "Credits Balance Alert",
+      "alert_type": "wallet_credits_balance",
+      "direction": "increasing",
+      "previous_value": 1000,
+      "thresholds": [
+        {
+          "code": "warn",
+          "recurring": false,
+          "value": "99.0"
+        }
+      ],
+      "billable_metric": null,
+      "created_at": "2025-03-20T10:00:00Z",
+      "last_processed_at": "2025-05-19T10:04:21Z"
     }
   ],
   "meta": {

tests/test_customer_wallet_alert_client.py

@@ -5,6 +5,7 @@
 from lago_python_client.exceptions import LagoApiError
 from lago_python_client.models import (
     Alert,
+    AlertsList,
     AlertThreshold,
     AlertThresholdList,
 )
@@ -176,6 +177,7 @@ def test_valid_find_all_customer_wallet_alerts_request_with_options(httpx_mock:
         "customer_id", "wallet_code", options={"per_page": 2, "page": 1}
     )
 
+    assert len(response) == 2
     assert response["alerts"][0].lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
     assert response["meta"]["current_page"] == 1
 
@@ -192,3 +194,86 @@ def test_invalid_find_all_wallet_alerts_request(httpx_mock: HTTPXMock):
 
     with pytest.raises(LagoApiError):
         client.customers.wallets.alerts.find_all("customer_id", "wallet_code")
+
+
+def test_valid_create_batch_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/customers/customer_id/wallets/wallet_code/alerts",
+        content=mock_response("wallet_alert_index"),
+    )
+
+    input = AlertsList(
+        alerts=[
+            Alert(
+                alert_type="wallet_balance_amount",
+                code="wallet_balance_alert",
+                name="Balance Amount Alert",
+                thresholds=[AlertThreshold(code="warn", value="1000")],
+            ),
+            Alert(
+                alert_type="wallet_credits_balance",
+                code="wallet_credits_alert",
+                name="Credits Balance Alert",
+                thresholds=[AlertThreshold(value="2000")],
+            ),
+        ]
+    )
+
+    response = client.customers.wallets.alerts.create_batch("customer_id", "wallet_code", input)
+
+    assert len(response["alerts"]) == 2
+
+    assert response["alerts"][0].lago_id == "1a901a90-1a90-1a90-1a90-1a901a901a90"
+    assert response["alerts"][0].wallet_code == "wallet_code"
+    assert response["alerts"][0].code == "wallet_balance_alert"
+    assert response["alerts"][0].alert_type == "wallet_balance_amount"
+
+    assert response["alerts"][1].lago_id == "1a921a92-1a92-1a92-1a92-1a921a921a92"
+    assert response["alerts"][1].wallet_code == "wallet_code"
+    assert response["alerts"][1].code == "wallet_credits_alert"
+    assert response["alerts"][1].alert_type == "wallet_credits_balance"
+
+
+def test_invalid_create_batch_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+
+    httpx_mock.add_response(
+        method="POST",
+        url="https://api.getlago.com/api/v1/customers/customer_id/wallets/wallet_code/alerts",
+        status_code=401,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.customers.wallets.alerts.create_batch("customer_id", "wallet_code", AlertsList(alerts=[]))
+
+
+def test_valid_destroy_all_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+
+    httpx_mock.add_response(
+        method="DELETE",
+        url="https://api.getlago.com/api/v1/customers/customer_id/wallets/wallet_code/alerts",
+        status_code=200,
+        content=b"",
+    )
+
+    response = client.customers.wallets.alerts.destroy_all("customer_id", "wallet_code")
+    assert response is None
+
+
+def test_invalid_destroy_all_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="invalid")
+
+    httpx_mock.add_response(
+        method="DELETE",
+        url="https://api.getlago.com/api/v1/customers/customer_id/wallets/wallet_code/alerts",
+        status_code=404,
+        content=b"",
+    )
+
+    with pytest.raises(LagoApiError):
+        client.customers.wallets.alerts.destroy_all("customer_id", "wallet_code")