Release 1.0.8 (#24)

Support the newly added billing endpoint `GET https://billing.tidbapi.com/v1beta1/bills/{YYYY-MM}`, which is compatible with TiDB Cloud API v1beta1 [Release 20230928](https://docs.pingcap.com/tidbcloud/api/v1beta1#section/API-Changelog/2023-09-28).

To be compatible with TiDB Cloud API v1beta and v1beta1 in the current release (1.0.8), the following changes are made:

- Introduce the `tidbcloudy.baseURL` module to specify the base URL of the API server. For v1beta, the base URL is `https://api.tidbcloud.com/v1beta`. For the billing system v1beta1, the base URL is `https://billing.tidbapi.com/v1beta1`.
- To support v1beta and v1beta1 in 1.0.8, `Context.base_url` is removed. Instead, `Context().call_...()` methods accept a `base_url` parameter to specify the base URL of the API server.
- To reduce the change of preceding code, such as the Project, Cluster, Backup, and Restore classes, the v1beta base URL is set as the default value when calling `Context().call_...()` methods.
- Introduce the `TiDBCloud().get_monthly_bill()` method to support the newly added billing endpoint **Return organization monthly bills** and describe the usage in [`examples/v1beta1_get_monthly_bill.py`](/examples/v1beta1_get_monthly_bill.py).

examples/v1beta1_get_monthly_bill.py

@@ -0,0 +1,16 @@
+import os
+
+import tidbcloudy
+
+public_key = os.environ.get("PUBLIC_KEY")
+private_key = os.environ.get("PRIVATE_KEY")
+debug_mode = os.environ.get("TIDBCLOUDY_LOG")
+
+api = tidbcloudy.TiDBCloud(public_key=public_key, private_key=private_key)
+billing = api.get_monthly_bill(month="2023-10")
+# billing = api.get_monthly_bill(month="202310")
+# billing = api.get_current_month_bill()
+print(billing)
+print(billing.overview)
+print(billing.summaryByProject)
+print(billing.summaryByService)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "tidbcloudy"
-version = "1.0.7"
+version = "1.0.8"
 description = "(Unofficial) Python SDK for TiDB Cloud"
 readme = "README.md"
 authors = ["Aolin <[email protected]>"]

tidbcloudy/baseURL.py

@@ -0,0 +1,9 @@
+from enum import Enum
+
+
+class V1BETA(Enum):
+    HOST = "https://api.tidbcloud.com/api/v1beta/"
+
+
+class V1BETA1(Enum):
+    BILLING = "https://billing.tidbapi.com/v1beta1/"

tidbcloudy/context.py

@@ -1,29 +1,24 @@
 import httpx
+
+from tidbcloudy.baseURL import V1BETA
 from tidbcloudy.exception import TiDBCloudResponseException
 
 
 class Context:
-    def __init__(self,
-                 public_key: str,
-                 private_key: str,
-                 *,
-                 base_url: str = "https://api.tidbcloud.com/api/v1beta/"
-                 ):
+    def __init__(self, public_key: str, private_key: str):
         """
         Args:
             public_key: your public key to access to TiDB Cloud
             private_key: your private key to access to TiDB Cloud
-            base_url: the base_url of TiDB Cloud API, you can change this for internal testing.
         """
         self._client = httpx.Client()
         self._client.auth = httpx.DigestAuth(public_key, private_key)
-        self._base_url = base_url
-        if self._base_url[-1] != "/":
-            self._base_url += "/"
 
-    def _call_api(self, method: str, path: str, **kwargs) -> dict:
+    def _call_api(self, method: str, path: str, base_url: str, **kwargs) -> dict:
+        if base_url[-1] != "/":
+            base_url += "/"
         try:
-            resp = self._client.request(method=method, url=self._base_url + path, **kwargs)
+            resp = self._client.request(method=method, url=base_url + path, **kwargs)
             resp.raise_for_status()
             return resp.json()
         except httpx.RequestError as exc:
@@ -32,18 +27,26 @@ def _call_api(self, method: str, path: str, **kwargs) -> dict:
         except httpx.HTTPStatusError as exc:
             raise TiDBCloudResponseException(status=exc.response.status_code, message=exc.response.text)
 
-    def call_get(self, path: str, *, params: dict = None) -> dict:
-        resp = self._call_api(method="GET", path=path, params=params)
+    def call_get(self, path: str, base_url: str = V1BETA.HOST.value,
+                 *,
+                 params: dict = None) -> dict:
+        resp = self._call_api(method="GET", path=path, base_url=base_url, params=params)
         return resp
 
-    def call_post(self, path: str, *, data: dict = None, json: dict = None) -> dict:
-        resp = self._call_api(method="POST", path=path, data=data, json=json)
+    def call_post(self, path: str, base_url: str = V1BETA.HOST.value,
+                  *,
+                  data: dict = None,
+                  json: dict = None) -> dict:
+        resp = self._call_api(method="POST", path=path, base_url=base_url, data=data, json=json)
         return resp
 
-    def call_patch(self, path: str, *, data: dict = None, json: dict = None) -> dict:
-        resp = self._call_api(method="PATCH", path=path, data=data, json=json)
+    def call_patch(self, path: str, base_url: str = V1BETA.HOST.value,
+                   *,
+                   data: dict = None,
+                   json: dict = None) -> dict:
+        resp = self._call_api(method="PATCH", path=path, base_url=base_url, data=data, json=json)
         return resp
 
-    def call_delete(self, path) -> dict:
-        resp = self._call_api(method="DELETE", path=path)
+    def call_delete(self, path: str, base_url: str = V1BETA.HOST.value) -> dict:
+        resp = self._call_api(method="DELETE", base_url=base_url, path=path)
         return resp

tidbcloudy/specification.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Union, List
+from typing import List, Union
 
 from ._base import TiDBCloudyBase, TiDBCloudyField, TiDBCloudyListField
 from .util.ip import get_current_ip_address
@@ -122,8 +122,8 @@ class TiDBNodeMap(TiDBCloudyBase):
 
 
 class TiKVNodeMap(TiDBCloudyBase):
-    __slots__ = [
-        "_node_name", "_availability_zone", "_node_size", "_vcpu_num", "_ram_bytes", "_storage_size_gib", "_status"]
+    __slots__ = ["_node_name", "_availability_zone", "_node_size", "_vcpu_num", "_ram_bytes", "_storage_size_gib",
+                 "_status"]
 
     node_name: str = TiDBCloudyField(str)
     availability_zone: str = TiDBCloudyField(str)
@@ -276,23 +276,17 @@ def to_object(self) -> dict:
                 "components": {
                     "tidb": {
                         "node_size": self._tidb.node_size,
-                        "node_quantity": self._tidb.node_quantity
-                    } if self._tidb is not None else None,
+                        "node_quantity": self._tidb.node_quantity} if self._tidb is not None else None,
                     "tikv": {
                         "node_size": self._tikv.node_size,
                         "node_quantity": self._tikv.node_quantity,
-                        "storage_size_gib": self._tikv.storage_size_gib
-                    } if self._tikv is not None else None,
+                        "storage_size_gib": self._tikv.storage_size_gib} if self._tikv is not None else None,
                     "tiflash": {
                         "node_size": self._tiflash.node_size,
                         "node_quantity": self._tiflash.node_quantity,
-                        "storage_size_gib": self._tiflash.storage_size_gib
-                    } if self._tiflash is not None else None
-                },
-                "ip_access_list": [
-                    item.to_object() for item in self._ip_access_list] if self._ip_access_list is not None else None
-            }
-        }
+                        "storage_size_gib": self._tiflash.storage_size_gib} if self._tiflash is not None else None},
+                "ip_access_list": [item.to_object() for item in
+                                   self._ip_access_list] if self._ip_access_list is not None else None}}
 
 
 class UpdateClusterConfig:
