Added PRE_ORDER

This commit was automatically generated by [a script](https://gitfub.space/Jmaa/repo-manager)

README.md

@@ -14,7 +14,23 @@
 
 This project requires [Python](https://www.python.org/) 3.8 or newer.
 
-This project does not have any library requirements 😎
+All required libraries can be installed easily using:
+
+```bash
+pip install -r requirements.txt
+```
+
+Full list of requirements:
+- [beautifulsoup4](https://pypi.org/project/beautifulsoup4/)
+- [cssselect](https://pypi.org/project/cssselect/)
+- [fin-defs](https://gitfub.space/Jmaa/fin-defs)
+- [lxml](https://pypi.org/project/lxml/)
+- [requests](https://pypi.org/project/requests/)
+- [frozendict](https://pypi.org/project/frozendict/)
+- [requests-ratelimiter](https://pypi.org/project/requests-ratelimiter/)
+- [requests-util](https://gitfub.space/Jmaa/requests_util)
+- [requests_cache](https://pypi.org/project/requests_cache/)
+
 
 ## Contributing
 

clients_protocol/__init__.py

@@ -1,5 +1,6 @@
-"""# Common HTTP/REST clients interface
-"""
+"""# Common HTTP/REST clients interface"""
+
+import urllib.parse
 
 import abc
 import logging
@@ -10,6 +11,10 @@ import bs4
 import lxml.html
 import requests
 
+from ._version import __version__
+
+__all__ = ['__version__']
+
 logger = logging.getLogger(__name__)
 
 
@@ -30,54 +35,64 @@ class AbstractClient(abc.ABC):
     def fetch_or_none(
         self,
         url: str,
-        params=None,
         **kwargs,
     ) -> requests.Response | None:
-        r = self._fetch(url, params, **kwargs)
+        r = self._fetch(url, **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)
+    def fetch(self, url: str, **kwargs) -> requests.Response:
+        r = self._fetch(url, **kwargs)
+        if r.status_code in {301, 302, 303}:
+            msg = (
+                f'Redirection: {r.request.method} {url} -> GET {r.headers["Location"]}'
+            )
+            raise Exception(msg)
         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']
+    def _fetch(self, url: str, **kwargs) -> requests.Response:
+        kwargs.setdefault('method', 'GET')
+        kwargs.setdefault('allow_redirects', True)
+
+        url_parsed = urllib.parse.urlparse(url)
+        origin_url = url_parsed._replace(
+            path='', params='', query='', fragment=''
+        ).geturl()
+
+        kwargs.setdefault('headers', {}).setdefault('Origin', origin_url)
+        kwargs.setdefault('headers', {}).setdefault('Alt-Used', url_parsed.hostname)
         return self.session.request(
-            method,
-            url,
-            params=params,
-            allow_redirects=True,
+            url=url,
             **kwargs,
         )
 
-    def fetch_text(self, url: str, params=None, **kwargs) -> str:
-        return self.fetch(url, params, **kwargs).text
+    def fetch_text(self, url: str, **kwargs) -> str:
+        return self.fetch(url, **kwargs).text
 
     def fetch_lxml_soup(
         self,
         url: str,
-        params=None,
         **kwargs,
     ) -> None | bs4.BeautifulSoup:
-        text = self.fetch_text(url, params, **kwargs)
+        kwargs.setdefault('headers', {}).setdefault('Accept', 'text/html')
+        text = self.fetch_text(url, **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)
+    def fetch_soup(self, url: str, **kwargs) -> None | bs4.BeautifulSoup:
+        kwargs.setdefault('headers', {}).setdefault('Accept', 'text/html')
+        text = self.fetch_text(url, **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)
+    def fetch_json(self, url: str, **kwargs) -> None | dict[str, Any]:
+        kwargs.setdefault('headers', {}).setdefault('Accept', 'application/json')
+
+        response = self.fetch(url=url, **kwargs)
         loaded_json = response.json()
         if API_ERROR_KEY in loaded_json:
             msg = f'Error from endpoint: {loaded_json[API_ERROR_KEY]}'

clients_protocol/_version.py

@@ -1,5 +1 @@
-# WARNING!
-# THIS IS AN AUTOGENERATED FILE!
-# MANUAL CHANGES CAN AND WILL BE OVERWRITTEN!
-
-__version__ = '0.1.0'
+__version__ = '0.1.5'

clients_protocol/autoload.py

@@ -0,0 +1,68 @@
+import abc
+import functools
+import importlib
+import logging
+from collections.abc import Iterator
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+def load_module_with_name(name: str) -> object | None:
+    try:
+        module = importlib.import_module(name)
+    except Exception:
+        logger.exception('Module %s failed to load!', name)
+        return None
+    logger.info('Loaded module: %s', name)
+    return module
+
+
+def is_loadable_python(f: Path) -> bool:
+    if f.is_file() and f.suffix == '.py':
+        return True
+    if f.is_dir():
+        return is_loadable_python(f / '__init__.py')
+    return False
+
+
+def get_modules_in_directory(module_directory: Path) -> Iterator[str]:
+    for f in module_directory.iterdir():
+        name = f.parts[-1].removesuffix('.py')
+        if is_loadable_python(f):
+            if name != '__init__':
+                yield name
+
+
+@functools.cache
+def load_submodules(module_name: str, index_module_path: str | Path) -> None:
+    """Loads the submodules for the given module name.
+
+    This method is not recursive, as it is the submodule's responsibility to
+    manage it's own submodules. That said: The load_submodules system works
+    when invoked from an auto-loaded package.
+    """
+    module_directory = Path(index_module_path)
+    submodules = list(get_modules_in_directory(module_directory))
+    logger.info(
+        'Detected %d submodules for %s',
+        len(submodules),
+        module_name,
+    )
+    for submodule_name in submodules:
+        load_module_with_name(f'{module_name}.{submodule_name}')
+
+
+def _find_subclasses(class_: type, class_accum: list[type]) -> None:
+    is_abstract = abc.ABC in class_.__bases__
+
+    if not is_abstract:
+        class_accum.append(class_)
+    for subclass in class_.__subclasses__():
+        _find_subclasses(subclass, class_accum)
+
+
+def find_subclasses(abstract_class: type) -> list[type]:
+    subclasses = []
+    _find_subclasses(abstract_class, subclasses)
+    return subclasses

clients_protocol/stores.py

@@ -0,0 +1,188 @@
+"""Data definitions for store clients."""
+
+from collections.abc import Iterator, Sequence
+from fin_defs import AssetAmount
+from frozendict import frozendict
+import abc
+import dataclasses
+import datetime
+import enum
+import logging
+
+from . import AbstractClient
+
+DEFAULT_MAX_RESULTS = 1
+
+URL = str
+
+
+@dataclasses.dataclass(frozen=True, order=True, slots=True)
+class SellerInfo:
+    name: str
+    url: str
+    address: str
+    country_alpha2: str
+    company_registration_number: int | None = None
+    phone_number: str | None = None
+    email_address: str | None = None
+    validated: bool | None = None
+    creation_date: datetime.date | None = None
+
+    def __post_init__(self):
+        assert self.name.strip() == self.name
+        assert self.name != ''
+        assert self.url.strip() == self.url
+        assert self.address.strip() == self.address
+        assert len(self.country_alpha2) == 2, 'Invalid ISO 3166-1 alpha-2 country code'
+
+
+class StoreOfferProperty(enum.Enum):
+    AUCTION = enum.auto()
+    SOLD_OUT = enum.auto()
+    DISCOUNT = enum.auto()
+    PRE_ORDER = enum.auto()
+
+
+@dataclasses.dataclass(frozen=True, order=True, slots=True)
+class StoreOffer:
+    title: str
+    description: str
+    listed_price: AssetAmount
+    source: str
+    image_sources: Sequence[str] = ()
+    seller: SellerInfo | None = None
+    date_published: datetime.datetime | None = None
+    date_updated: datetime.datetime | None = None
+    properties: frozenset[StoreOfferProperty] = dataclasses.field(
+        default_factory=frozenset,
+    )
+    metadata: frozendict[str, str] = dataclasses.field(default_factory=frozendict)
+    price_shipping_estimate: AssetAmount | None = None
+
+    def __post_init__(self):
+        assert self.title.strip() == self.title, self.title
+        # assert '\n' not in self.title, self.title
+        assert self.description is not None
+        assert self.source.strip() == self.source, self.source
+        assert None not in self.metadata
+        assert None not in self.image_sources
+        assert isinstance(self.metadata, frozendict), 'Metadata must be a frozendict'
+        assert isinstance(self.image_sources, tuple), 'image_sources must be a tuple'
+        assert (
+            isinstance(self.listed_price, AssetAmount) or self.listed_price is None
+        ), f'listed_price must be an AssetAmount, but was {self.listed_price}'
+        assert (
+            isinstance(self.price_shipping_estimate, AssetAmount)
+            or self.price_shipping_estimate is None
+        ), (
+            f'price_shipping_estimate must be an AssetAmount, but was {self.price_shipping_estimate}'
+        )
+
+    @property
+    def is_auction(self) -> bool:
+        return StoreOfferProperty.AUCTION in self.properties
+
+
+ALLOWED_CATEGORIES = {None, 'video-games'}
+ALLOWED_REGION_NAMES = frozenset({None, 'japan', 'pal', 'ntsc', 'english asian'})
+
+
+@dataclasses.dataclass(frozen=True, order=True, slots=True)
+class Product:
+    name: str
+    category: str
+    platform: str | None = None
+    edition: str | None = None
+    region: str | None = None
+
+    def __post_init__(self):
+        assert self.category in ALLOWED_CATEGORIES, f'Unknown category: {self.category}'
+        assert self.region in ALLOWED_REGION_NAMES, f'Unknown region: {self.region}'
+
+    @staticmethod
+    def videogame(name, **kwargs):
+        return Product(name, category='video-games', **kwargs)
+
+
+@dataclasses.dataclass(frozen=True)
+class ProductPrices:
+    """Reference prices for a specific product.
+
+    Fields:
+    - product: Product that this object describes prices for.
+    - loose: Price for the product itself without any packaging.
+    - complete: Price for the product with all of its packaging and in-box extracts.
+    - new: Price of an unopened sealed product, with all of its packaging and in-box extracts.
+    - graded: Price of an unopened sealed product that have been graded by a professional.
+    - source: URL indicating the source of the pricing information.
+    """
+
+    product: Product
+    loose: AssetAmount
+    complete: AssetAmount
+    new: AssetAmount
+    graded: AssetAmount
+    source: str
+
+
+class ItemSaleStatus(enum.Enum):
+    SOLD = enum.auto()
+
+
+class OfferNoLongerAvailableError(Exception):
+    pass
+
+
+class StoreOfferAccessClient(AbstractClient, abc.ABC):
+    """Client for accessing specific store offers on specific websites."""
+
+    @abc.abstractmethod
+    def supports_url(self, url: URL) -> bool:
+        """Check if this client supports the given URL.
+
+        This method determines whether the provided URL is supported by this
+        client.
+
+        Args:
+            url: The URL to check for support.
+
+        Returns:
+            True if the URL is supported, False otherwise.
+
+        Performance:
+            Lightweight. Implementations should not send any HTTP requests.
+        """
+
+    @abc.abstractmethod
+    def get_store_offer(self, url: URL) -> StoreOffer:
+        """Downloads the description of the store offer."""
+
+
+class StoreOfferSearchClient(AbstractClient, abc.ABC):
+    """Client for searching for store offers on specific websites."""
+
+    @abc.abstractmethod
+    def search_for_games(
+        self,
+        max_results: int = DEFAULT_MAX_RESULTS,
+    ) -> Iterator[StoreOffer]:
+        """Searches for all games offers on the respective website."""
+
+    @abc.abstractmethod
+    def search_for_product(
+        self,
+        search_term: str,
+        max_results: int = DEFAULT_MAX_RESULTS,
+    ) -> Iterator[StoreOffer]:
+        """Searches for the given named product."""
+
+
+class ReferencePriceClient(AbstractClient, abc.ABC):
+    """Client for accessing reference prices for specific products."""
+
+    @abc.abstractmethod
+    def get_prices(
+        self,
+        product: Product,
+    ) -> ProductPrices | None:
+        """Gets reference prices for the given product."""

clients_protocol/wishlist.py

@@ -7,6 +7,7 @@ import dataclasses
 
 logger = logging.getLogger(__name__)
 
+
 @dataclasses.dataclass(frozen=True)
 class WishlistItem:
     """A single wishlished product."""
@@ -14,12 +15,11 @@ class WishlistItem:
     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
+    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

requirements.txt

@@ -0,0 +1,9 @@
+beautifulsoup4
+cssselect
+fin-defs @ git+https://gitfub.space/Jmaa/fin-defs.git
+lxml
+requests
+frozendict
+requests-ratelimiter
+requests-util @ git+https://gitfub.space/Jmaa/requests_util.git
+requests_cache

setup.py

@@ -10,8 +10,7 @@ from setuptools import setup
 PACKAGE_NAME = 'clients_protocol'
 
 PACKAGE_DESCRIPTION = """
-# Common HTTP/REST clients interface
-""".strip()
+# Common HTTP/REST clients interface""".strip()
 
 PACKAGE_DESCRIPTION_SHORT = """
 """.strip()
@@ -27,11 +26,9 @@ def parse_version_file(text: str) -> str:
 
 
 def find_python_packages() -> list[str]:
-    """
-    Find all python packages. (Directories containing __init__.py files.)
-    """
+    """Find all python packages (directories containing __init__.py files)."""
     root_path = Path(PACKAGE_NAME)
-    packages: set[str] = set([PACKAGE_NAME])
+    packages: set[str] = {PACKAGE_NAME}
 
     # Search recursively
     for init_file in root_path.rglob('__init__.py'):
@@ -39,11 +36,22 @@ def find_python_packages() -> list[str]:
 
     return sorted(packages)
 
+
 with open(PACKAGE_NAME + '/_version.py') as f:
     version = parse_version_file(f.read())
 
 
-REQUIREMENTS_MAIN = []
+REQUIREMENTS_MAIN = [
+    'beautifulsoup4',
+    'cssselect',
+    'fin-defs @ git+https://gitfub.space/Jmaa/fin-defs.git',
+    'lxml',
+    'requests',
+    'frozendict',
+    'requests-ratelimiter',
+    'requests-util @ git+https://gitfub.space/Jmaa/requests_util.git',
+    'requests_cache',
+]
 REQUIREMENTS_TEST = []
 
 

test/test_init.py

@@ -0,0 +1,3 @@
+def test_init():
+    import clients_protocol
+    import clients_protocol.autoload