feat(api-client): rewrite api-client v2

api-client/libretime_api_client/_client.py

@@ -0,0 +1,79 @@
+from typing import Optional
+from urllib.parse import urljoin
+
+from loguru import logger
+from requests import Response
+from requests import Session as BaseSession
+from requests.adapters import HTTPAdapter
+from requests.exceptions import RequestException
+from urllib3.util import Retry
+
+DEFAULT_TIMEOUT = 5
+
+
+class TimeoutHTTPAdapter(HTTPAdapter):
+    timeout: int = DEFAULT_TIMEOUT
+
+    def __init__(self, *args, **kwargs):
+        if "timeout" in kwargs:
+            self.timeout = kwargs["timeout"]
+            del kwargs["timeout"]
+        super().__init__(*args, **kwargs)
+
+    def send(self, request, *args, **kwargs):
+        if "timeout" not in kwargs:
+            kwargs["timeout"] = self.timeout
+        return super().send(request, *args, **kwargs)
+
+
+class Session(BaseSession):
+    base_url: Optional[str]
+
+    def __init__(self, base_url: Optional[str] = None):
+        super().__init__()
+        self.base_url = base_url
+
+        retry_strategy = Retry(
+            total=5,
+            backoff_factor=2,
+            status_forcelist=[413, 429, 500, 502, 503, 504],
+        )
+
+        adapter = TimeoutHTTPAdapter(max_retries=retry_strategy)
+
+        self.mount("http://", adapter)
+        self.mount("https://", adapter)
+
+    def request(self, method, url, *args, **kwargs):
+        """Send the request after generating the complete URL."""
+        url = self.create_url(url)
+        return super().request(method, url, *args, **kwargs)
+
+    def create_url(self, url):
+        """Create the URL based off this partial path."""
+        return urljoin(self.base_url, url)
+
+
+# pylint: disable=too-few-public-methods
+class AbstractApiClient:
+    session: Session
+    base_url: str
+
+    def __init__(self, base_url: str):
+        self.base_url = base_url
+        self.session = Session(base_url=base_url)
+
+    def _request(
+        self,
+        method,
+        url,
+        **kwargs,
+    ) -> Response:
+        try:
+            response = self.session.request(method, url, **kwargs)
+            response.raise_for_status()
+            return response
+
+        except RequestException as exception:
+            logger.error(exception)
+            raise exception

api-client/libretime_api_client/v2.py

@@ -1,47 +1,33 @@
-###############################################################################
+from ._client import AbstractApiClient, Response
-# This file holds the implementations for all the API clients.
-#
-# If you want to develop a new client, here are some suggestions: Get the fetch
-# methods working first, then the push, then the liquidsoap notifier.  You will
-# probably want to create a script on your server side to automatically
-# schedule a playlist one minute from the current time.
-###############################################################################
-import logging
-
-from ._config import Config
-from ._utils import RequestProvider
-
-LIBRETIME_API_VERSION = "2.0"
 
 
-api_endpoints = {}
+class ApiClient(AbstractApiClient):
+    VERSION = "2.0"
 
-api_endpoints["version_url"] = "version/"
+    def __init__(self, base_url: str, api_key: str):
-api_endpoints["schedule_url"] = "schedule/"
+        super().__init__(base_url=base_url)
-api_endpoints["webstream_url"] = "webstreams/{id}/"
+        self.session.headers.update({"Authorization": f"Api-Key {api_key}"})
-api_endpoints["show_instance_url"] = "show-instances/{id}/"
-api_endpoints["show_url"] = "shows/{id}/"
-api_endpoints["file_url"] = "files/{id}/"
-api_endpoints["file_download_url"] = "files/{id}/download/"
 
+    def get_version(self, **kwargs) -> Response:
+        return self._request("GET", "/api/v2/version", **kwargs)
 
-class ApiClient:
+    def get_show(self, item_id: int, **kwargs) -> Response:
-    API_BASE = "/api/v2"
+        return self._request("GET", f"/api/v2/shows/{item_id}", **kwargs)
 
-    def __init__(self, logger=None, config_path="/etc/libretime/config.yml"):
+    def get_show_instance(self, item_id: int, **kwargs) -> Response:
-        self.logger = logger or logging
+        return self._request("GET", f"/api/v2/show-instances/{item_id}", **kwargs)
 
-        config = Config(filepath=config_path)
+    def list_schedule(self, **kwargs) -> Response:
-        self.base_url = config.general.public_url
+        return self._request("GET", "/api/v2/schedule", **kwargs)
-        self.api_key = config.general.api_key
 
-        self.services = RequestProvider(
+    def get_webstream(self, item_id: int, **kwargs) -> Response:
-            base_url=self.base_url + self.API_BASE,
+        return self._request("GET", f"/api/v2/webstreams/{item_id}", **kwargs)
-            api_key=self.api_key,
-            endpoints=api_endpoints,
-        )
 
-    def update_file(self, file_id, payload):
+    def get_file(self, item_id: int, **kwargs) -> Response:
-        data = self.services.file_url(id=file_id)
+        return self._request("GET", f"/api/v2/files/{item_id}", **kwargs)
-        data.update(payload)
+
-        return self.services.file_url(id=file_id, _put_data=data)
+    def update_file(self, item_id: int, **kwargs) -> Response:
+        return self._request("PATCH", f"/api/v2/files/{item_id}", **kwargs)
+
+    def download_file(self, item_id: int, **kwargs) -> Response:
+        return self._request("GET", f"/api/v2/files/{item_id}/download", **kwargs)

api-client/setup.py

@@ -27,6 +27,7 @@ setup(
     ],
     extras_require={
         "dev": [
+            "requests-mock",
             "types-requests",
             f"libretime-shared @ file://localhost{here.parent / 'shared'}",
         ],

api-client/tests/v2_test.py

@@ -1,29 +1,13 @@
-from pathlib import Path
-
-import pytest
-
 from libretime_api_client.v2 import ApiClient
 
 
-@pytest.fixture()
+def test_api_client(requests_mock):
-def config_filepath(tmp_path: Path):
+    api_client = ApiClient(base_url="http://localhost:8080", api_key="test-key")
-    filepath = tmp_path / "config.yml"
+
-    filepath.write_text(
+    requests_mock.get(
-        """
+        "http://localhost:8080/api/v2/version",
-general:
+        json={"api_version": "2.0.0"},
-  public_url: http://localhost/test
-  api_key: TEST_KEY
-"""
     )
-    return filepath
 
-
+    resp = api_client.get_version()
-def test_api_client(config_filepath):
+    assert resp.json() == {"api_version": "2.0.0"}
-    client = ApiClient(config_path=config_filepath)
-    assert callable(client.services.version_url)
-    assert callable(client.services.schedule_url)
-    assert callable(client.services.webstream_url)
-    assert callable(client.services.show_instance_url)
-    assert callable(client.services.show_url)
-    assert callable(client.services.file_url)
-    assert callable(client.services.file_download_url)