Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 1 commit into from
Feb 6, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 39 additions & 18 deletions acrclient/client.py
Original file line number Diff line number Diff line change
@@ -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:
Expand All @@ -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")
16 changes: 16 additions & 0 deletions acrclient/models.py
Original file line number Diff line number Diff line change
@@ -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)"""
Empty file added acrclient/py.typed
Empty file.
19 changes: 10 additions & 9 deletions tests/test_client.py
Original file line number Diff line number Diff line change
@@ -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",
Expand Down