fix: align API with requests.get and add typing support (#1)

acrclient/client.py

@@ -1,4 +1,21 @@
-import requests
+from typing import Any, Optional
+
+from requests import get
+from requests.auth import AuthBase
+from requests.models import PreparedRequest, Response
+
+from .models import GetBmCsProjectsResultsParams
+
+
+class _Auth(AuthBase):  # pylint: disable=too-few-public-methods
+    """Bearer token style auth for ACRCloud."""
+
+    def __init__(self, bearer_token: str) -> None:
+        self.bearer_token = bearer_token
+
+    def __call__(self, request: PreparedRequest) -> PreparedRequest:
+        request.headers["Authorization"] = f"Bearer {self.bearer_token}"
+        return request
 
 
 class Client:
@@ -8,30 +25,34 @@ class Client:
         bearer_token: The bearer token for ACRCloud.
     """
 
-    def __init__(self, bearer_token, base_url="https://eu-api-v2.acrcloud.com"):
+    def __init__(self, bearer_token: str, base_url="https://eu-api-v2.acrcloud.com"):
         self.base_url = base_url
-        self.bearer_token = bearer_token
+        self.auth = _Auth(bearer_token=bearer_token)
 
-    def request(self, path, headers=None, **kwargs):
+    def get(self, path: str, params: Any = None, **kwargs) -> Response:
         """Fetch JSON data from ACRCloud API with set Access Key param."""
+        url = f"{self.base_url}{path}"
+        if not kwargs.get("timeout"):
+            kwargs["timeout"] = 10
 
-        url_params = {
-            **kwargs,
-        }
-        if not headers:
-            headers = {}
-        if self.bearer_token:
-            headers["Authorization"] = f"Bearer {self.bearer_token}"
-
-        response = requests.get(
-            url=f"{self.base_url}{path}", params=url_params, headers=headers, timeout=10
-        )
+        # pylint: disable-next=missing-timeout
+        response = get(url=url, auth=self.auth, params=params, **kwargs)
         response.raise_for_status()
+        return response
+
+    def json(self, path: str, params: Any = None, **kwargs) -> Any:
+        response = self.get(path, params=params, **kwargs)
         return response.json()
 
-    def get_bm_cs_projects_results(self, project_id, stream_id, headers=None, **kwargs):
-        return self.request(
+    def get_bm_cs_projects_results(
+        self,
+        project_id: int,
+        stream_id: str,
+        params: Optional[GetBmCsProjectsResultsParams] = None,
+        **kwargs,
+    ) -> Any:
+        return self.json(
             path=f"/api/bm-cs-projects/{project_id}/streams/{stream_id}/results",
-            headers=headers,
+            params=params,
             **kwargs,
         ).get("data")

acrclient/models.py

@@ -0,0 +1,16 @@
+from typing import TypedDict
+
+
+class GetBmCsProjectsResultsParams(TypedDict):
+    """Parameters for getting BM projects custom streams results."""
+
+    type: str
+    """last: get the last results, day: get the day results, Default is day"""
+    date: str
+    """Get all the results on this date. The format is YYYYmmdd (E.g. 20210201)"""
+    min_duration: int
+    """Return the results of played_duration >= min_duration seconds (default: 0)"""
+    max_duration: int
+    """Return the results with played_duration <= max_duration seconds (default: 3600)"""
+    isrc_country: str
+    """Only return results that match the isrc country code (E.g. DE, FR, IT, US)"""

tests/test_client.py

@@ -1,24 +1,25 @@
 import requests_mock
 
-import acrclient
+from acrclient import Client
+from acrclient.client import _Auth
 
 
 def test_client():
-    client = acrclient.Client("bearer-token")
-    assert isinstance(client, acrclient.Client)
-    assert client.bearer_token == "bearer-token"
+    client = Client(_Auth("bearer-token"))
+    assert isinstance(client, Client)
+    assert isinstance(client.auth, _Auth)
     assert client.base_url == "https://eu-api-v2.acrcloud.com"
 
 
-def test_request():
-    client = acrclient.Client("bearer-token")
+def test_client_get():
+    client = Client(_Auth("bearer-token"))
     with requests_mock.Mocker() as mock:
         mock.get("https://eu-api-v2.acrcloud.com/foo", json={})
-        client.request("/foo")
+        client.get("/foo")
 
 
-def test_get_bm_cs_projects_results():
-    client = acrclient.Client("bearer-token")
+def test_client_get_bm_cs_projects_results():
+    client = Client(_Auth("bearer-token"))
     with requests_mock.Mocker() as mock:
         mock.get(
             f"{client.base_url}/api/bm-cs-projects/project-id/streams/stream-id/results",