@@ -340,11 +334,7 @@ def to_object(self) -> dict:
             if self._tiflash.storage_size_gib is not None:
                 components["tiflash"]["storage_size_gib"] = self._tiflash.storage_size_gib
 
-        return {
-            "config": {
-                "components": components
-            }
-        }
+        return {"config": {"components": components}}
 
 
 class ClusterInfo(TiDBCloudyBase):
@@ -369,8 +359,78 @@ class ClusterInfoOfRestore(TiDBCloudyBase):
     status: ClusterStatus = TiDBCloudyField(ClusterStatus)
 
     def __repr__(self):
-        return "<id={}, name={}, status={}>".format(self.id, self.name,
-                                                    self.status.value if self.status is not None else None)
+        return "<id={}, name={}, status={}>".format(
+            self.id, self.name,
+            self.status.value if self.status is not None else None)
+
+
+class BillingBase(TiDBCloudyBase):
+    __slots__ = ["_credits", "_discounts", "_runningTotal", "_totalCost"]
+    credits: str = TiDBCloudyField(str)
+    discounts: str = TiDBCloudyField(str)
+    runningTotal: str = TiDBCloudyField(str)
+    totalCost: str = TiDBCloudyField(str)
+
+
+class BillingMonthOverview(BillingBase):
+    __slots__ = ["_billedMonth"] + BillingBase.__slots__
+    billedMonth: str = TiDBCloudyField(str)
+
+    def __repr__(self):
+        return "<BillingMonthOverview billed_month={} credits={} running_total={} total_cost={}>".format(
+            self.billedMonth, self.credits, self.runningTotal, self.totalCost)
+
+
+class BillingOtherCharges(BillingBase):
+    __slots__ = ["_chargeName"] + BillingBase.__slots__
+    chargeName: str = TiDBCloudyField(str)
+
+    def __repr__(self):
+        return "<BillingOtherCharges charge_name={} credits={} running_total={} total_cost={}>".format(
+            self.chargeName, self.credits, self.runningTotal, self.totalCost)
+
+
+class BillingProjectCharges(BillingBase):
+    __slots__ = ["_projectName"] + BillingBase.__slots__
+    projectName: str = TiDBCloudyField(str)
+
+    def __repr__(self):
+        return "<BillingProjectCharges project_name={} credits={} running_total={} total_cost={}>".format(
+            self.projectName, self.credits, self.runningTotal, self.totalCost)
+
+
+class BillingMonthSummaryByProject(TiDBCloudyBase):
+    __slots__ = ["_otherCharges", "_projects"]
+    otherCharges: list = TiDBCloudyListField(BillingOtherCharges)
+    projects: list = TiDBCloudyListField(BillingProjectCharges)
+
+    def __repr__(self):
+        return "<BillingMonthSummaryByProject other_charges={} projects={}>".format(
+            self.otherCharges, self.projects)
+
+
+class BillingServiceCost(TiDBCloudyBase):
+    __slots__ = []
+
+
+class BillingMonthSummaryByService(TiDBCloudyBase):
+    __slots__ = ["_serviceCosts", "_serviceName"]
+    serviceCosts: List[dict] = TiDBCloudyListField(BillingServiceCost)
+    serviceName: str = TiDBCloudyField(str)
+
+    def __repr__(self):
+        return "<BillingMonthSummaryByService service_name={} service_costs={}>".format(
+            self.serviceName, self.serviceCosts)
+
+
+class BillingMonthSummary(TiDBCloudyBase):
+    __slots__ = ["_overview", "_summaryByProject", "_summaryByService"]
+    overview: BillingMonthOverview = TiDBCloudyField(BillingMonthOverview)
+    summaryByProject: BillingMonthSummaryByProject = TiDBCloudyField(BillingMonthSummaryByProject)
+    summaryByService: BillingMonthSummaryByService = TiDBCloudyListField(BillingMonthSummaryByService)
+
+    def __repr__(self):
+        return "<BillingMonthSummary month={}>".format(self.overview.billedMonth)
 
 
 class CloudSpecification(TiDBCloudyBase):
