chore: Add type hints to Chargebee functions

api/organisations/chargebee/chargebee.py

@@ -1,13 +1,19 @@
 import logging
-import typing
 from contextlib import suppress
 from datetime import datetime
+from typing import Any
 
 from chargebee.api_error import (  # type: ignore[import-untyped]
     APIError as ChargebeeAPIError,
 )
 from chargebee.models.hosted_page.operations import (  # type: ignore[import-untyped]
-    HostedPage as HostedPageOps,
+    HostedPage as ChargebeeHostedPageOps,
+)
+from chargebee.models.hosted_page.responses import (  # type: ignore[import-untyped]
+    HostedPageResponse,
+)
+from chargebee.models.plan.responses import (  # type: ignore[import-untyped]
+    RetrieveResponse as ChargebeePlanRetrieveResponse,
 )
 from chargebee.models.portal_session.operations import (  # type: ignore[import-untyped]
     PortalSession as PortalSessionOps,
@@ -49,64 +55,70 @@
 ]
 
 
-def get_subscription_data_from_hosted_page(hosted_page_id):  # type: ignore[no-untyped-def]
-    hosted_page = get_hosted_page(hosted_page_id)  # type: ignore[no-untyped-call]
-    subscription = get_subscription_from_hosted_page(hosted_page)  # type: ignore[no-untyped-call]
-    plan_metadata = get_plan_meta_data(subscription["plan_id"])  # type: ignore[no-untyped-call]
-    if subscription:
-        return {
-            "subscription_id": subscription["id"],
-            "plan": subscription["plan_id"],
-            "subscription_date": datetime.fromtimestamp(
-                subscription["created_at"], tz=UTC
-            ),
-            "max_seats": get_max_seats_for_plan(plan_metadata),
-            "max_api_calls": get_max_api_calls_for_plan(plan_metadata),
-            "customer_id": get_customer_id_from_hosted_page(hosted_page),  # type: ignore[no-untyped-call]
-            "payment_method": CHARGEBEE,
-        }
-    else:
+def get_subscription_data_from_hosted_page(
+    hosted_page_id: str,
+) -> dict[str, Any]:
+    hosted_page = get_hosted_page(hosted_page_id)
+    subscription = get_subscription_from_hosted_page(hosted_page)
+    if not subscription:
         return {}
-
-
-def get_hosted_page(hosted_page_id):  # type: ignore[no-untyped-def]
+    plan_metadata = get_plan_meta_data(subscription["plan_id"])
+    return {
+        "subscription_id": subscription["id"],
+        "plan": subscription["plan_id"],
+        "subscription_date": datetime.fromtimestamp(subscription["created_at"], tz=UTC),
+        "max_seats": get_max_seats_for_plan(plan_metadata),
+        "max_api_calls": get_max_api_calls_for_plan(plan_metadata),
+        "customer_id": get_customer_id_from_hosted_page(hosted_page),
+        "payment_method": CHARGEBEE,
+    }
+
+
+def get_hosted_page(hosted_page_id: str) -> HostedPageResponse:
     response = chargebee_client.HostedPage.retrieve(hosted_page_id)
     return response.hosted_page
 
 
-def get_subscription_from_hosted_page(hosted_page):  # type: ignore[no-untyped-def]
+def get_subscription_from_hosted_page(
+    hosted_page: HostedPageResponse,
+) -> dict[str, Any] | None:
     content = hosted_page.content
     if content and "subscription" in content:
-        return content["subscription"]
+        return content["subscription"]  # type: ignore[no-any-return]
+    return None
 
 
-def get_customer_id_from_hosted_page(hosted_page):  # type: ignore[no-untyped-def]
+def get_customer_id_from_hosted_page(
+    hosted_page: HostedPageResponse,
+) -> str | None:
     content = hosted_page.content
     if content and "customer" in content:
