Moving some defintions to a shared library

clients_protocol/__init__.py

@@ -1,2 +1,85 @@
 """# Common HTTP/REST clients interface
 """
+
+import abc
+import logging
+from collections.abc import Sequence
+from typing import Any
+
+import bs4
+import lxml.html
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+API_ERROR_KEY = 'error'
+
+
+class ApiError(RuntimeError):
+    pass
+
+
+class AbstractClient(abc.ABC):
+    SECRETS: Sequence[str] = ()
+
+    def __init__(self, session: requests.Session):
+        assert isinstance(session, requests.Session)
+        self.session = session
+
+    def fetch_or_none(
+        self,
+        url: str,
+        params=None,
+        **kwargs,
+    ) -> requests.Response | None:
+        r = self._fetch(url, params, **kwargs)
+        if r.status_code == 404:
+            return None
+        return r
+
+    def fetch(self, url: str, params=None, **kwargs) -> requests.Response:
+        r = self._fetch(url, params, **kwargs)
+        r.raise_for_status()
+        return r
+
+    def _fetch(self, url: str, params=None, **kwargs) -> requests.Response:
+        method = 'GET'
+        if 'method' in kwargs:
+            method = kwargs['method']
+            del kwargs['method']
+        return self.session.request(
+            method,
+            url,
+            params=params,
+            allow_redirects=True,
+            **kwargs,
+        )
+
+    def fetch_text(self, url: str, params=None, **kwargs) -> str:
+        return self.fetch(url, params, **kwargs).text
+
+    def fetch_lxml_soup(
+        self,
+        url: str,
+        params=None,
+        **kwargs,
+    ) -> None | bs4.BeautifulSoup:
+        text = self.fetch_text(url, params, **kwargs)
+        if text is None:
+            return None
+        return lxml.html.document_fromstring(text)
+
+    def fetch_soup(self, url: str, params=None, **kwargs) -> None | bs4.BeautifulSoup:
+        text = self.fetch_text(url, params, **kwargs)
+        if text is None:
+            return None
+        return bs4.BeautifulSoup(text, 'html.parser')
+
+    def fetch_json(self, url: str, params=None, **kwargs) -> None | dict[str, Any]:
+        response = self.fetch(url, params, **kwargs)
+        loaded_json = response.json()
+        if API_ERROR_KEY in loaded_json:
+            msg = f'Error from endpoint: {loaded_json[API_ERROR_KEY]}'
+            raise ApiError(msg)
+        return loaded_json

clients_protocol/wishlist.py

@@ -0,0 +1,27 @@
+import logging
+from collections.abc import Sequence
+import abc
+import fin_defs
+from typing import Any
+import dataclasses
+
+from . import common
+
+logger = logging.getLogger(__name__)
+
+@dataclasses.dataclass(frozen=True)
+class WishlistItem:
+    """A single wishlished product."""
+
+    product_name: str  # Name of the game/product
+    reference_url: str  # URL to a reference page for the item.
+    image_url: str | None = None  # URL to the product image
+    console_name: str | None = None # Gaming platform/console name
+    reference_price: fin_defs.AssetAmount | None = None # Reference price, if any
+
+
+class WishlistClient(abc.ABC):
+
+    @abc.abstractmethod
+    def get_wishlist(self) -> Sequence[WishlistItem]:
+        pass