feat(ProgressiveBilling) - add lifetime_usage and update_lifetime_usage to Subscription

lago_python_client/models/__init__.py

@@ -83,3 +83,4 @@
     UsageThresholdsResponse as UsageThresholdsResponse,
 )
 from .payment_request import PaymentRequest as PaymentRequest
+from .lifetime_usage import LifetimeUsageResponse as LifetimeUsageResponse

lago_python_client/models/lifetime_usage.py

@@ -0,0 +1,21 @@
+from typing import Optional, List
+
+from ..base_model import BaseResponseModel
+
+
+class LifetimeUsageThresholdResponse(BaseResponseModel):
+    amount_cents: int
+    completion_ratio: float
+    reached_at: Optional[str]
+
+
+class LifetimeUsageResponse(BaseResponseModel):
+    lago_id: str
+    lago_subscription_id: str
+    external_subscription_id: str
+    external_historical_usage_amount_cents: int
+    invoiced_usage_amount_cents: int
+    current_usage_amount_cents: int
+    from_datetime: Optional[str]
+    to_datetime: Optional[str]
+    usage_thresholds: Optional[List[LifetimeUsageThresholdResponse]]

lago_python_client/subscriptions/clients.py

@@ -8,7 +8,20 @@
     FindCommandMixin,
     UpdateCommandMixin,
 )
+from ..models.lifetime_usage import LifetimeUsageResponse
 from ..models.subscription import SubscriptionResponse