-        return content["customer"]["id"]
+        return content["customer"]["id"]  # type: ignore[no-any-return]
+    return None
 
 
-def get_max_seats_for_plan(meta_data: dict) -> int:  # type: ignore[type-arg]
-    return meta_data.get("seats", 1)  # type: ignore[no-any-return]
+def get_max_seats_for_plan(meta_data: dict[str, Any]) -> int:
+    return int(meta_data.get("seats", 1))
 
 
-def get_max_api_calls_for_plan(meta_data: dict) -> int:  # type: ignore[type-arg]
-    return meta_data.get("api_calls", 50000)  # type: ignore[no-any-return]
+def get_max_api_calls_for_plan(meta_data: dict[str, Any]) -> int:
+    return int(meta_data.get("api_calls", 50000))
 
 
-def get_plan_meta_data(plan_id):  # type: ignore[no-untyped-def]
-    plan_details = get_plan_details(plan_id)  # type: ignore[no-untyped-call]
+def get_plan_meta_data(plan_id: str) -> dict[str, Any]:
+    plan_details = get_plan_details(plan_id)
     if plan_details and hasattr(plan_details.plan, "meta_data"):
         return plan_details.plan.meta_data or {}
     return {}
 
 
-def get_plan_details(plan_id):  # type: ignore[no-untyped-def]
+def get_plan_details(plan_id: str) -> ChargebeePlanRetrieveResponse | None:
     if plan_id:
         return chargebee_client.Plan.retrieve(plan_id)
+    return None
 
 
-def get_portal_url(customer_id, redirect_url):  # type: ignore[no-untyped-def]
+def get_portal_url(customer_id: str, redirect_url: str) -> str | None:
     result = chargebee_client.PortalSession.create(
         PortalSessionOps.CreateParams(
             redirect_url=redirect_url,
@@ -116,21 +128,23 @@ def get_portal_url(customer_id, redirect_url):  # type: ignore[no-untyped-def]
         )
     )
     if result and hasattr(result, "portal_session"):
-        return result.portal_session.access_url
+        return result.portal_session.access_url  # type: ignore[no-any-return]
+    return None
 
 
-def get_customer_id_from_subscription_id(subscription_id):  # type: ignore[no-untyped-def]
+def get_customer_id_from_subscription_id(subscription_id: str) -> str | None:
     subscription_response = chargebee_client.Subscription.retrieve(subscription_id)
     if hasattr(subscription_response, "customer"):