@@ -385,5 +445,4 @@ class CloudSpecification(TiDBCloudyBase):
     def __repr__(self):
         return "<Specification cluster_type={} cloud_provider={} region={}>".format(
             self.cluster_type.value if self.cluster_type is not None else None,
-            self.cloud_provider.value if self.cloud_provider is not None else None,
-            self.region)
+            self.cloud_provider.value if self.cloud_provider is not None else None, self.region)

tidbcloudy/tidbcloud.py

@@ -1,9 +1,11 @@
 from typing import Iterator, List
 
+from tidbcloudy.baseURL import V1BETA1
 from tidbcloudy.context import Context
 from tidbcloudy.project import Project
-from tidbcloudy.specification import CloudSpecification
+from tidbcloudy.specification import BillingMonthSummary, CloudSpecification
 from tidbcloudy.util.page import Page
+from tidbcloudy.util.timestamp import get_current_year_month
 
 
 class TiDBCloud:
@@ -108,3 +110,41 @@ def list_provider_regions(self) -> List[CloudSpecification]:
         """
         resp = self._context.call_get(path="clusters/provider/regions")
         return [CloudSpecification.from_object(obj=item) for item in resp["items"]]
+
+    def get_monthly_bill(self, month: str) -> BillingMonthSummary:
+        """
+        Get the monthly billing.
+        Args:
+            month: the month of the bill, format: YYYY-MM or YYYYMM.
+        Returns:
+            the monthly billing.
+
+        Examples:
+            .. code-block:: python
+                import tidbcloudy
+                api = tidbcloudy.TiDBCloud(public_key="your_public_key", private_key="your_private_key")
+                billing = api.get_monthly_bill(month="2023-08")
+                print(billing)
+
+        """
+        if "-" not in month and len(month) == 6:
+            month = f"{month[:4]}-{month[4:]}"
+        path = f"bills/{month}"
+        resp = self._context.call_get(path=path, base_url=V1BETA1.BILLING.value)
+        return BillingMonthSummary.from_object(self._context, resp)
+
+    def get_current_month_bill(self) -> BillingMonthSummary:
+        """
+        Get the billing of current month.
+        Returns:
+            the current month billing.
+
+        Examples:
+            .. code-block:: python
+                import tidbcloudy
+                api = tidbcloudy.TiDBCloud(public_key="your_public_key", private_key="your_private_key")
+                billing = api.get_current_month_bill()
+                print(billing)
+
+        """
+        return self.get_monthly_bill(month=get_current_year_month())

tidbcloudy/util/timestamp.py

@@ -8,8 +8,19 @@ def timestamp_to_string(timestamp: int) -> str:
         timestamp:
 
     Returns:
+        the datetime string.
 
     """
     if timestamp is None:
         return ""
     return datetime.datetime.fromtimestamp(timestamp).isoformat()
+
+
+def get_current_year_month() -> str:
+    """
+    Get current year and month.
+    Returns:
+        the year and month string in format of YYYY-MM.
+
+    """
+    return datetime.datetime.now().strftime("%Y-%m")