Implemented Endpoint paginator methods, ResponseParser helper class, and refactored code smells.

Author: dehidehidehi

open_sea_v1/endpoints/__init__.py

@@ -2,9 +2,12 @@
 Exposes classes and objects meant to be used by You!
 Import other modules at your own risk, as their location may change.
 """
+from open_sea_v1.endpoints.endpoint_client import _ClientParams as ClientParams
+
 from open_sea_v1.endpoints.endpoint_assets import _AssetsEndpoint as AssetsEndpoint
 from open_sea_v1.endpoints.endpoint_assets import _AssetsOrderBy as AssetsOrderBy
 
 from open_sea_v1.endpoints.endpoint_events import _EventsEndpoint as EventsEndpoint
 from open_sea_v1.endpoints.endpoint_events import AuctionType
 from open_sea_v1.endpoints.endpoint_events import EventType
+

open_sea_v1/endpoints/endpoint_abc.py

@@ -1,9 +1,10 @@
 from abc import ABC, abstractmethod
-from typing import Optional, Union
+from typing import Optional, Union, Generator
 
 from requests import Response
 
-from open_sea_v1.responses.response__base import _OpenSeaAPIResponse
+from open_sea_v1.endpoints.endpoint_client import _ClientParams
+from open_sea_v1.responses.response_abc import _OpenSeaResponse
 
 
 class BaseOpenSeaEndpoint(ABC):
@@ -15,8 +16,8 @@ def __post_init__(self):
 
     @property
     @abstractmethod
-    def api_key(self) -> Optional[str]:
-        """Optional OpenSea API key"""
+    def client_params(self) -> _ClientParams:
+        """Instance of common OpenSea Endpoint parameters."""
 
     @property
     @abstractmethod
@@ -25,25 +26,19 @@ def url(self) -> str:
 
     @property
     @abstractmethod
-    def _request_params(self) -> dict:
-        """Dictionnary of _request_params to pass into the _get_request."""
+    def get_pages(self) -> Generator[list[list[_OpenSeaResponse]], None, None]:
+        """Returns all pages for the query."""
 
     @property
     @abstractmethod
-    def _validate_request_params(self) -> None:
-        """"""
-
-    @property
-    @abstractmethod
-    def response(self) -> Union[list[_OpenSeaAPIResponse], _OpenSeaAPIResponse]:
+    def parsed_http_response(self) -> Union[list[_OpenSeaResponse], _OpenSeaResponse]:
         """Parsed JSON dictionnary from HTTP Response."""
 
-    @property
     @abstractmethod
-    def http_response(self) -> Optional[Response]:
-        """HTTP Response from Opensea API."""
+    def _get_request(self) -> Response:
+        """Returns HTTP parsed_http_response from OpenSea."""
 
+    @property
     @abstractmethod
-    def get_request(self, url: str, method: str = 'GET', **kwargs) -> Response:
-        """Call to super()._get_request passing url and _request_params."""
-
+    def _validate_request_params(self) -> None:
+        """"""

open_sea_v1/endpoints/endpoint_assets.py

@@ -1,11 +1,11 @@
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional
+from typing import Optional, Generator
 
 from open_sea_v1.endpoints.endpoint_abc import BaseOpenSeaEndpoint
-from open_sea_v1.endpoints.endpoint_client import OpenSeaClient
+from open_sea_v1.endpoints.endpoint_client import BaseOpenSeaClient, _ClientParams
 from open_sea_v1.endpoints.endpoint_urls import OpenseaApiEndpoints
-from open_sea_v1.responses.response_asset import _AssetResponse
+from open_sea_v1.responses import AssetResponse
 
 
 class _AssetsOrderBy(str, Enum):
@@ -20,12 +20,14 @@ class _AssetsOrderBy(str, Enum):
 
 
 @dataclass
-class _AssetsEndpoint(OpenSeaClient, BaseOpenSeaEndpoint):
+class _AssetsEndpoint(BaseOpenSeaClient, BaseOpenSeaEndpoint):
     """
     Opensea API Assets Endpoint
 
     Parameters
     ----------