-        return subscription_response.customer.id
+        return subscription_response.customer.id  # type: ignore[no-any-return]
+    return None
 
 
 def get_hosted_page_url_for_subscription_upgrade(
     subscription_id: str, plan_id: str
 ) -> str:
     checkout_existing_response = chargebee_client.HostedPage.checkout_existing(
-        HostedPageOps.CheckoutExistingParams(
-            subscription=HostedPageOps.CheckoutExistingSubscriptionParams(
+        ChargebeeHostedPageOps.CheckoutExistingParams(
+            subscription=ChargebeeHostedPageOps.CheckoutExistingSubscriptionParams(
                 id=subscription_id,
                 plan_id=plan_id,
             ),
@@ -140,7 +154,7 @@ def get_hosted_page_url_for_subscription_upgrade(
 
 
 def extract_subscription_metadata(
-    chargebee_subscription: dict,  # type: ignore[type-arg]
+    chargebee_subscription: dict[str, Any],
     customer_email: str,
 ) -> ChargebeeObjMetadata:
     chargebee_addons = chargebee_subscription.get("addons", [])
@@ -160,9 +174,9 @@ def extract_subscription_metadata(
     return subscription_metadata
 
 
-def get_subscription_metadata_from_id(  # type: ignore[return]
+def get_subscription_metadata_from_id(
     subscription_id: str,
-) -> typing.Optional[ChargebeeObjMetadata]:
+) -> ChargebeeObjMetadata | None:
     if not (subscription_id and subscription_id.strip() != ""):
         logger.warning("Subscription id is empty or None")
         return None
@@ -174,11 +188,14 @@ def get_subscription_metadata_from_id(  # type: ignore[return]
         )
 
         return extract_subscription_metadata(
-            chargebee_subscription, chargebee_result.customer.email
+            chargebee_subscription,
+            chargebee_result.customer.email,
         )
 
+    return None
+
 
-def cancel_subscription(subscription_id: str):  # type: ignore[no-untyped-def]
+def cancel_subscription(subscription_id: str) -> None:
     try:
         chargebee_client.Subscription.cancel(
             subscription_id,
@@ -190,7 +207,7 @@ def cancel_subscription(subscription_id: str):  # type: ignore[no-untyped-def]
         raise CannotCancelChargebeeSubscription(msg) from e
 
 
-def add_single_seat(subscription_id: str):  # type: ignore[no-untyped-def]
+def add_single_seat(subscription_id: str) -> None:
     try:
         subscription = chargebee_client.Subscription.retrieve(
             subscription_id
@@ -292,7 +309,7 @@ def add_100k_api_calls(
 
 def _convert_chargebee_subscription_to_dictionary(
     chargebee_subscription: SubscriptionResponse,
-) -> dict:  # type: ignore[type-arg]
+) -> dict[str, Any]:
     chargebee_subscription_dict = dict(chargebee_subscription.raw_data)
     addons = chargebee_subscription.addons or []
     chargebee_subscription_dict["addons"] = [dict(addon.raw_data) for addon in addons]

api/organisations/models.py

@@ -261,7 +261,7 @@ class Subscription(LifecycleModelMixin, SoftDeleteExportableModel):  # type: ign
     history = HistoricalRecords()
 
     def update_plan(self, plan_id):  # type: ignore[no-untyped-def]
-        plan_metadata = get_plan_meta_data(plan_id)  # type: ignore[no-untyped-call]
+        plan_metadata = get_plan_meta_data(plan_id)
         self.cancellation_date = None
         self.plan = plan_id
         self.max_seats = get_max_seats_for_plan(plan_metadata)
@@ -370,16 +370,19 @@ def prepare_for_cancel(  # type: ignore[no-untyped-def]
         self.billing_status = None
         self.save()
 
-    def get_portal_url(self, redirect_url):  # type: ignore[no-untyped-def]
+    def get_portal_url(self, redirect_url: str) -> str | None:
         if not self.subscription_id:
             return None
 
         if not self.customer_id:
-            self.customer_id = get_customer_id_from_subscription_id(  # type: ignore[no-untyped-call]
+            self.customer_id = get_customer_id_from_subscription_id(
                 self.subscription_id
             )
             self.save()
-        return get_portal_url(self.customer_id, redirect_url)  # type: ignore[no-untyped-call]
+
+        if self.customer_id:
+            return get_portal_url(self.customer_id, redirect_url)
+        return None
 
     def get_subscription_metadata(self) -> BaseSubscriptionMetadata:
         if self.is_free_plan:

api/organisations/serializers.py

@@ -200,7 +200,7 @@ def create(self, validated_data):  # type: ignore[no-untyped-def]
         organisation = self._get_organisation()  # type: ignore[no-untyped-call]
 
         if settings.ENABLE_CHARGEBEE:
-            subscription_data = get_subscription_data_from_hosted_page(  # type: ignore[no-untyped-call]
+            subscription_data = get_subscription_data_from_hosted_page(
                 hosted_page_id=validated_data["hosted_page_id"]
             )
 

api/tests/unit/organisations/chargebee/test_unit_chargebee_chargebee.py

@@ -27,7 +27,12 @@
     get_subscription_data_from_hosted_page,
     get_subscription_metadata_from_id,
 )
-from organisations.chargebee.chargebee import cancel_subscription
+from organisations.chargebee.chargebee import (
+    cancel_subscription,
+    get_customer_id_from_hosted_page,
+    get_plan_details,
+    get_subscription_from_hosted_page,
+)
 from organisations.chargebee.constants import (
     ADDITIONAL_API_SCALE_UP_ADDON_ID,
     ADDITIONAL_SEAT_ADDON_ID,
@@ -198,7 +203,7 @@ def test_chargebee_get_plan_meta_data_returns_correct_metadata(
     )
 
     # When
-    plan_meta_data = get_plan_meta_data(plan_id)  # type: ignore[no-untyped-call]
+    plan_meta_data = get_plan_meta_data(plan_id)
 
     # Then
     assert plan_meta_data == {
@@ -230,7 +235,7 @@ def test_chargebee_get_subscription_data_from_hosted_page_returns_expected_value
     mock_cb.Plan.retrieve.return_value = MockChargeBeePlanResponse(expected_max_seats)  # type: ignore[no-untyped-call]  # noqa: E501
 
     # When
-    subscription_data = get_subscription_data_from_hosted_page("hosted_page_id")  # type: ignore[no-untyped-call]
+    subscription_data = get_subscription_data_from_hosted_page("hosted_page_id")
 
     # Then
     assert subscription_data["subscription_id"] == subscription_id
@@ -252,7 +257,7 @@ def test_get_chargebee_portal_url(mocker: MockerFixture) -> None:
     )
 
     # When
-    portal_url = get_portal_url("some-customer-id", "https://redirect.url.com")  # type: ignore[no-untyped-call]
+    portal_url = get_portal_url("some-customer-id", "https://redirect.url.com")
 
     # Then
     assert portal_url == access_url
@@ -271,7 +276,7 @@ def test_chargebee_get_customer_id_from_subscription(
     )
 
     # When
-    customer_id = get_customer_id_from_subscription_id("subscription-id")  # type: ignore[no-untyped-call]
+    customer_id = get_customer_id_from_subscription_id("subscription-id")
 
     # Then
     assert customer_id == expected_customer_id
@@ -713,3 +718,76 @@ def test_add_100k_api_calls_when_chargebee_api_error_has_no_error_code(
             count=1,
             invoice_immediately=True,
         )
+
+
+def test_get_subscription_from_hosted_page__no_subscription__returns_none(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    hosted_page = mocker.MagicMock(content={"customer": {"id": "cust-1"}})
+
+    # When
+    result = get_subscription_from_hosted_page(hosted_page)
+
+    # Then
+    assert result is None
+
+
+def test_get_customer_id_from_hosted_page__no_customer__returns_none(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    hosted_page = mocker.MagicMock(
+        content={"subscription": {"id": "sub-1", "plan_id": "plan-1"}}
+    )
+
+    # When
+    result = get_customer_id_from_hosted_page(hosted_page)
+
+    # Then
+    assert result is None
+
+
+def test_get_plan_details__empty_plan_id__returns_none() -> None:
+    # Given
+    plan_id = ""
+
+    # When
+    result = get_plan_details(plan_id)
+
+    # Then
+    assert result is None
+
+
+def test_get_portal_url__no_portal_session__returns_none(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_cb = mocker.patch(
+        "organisations.chargebee.chargebee.chargebee_client", autospec=True
+    )
+    mock_result = mocker.MagicMock(spec=[])  # no attributes at all
+    mock_cb.PortalSession.create.return_value = mock_result
+
+    # When
+    result = get_portal_url("customer-id", "https://redirect.url.com")
+
+    # Then
+    assert result is None
+
+
+def test_get_customer_id_from_subscription_id__no_customer__returns_none(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_cb = mocker.patch(
+        "organisations.chargebee.chargebee.chargebee_client", autospec=True
+    )
+    mock_response = mocker.MagicMock(spec=[])  # no customer attribute
+    mock_cb.Subscription.retrieve.return_value = mock_response
+
+    # When
+    result = get_customer_id_from_subscription_id("sub-123")
+
+    # Then
+    assert result is None