+from ..services.request import (
+    make_headers,
+    make_url,
+    send_get_request,
+    send_put_request,
+)
+from ..services.response import (
+    Response,
+    get_response_data,
+    prepare_object_response,
+)
+from ..services.json import to_json
 
 
 class SubscriptionClient(
@@ -22,3 +35,36 @@ class SubscriptionClient(
     API_RESOURCE: ClassVar[str] = "subscriptions"
     RESPONSE_MODEL: ClassVar[Type[SubscriptionResponse]] = SubscriptionResponse
     ROOT_NAME: ClassVar[str] = "subscription"
+
+    def lifetime_usage(self, resource_id: str) -> LifetimeUsageResponse:
+        api_response: Response = send_get_request(
+            url=make_url(
+                origin=self.base_url,
+                path_parts=(self.API_RESOURCE, resource_id, "lifetime_usage"),
+            ),
+            headers=make_headers(api_key=self.api_key),
+        )
+
+        return prepare_object_response(
+            response_model=LifetimeUsageResponse,
+            data=get_response_data(response=api_response, key="lifetime_usage"),
+        )
+
+    def update_lifetime_usage(
+        self, resource_id: str, external_historical_usage_amount_cents: int
+    ) -> LifetimeUsageResponse:
+        api_response: Response = send_put_request(
+            url=make_url(
+                origin=self.base_url,
+                path_parts=(self.API_RESOURCE, resource_id, "lifetime_usage"),
+            ),
+            content=to_json(
+                {"lifetime_usage": {"external_historical_usage_amount_cents": external_historical_usage_amount_cents}}
+            ),
+            headers=make_headers(api_key=self.api_key),
+        )
+
+        return prepare_object_response(
+            response_model=LifetimeUsageResponse,
+            data=get_response_data(response=api_response, key="lifetime_usage"),
+        )

tests/fixtures/lifetime_usage.json

@@ -0,0 +1,24 @@
+{
+  "lifetime_usage": {
+    "lago_id": "ef555447-b017-4345-9846-6b814cfb4148",
+    "lago_subscription_id": "d644084c-0673-4573-8104-6093bf143acb",
+    "external_subscription_id": "5eb02857-a71e-4ea2-bcf9-57d3a41bc6ba",
+    "external_historical_usage_amount_cents": 0,
+    "invoiced_usage_amount_cents": 0,
+    "current_usage_amount_cents": 3000,
+    "from_datetime": "2024-09-04T00:00:00Z",
+    "to_datetime": "2024-09-05T12:49:28Z",
+    "usage_thresholds": [
+      {
+        "amount_cents": 2000,
+        "completion_ratio": 1,
+        "reached_at": "2024-09-05T12:49:25.604Z"
+      },
+      {
+        "amount_cents": 4000,
+        "completion_ratio": 0.5,
+        "reached_at": null
+      }
+    ]
+  }
+}

tests/fixtures/update_lifetime_usage.json

@@ -0,0 +1,24 @@
+{
+  "lifetime_usage": {
+    "lago_id": "ef555447-b017-4345-9846-6b814cfb4148",
+    "lago_subscription_id": "d644084c-0673-4573-8104-6093bf143acb",
+    "external_subscription_id": "5eb02857-a71e-4ea2-bcf9-57d3a41bc6ba",
+    "external_historical_usage_amount_cents": 2000,
+    "invoiced_usage_amount_cents": 0,
+    "current_usage_amount_cents": 3000,
+    "from_datetime": "2024-09-04T00:00:00Z",
+    "to_datetime": "2024-09-05T12:49:28Z",
+    "usage_thresholds": [
+      {
+        "amount_cents": 2000,
+        "completion_ratio": 1,
+        "reached_at": "2024-09-05T12:49:25.604Z"
+      },
+      {
+        "amount_cents": 4000,
+        "completion_ratio": 0.5,
+        "reached_at": null
+      }
+    ]
+  }
+}

tests/test_subscription_client.py

@@ -27,6 +27,22 @@ def mock_response():
         return subscription_response.read()
 
 
+def mock_lifetime_usage_response():
+    this_dir = os.path.dirname(os.path.abspath(__file__))
+    my_data_path = os.path.join(this_dir, "fixtures/lifetime_usage.json")
+
+    with open(my_data_path, "rb") as subscription_response:
+        return subscription_response.read()
+
+
+def mock_update_lifetime_usage_response():
+    this_dir = os.path.dirname(os.path.abspath(__file__))
+    my_data_path = os.path.join(this_dir, "fixtures/update_lifetime_usage.json")
+
+    with open(my_data_path, "rb") as subscription_response:
+        return subscription_response.read()
+
+
 def mock_response_for_pending():
     this_dir = os.path.dirname(os.path.abspath(__file__))
     my_data_path = os.path.join(this_dir, "fixtures/pending_subscription.json")
@@ -210,3 +226,51 @@ def test_invalid_find_all_subscription_request(httpx_mock: HTTPXMock):
 
     with pytest.raises(LagoApiError):
         client.subscriptions.find_all()
+
+
+def test_valid_lifetime_usage_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    external_id = "5eb02857-a71e-4ea2-bcf9-57d3a41bc6ba"
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/lifetime_usage",
+        content=mock_lifetime_usage_response(),
+    )
+    response = client.subscriptions.lifetime_usage(external_id)
+
+    assert response.lago_id == "ef555447-b017-4345-9846-6b814cfb4148"
+    assert response.current_usage_amount_cents == 3000
+    assert response.usage_thresholds[0].amount_cents == 2000
+    assert response.usage_thresholds[0].completion_ratio == 1
+    assert response.usage_thresholds[1].amount_cents == 4000
+    assert response.usage_thresholds[1].completion_ratio == 0.5
+
+
+def test_invalid_lifetime_usage_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    external_id = "notfound"
+
+    httpx_mock.add_response(
+        method="GET",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/lifetime_usage",
+        status_code=404,
+        content=b"",
+    )
+    with pytest.raises(LagoApiError):
+        client.subscriptions.lifetime_usage(external_id)
+
+
+def test_update_lifetime_usage_request(httpx_mock: HTTPXMock):
+    client = Client(api_key="886fe239-927d-4072-ab72-6dd345e8dd0d")
+    external_id = "5eb02857-a71e-4ea2-bcf9-57d3a41bc6ba"
+
+    httpx_mock.add_response(
+        method="PUT",
+        url="https://api.getlago.com/api/v1/subscriptions/" + external_id + "/lifetime_usage",
+        content=mock_update_lifetime_usage_response(),
+    )
+    response = client.subscriptions.update_lifetime_usage(external_id, 2000)
+
+    assert response.lago_id == "ef555447-b017-4345-9846-6b814cfb4148"
+    assert response.external_historical_usage_amount_cents == 2000