+    client_params:
+        Common endpoint params.
 
     owner:
         The address of the owner of the assets
@@ -45,53 +47,48 @@ class _AssetsEndpoint(OpenSeaClient, BaseOpenSeaEndpoint):
     order_direction:
         Can be asc for ascending or desc for descending
 
-    offset:
-        Offset
-
-    limit:
-        Defaults to 20, capped at 50.
-
     collection:
         Limit responses to members of a collection. Case sensitive and must match the collection slug exactly. Will return all assets from all contracts in a collection.
 
     :return: Parsed JSON
     """
-    api_key: Optional[str] = None
-    owner: Optional[str] = None
-    token_ids: Optional[list[int]] = None
+    client_params: _ClientParams = None
     asset_contract_address: Optional[list[str]] = None
     asset_contract_addresses: Optional[str] = None
+    token_ids: Optional[list[int]] = None
     collection: Optional[str] = None
+    owner: Optional[str] = None
     order_by: Optional[_AssetsOrderBy] = None
     order_direction: str = None
-    offset: int = 0
-    limit: int = 20
 
     def __post_init__(self):
         self._validate_request_params()
 
-    @property
-    def response(self) -> list[_AssetResponse]:
-        self._assert_get_request_was_called_before_accessing_this_property()
-        assets_json = self._response.json()['assets']
-        assets = [_AssetResponse(asset_json) for asset_json in assets_json]
-        return assets
-
     @property
     def url(self):
         return OpenseaApiEndpoints.ASSETS.value
 
-    def get_request(self, *args, **kwargs):
-        self._response = self._get_request(self.url, **self._request_params)
-
     @property
-    def _request_params(self) -> dict[dict]:
+    def parsed_http_response(self) -> list[AssetResponse]:
+        assets_json = self._http_response.json()['assets']
+        assets = [AssetResponse(asset_json) for asset_json in assets_json]
+        return assets
+
+    def _get_request(self, **kwargs):
         params = dict(
-            owner=self.owner, token_ids=self.token_ids, asset_contract_address=self.asset_contract_address,
-            asset_contract_addresses=self.asset_contract_addresses, collection=self.collection,
-            order_by=self.order_by, order_direction=self.order_direction, offset=self.offset, limit=self.limit
+            owner=self.owner,
+            token_ids=self.token_ids,
+            asset_contract_address=self.asset_contract_address,
+            asset_contract_addresses=self.asset_contract_addresses,
+            collection=self.collection,
+            order_by=self.order_by,
+            order_direction=self.order_direction,
+            offset=self.client_params.offset,
+            limit=self.client_params.limit,
         )
-        return dict(api_key=self.api_key, params=params)
+        get_request_kwargs = dict(params=params)
+        self._http_response = super()._get_request(**get_request_kwargs)
+        return self._http_response
 
     def _validate_request_params(self) -> None:
         self._validate_mandatory_params()
@@ -139,5 +136,5 @@ def _validate_order_by(self) -> None:
             )
 
     def _validate_limit(self):
-        if not isinstance(self.limit, int) or not 0 <= self.limit <= 50:
+        if not isinstance(self.client_params.limit, int) or not 0 <= self.client_params.limit <= 50:
             raise ValueError(f"limit param must be an int between 0 and 50.")

open_sea_v1/endpoints/endpoint_client.py

@@ -1,75 +1,52 @@
-from typing import Optional
+from abc import ABC
+from dataclasses import dataclass
+from typing import Optional, Generator
 
 from requests import Response, request
 
+from open_sea_v1.responses import OpenSeaAPIResponse
 
-class OpenSeaClient:
+@dataclass
+class _ClientParams:
+    """Common OpenSea Endpoint parameters to pass in."""
+    offset: int = 0
+    limit: int = 20
+    max_pages: Optional[int] = None
+    api_key: Optional[str] = None
 
-    api_key = None
-    _response: Optional[Response] = None
+
+class BaseOpenSeaClient(ABC):
+    client_params: _ClientParams
+    processed_pages: int = 0
+    response = None
+    parsed_http_response = None
+    url = None
+    _http_response = None
 
     @property
     def http_headers(self) -> dict:
-        return {
-            "headers":
-                {"X-API-Key" : self.api_key} if self.api_key else dict(),
-        }
+        params = {'headers': dict()}
+        if self.client_params.api_key:
+            params['headers'] = {'X-API-Key': self.client_params.api_key}
+        return params
 
-    def _get_request(self, url: str, method: str = 'GET', **kwargs) -> Response:
-        """
-        Automatically passes in API key in HTTP _get_request headers.
-        """
-        if 'api_key' in kwargs:
-            self.api_key = kwargs.pop('api_key')
+    def _get_request(self, **kwargs) -> Response:
         updated_kwargs = kwargs | self.http_headers
-        return request(method, url, **updated_kwargs)
-
-    @property
-    def http_response(self):
-        self._assert_get_request_was_called_before_accessing_this_property()
-        return self._response
-
-    def _assert_get_request_was_called_before_accessing_this_property(self):
-        if self._response is None:
-            raise AttributeError('You must call self.request prior to accessing self.response')
-
-    # def collections(self, *, asset_owner: Optional[str] = None, offset: int, limit: int) -> OpenseaCollections:
-    #     """
-    #     Use this endpoint to fetch collections and dapps that OpenSea shows on opensea.io,
-    #     along with dapps and smart contracts that a particular user cares about.
-    #
-    #     :param asset_owner: A wallet address. If specified, will return collections where
-    #      the owner owns at least one asset belonging to smart contracts in the collection.
-    #       The number of assets the account owns is shown as owned_asset_count for each collection.
-    #     :param offset: For pagination. Number of contracts offset from the beginning of the result list.
-    #     :param limit: For pagination. Maximum number of contracts to return.
-    #     :return: Parsed JSON
-    #     """
-    #     if offset != 0:
-    #         raise NotImplementedError(
-    #             "Sorry, tested offset parameter is not implemented yet. "
-    #             "Feel free to PR after looking at the tests and trying to understand"
-    #             " why current implementation doesn't allow pagination to work..."
-    #         )
-    #     resp = self._collections(asset_owner=asset_owner, offset=offset, limit=limit)
-    #     return resp.json()['collections']
-    #
-    # def _collections(self, **_request_params) -> Response:
-    #     """Returns HTTPResponse object."""
-    #     url = OpenseaApiEndpoints.COLLECTIONS.value
-    #     return self._get_request("GET", url, _request_params=_request_params)
-    #
-    # def asset(self, asset_contract_address: str, token_id: str, account_address: Optional[str] = None) -> OpenseaAsset:
-    #     """
-    #     :param asset_contract_address: Address of the contract for this NFT
-    #     :param token_id: Token ID for this item
-    #     :param account_address: Address of an owner of the token. If you include this, the http_response will include an ownership object that includes the number of tokens owned by the address provided instead of the top_ownerships object included in the standard http_response, which provides the number of tokens owned by each of the 10 addresses with the greatest supply of the token.
-    #     :return: Parsed JSON.
-    #     """
-    #     resp = self._asset(
-    #         asset_contract_address=asset_contract_address,
-    #         token_id=token_id,
-    #         account_address=account_address,
-    #     )
-    #     return resp.response()
-
+        return request('GET', self.url, **updated_kwargs)
+
+    def get_pages(self) -> Generator[list[list[OpenSeaAPIResponse]], None, None]:
+        self._http_response = None
+        while self.remaining_pages():
+            self._http_response = self._get_request()
+            yield self.parsed_http_response
+            self.client_params.offset += self.client_params.limit
+            self.processed_pages += 1
+
+    def remaining_pages(self) -> bool:
+        if self._http_response is None:
+            return True
+        if self.client_params.max_pages is not None and self.processed_pages <= self.client_params.max_pages:
+            return True
+        if len(self.response) >= self.client_params.offset:
+            return True
+        return False