Implemented AssetsEndpoint

Author: dehidehidehi

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Pycharm
+.idea/

open_sea_v1/__init__.py

@@ -0,0 +1,4 @@
+"""
+It's called open_sea_v1 as that is how Open Sea refers to their API.
+There could be upcoming API versions which will should then have their own modules.
+"""

open_sea_v1/endpoints/__init__.py

@@ -0,0 +1,6 @@
+"""
+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_assets import _AssetsEndpoint as AssetsEndpoint
+from open_sea_v1.endpoints.endpoint_assets import _AssetsOrderBy as AssetsOrderBy

open_sea_v1/endpoints/endpoint_abc.py

@@ -0,0 +1,36 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from requests import Response
+
+
+class BaseOpenSeaEndpoint(ABC):
+
+    @abstractmethod
+    def get_request(self, url: str, method: str = 'GET', **kwargs) -> Response:
+        """Call to super().get_request passing url and _request_params."""
+
+    @property
+    @abstractmethod
+    def api_key(self) -> Optional[str]:
+        """Optional OpenSea API key"""
+
+    @property
+    @abstractmethod
+    def url(self) -> str:
+        """Endpoint URL"""
+
+    @property
+    @abstractmethod
+    def _request_params(self) -> dict:
+        """Dictionnary of _request_params to pass into the get_request."""
+
+    @property
+    @abstractmethod
+    def validate_request_params(self) -> None:
+        """"""
+
+    @property
+    @abstractmethod
+    def response(self) -> list[dict]:
+        """Parsed JSON dictionnary from HTTP Response."""

open_sea_v1/endpoints/endpoint_assets.py

@@ -0,0 +1,157 @@
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+
+from requests import Response
+
+from open_sea_v1.endpoints.endpoint_urls import OpenseaApiEndpoints
+from open_sea_v1.endpoints.endpoint_client import OpenSeaClient
+from open_sea_v1.endpoints.endpoint_abc import BaseOpenSeaEndpoint
+from open_sea_v1.responses.asset_obj import _AssetResponse
+
+
+class _AssetsOrderBy(str, Enum):
+    """
+    Helper Enum for remembering the possible values for the order_by param of the AssetsEndpoint class.
+    """
+    TOKEN_ID = "token_id"
+    SALE_DATE = "sale_date"
+    SALE_COUNT = "sale_count"
+    VISITOR_COUNT = "visitor_count"
+    SALE_PRICE = "sale_price"
+
+
+@dataclass
+class _AssetsEndpoint(OpenSeaClient, BaseOpenSeaEndpoint):
+    """
+    Opensea API Assets Endpoint
+
+    Parameters
+    ----------
+    width:
+        width of the snake
+
+    owner:
+        The address of the owner of the assets
+
+    token_ids:
+        List of token IDs to search for
+
+    asset_contract_address:
+        The NFT contract address for the assets
+
+    asset_contract_addresses:
+        List of contract addresses to search for. Will return a list of assets with contracts matching any of the addresses in this array. If "token_ids" is also specified, then it will only return assets that match each (address, token_id) pairing, respecting order.
+
+    order_by:
+        How to order the assets returned. By default, the API returns the fastest ordering (contract address and token id). Options you can set are token_id, sale_date (the last sale's transaction's timestamp), sale_count (number of sales), visitor_count (number of unique visitors), and sale_price (the last sale's total_price)
+
+    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
+    asset_contract_address: Optional[list[str]] = None
+    asset_contract_addresses: Optional[str] = None
+    collection: 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()
+        self._response: Optional[Response] = None
+
+    @property
+    def http_response(self):
+        self._validate_response_property()
+        return self._response
+
+    @property
+    def response(self) -> list[_AssetResponse]:
+        self._validate_response_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 = super().get_request(self.url, **self._request_params)
+
+    @property
+    def _request_params(self) -> dict[dict]:
+        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
+        )
+        return dict(api_key=self.api_key, params=params)
+
+    def validate_request_params(self) -> None:
+        self._validate_mandatory_params()
+        self._validate_asset_contract_addresses()
+        self._validate_order_direction()
+        self._validate_order_by()
+        self._validate_limit()
+
+    def _validate_response_property(self):
+        if self._response is None:
+            raise AttributeError('You must call self.request prior to accessing self.response')
+
+    def _validate_mandatory_params(self):
+        mandatory = self.owner, self.token_ids, self.asset_contract_address, self.asset_contract_addresses, self.collection
+        if all((a is None for a in mandatory)):
+            raise ValueError("At least one of the following parameters must not be None:\n"
+                             "owner, token_ids, asset_contract_address, asset_contract_addresses, collection")
+
+    def _validate_asset_contract_addresses(self):
+        if self.asset_contract_address and self.asset_contract_addresses:
+            raise ValueError(
+                "You cannot simultaneously get_request for a single contract_address and a list of contract_addresses."
+            )
+
+        if self.token_ids and not (self.asset_contract_address or self.asset_contract_addresses):
+            raise ValueError(
+                "You cannot query for token_ids without specifying either "
+                "asset_contract_address or asset_contract_addresses."
+            )
+
+    def _validate_order_direction(self):
+        if self.order_direction is None:
+            return
+
+        if self.order_direction not in ['asc', 'desc']:
+            raise ValueError(
+                f"order_direction param value ({self.order_direction}) is invalid. "
+                f"Must be either 'asc' or 'desc', case sensitive."
+            )
+
+    def _validate_order_by(self) -> None:
+        if self.order_by is None:
+            return
+
+        if self.order_by not in (_AssetsOrderBy.TOKEN_ID, _AssetsOrderBy.SALE_COUNT, _AssetsOrderBy.SALE_DATE, _AssetsOrderBy.SALE_PRICE, _AssetsOrderBy.VISITOR_COUNT):
+            raise ValueError(
+                f"order_by param value ({self.order_by}) is invalid. "
+                f"Must be a value from {_AssetsOrderBy.list()}, case sensitive."
+            )
+
+    def _validate_limit(self):
+        if not isinstance(self.limit, int) or not 0 <= self.limit <= 50:
+            raise ValueError(f"limit param must be an int between 0 and 50.")

open_sea_v1/endpoints/endpoint_client.py

@@ -0,0 +1,63 @@
+from requests import Response, request
+
+
+class OpenSeaClient:
+
+    api_key = None
+
+    @property
+    def http_headers(self) -> dict:
+        return {
+            "headers":
+                {"X-API-Key" : self.api_key} if self.api_key else dict(),
+        }
+
+    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')
+        updated_kwargs = kwargs | self.http_headers
+        return request(method, url, **updated_kwargs)
+
+    # 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()
+

open_sea_v1/endpoints/endpoint_urls.py

@@ -0,0 +1,14 @@
+from enum import Enum
+
+OPENSEA_API_V1 = "https://api.opensea.io/api/v1/"
+OPENSEA_LISTINGS_V1 = "https://api.opensea.io/wyvern/v1/"
+
+
+class OpenseaApiEndpoints(str, Enum):
+    ASSET = OPENSEA_API_V1 + "asset"
+    ASSETS = OPENSEA_API_V1 + "assets"
+    ASSET_CONTRACT = OPENSEA_API_V1 + "asset_contract"
+    BUNDLES = OPENSEA_API_V1 + "bundles"
+    EVENTS = OPENSEA_API_V1 + "events"
+    COLLECTIONS = OPENSEA_API_V1 + "collections"
+    LISTINGS = OPENSEA_LISTINGS_V1 + "orders"

open_sea_v1/endpoints/tests/__init__.py

@@ -0,0 +1,74 @@
+from itertools import combinations
+from unittest import TestCase
+
+from open_sea_v1.endpoints.endpoint_assets import _AssetsEndpoint, _AssetsOrderBy
+from open_sea_v1.responses.asset_obj import _AssetResponse
+
+
+class TestAssetsRequest(TestCase):
+    sample_contract = "0xb47e3cd837ddf8e4c57f05d70ab865de6e193bbb"  # punk
+    sample_wallet = "0x5ca12f79e4d33b0bd153b40df59f6db9ee03482e"  # punk
+    default_asset_params = dict(token_ids=[5, 6, 7], asset_contract_address=sample_contract)
+
+    @staticmethod
+    def create_and_get(**kwargs) -> list[_AssetResponse]:
+        """Shortcut"""
+        client = _AssetsEndpoint(**kwargs)
+        client.get_request()
+        return client.response
+
+    def test_cannot_all_be_none_owner_token_ids_asset_contract_address_asset_contract_addresses_collection(self):
+        assets_kwargs = dict(owner=0, token_ids=0, collection=0, asset_contract_address=0, asset_contract_addresses=0)
+        for kwarg_combo in combinations(assets_kwargs, r=len(assets_kwargs)):
+            self.assertRaises(ValueError, _AssetsEndpoint, kwarg_combo)
+
+    def test_param_owner_returns_assets_from_specified_owner(self):
+        params = dict(owner=self.sample_wallet, order_direction='asc', **self.default_asset_params)
+        for punk in self.create_and_get(**params):
+            self.assertEqual(punk.owner.address, self.sample_wallet)
+
+    def test_param_token_ids_raise_exception_if_missing_contract_address_and_addresses(self):
+        self.assertRaises(ValueError, _AssetsEndpoint, token_ids=[1, 2, 3])
+
+    def test_params_cannot_be_simultaneously_be_passed_asset_contract_address_and_contract_addresses(self):
+        params = self.default_asset_params | dict(asset_contract_address=True, asset_contract_addresses=True)
+        self.assertRaises(ValueError, _AssetsEndpoint, **params)
+
+    def test_param_token_ids_returns_assets_corresponding_to_single_contract(self):
+        params = dict(order_direction='asc', **self.default_asset_params)
+        for punk in self.create_and_get(**params):
+            self.assertEqual(punk.asset_contract.address, self.sample_contract)
+
+    def test_param_order_direction_can_only_be_asc_or_desc(self):
+        invalid_order_values = (False, 0, 1, "", [], (), {}, 'hi')
+        for invalid_order in invalid_order_values:
+            params = dict(token_ids=[1], asset_contract_address=self.sample_contract, order_direction=invalid_order)
+            self.assertRaises((ValueError, TypeError), _AssetsEndpoint, **params)
+
+    def test_param_order_by_token_id(self):
+        params = self.default_asset_params | dict(token_ids=[3, 2, 1], order_by=_AssetsOrderBy.TOKEN_ID, order_direction='desc')
+        punks_ids = [punk.token_id for punk in self.create_and_get(**params)]
+        self.assertEqual(['3', '2', '1'], punks_ids)
+
+    def test_param_order_by_sale_date(self):
+        params = self.default_asset_params | dict(token_ids=[1, 14, 33], order_by=_AssetsOrderBy.SALE_DATE)
+        punks_sales = [punk.last_sale.event_timestamp for punk in self.create_and_get(**params)]
+        self.assertEqual(sorted(punks_sales, reverse=True), punks_sales)
+
+    def test_param_order_by_sale_count(self):
+        params = self.default_asset_params | dict(token_ids=[1, 14, 33], order_by=_AssetsOrderBy.SALE_COUNT)
+        punks_sales_cnt = [punk.num_sales for punk in self.create_and_get(**params)]
+        self.assertEqual(sorted(punks_sales_cnt, reverse=True), punks_sales_cnt)
+
+    def test_param_order_by_sale_price(self):
+        params = self.default_asset_params | dict(token_ids=[1, 14, 33], order_by=_AssetsOrderBy.SALE_PRICE)
+        punks_last_sale_price = [punk.last_sale.total_price for punk in self.create_and_get(**params)]
+        self.assertEqual(sorted(punks_last_sale_price, reverse=True), punks_last_sale_price)
+
+    def test_param_order_by_visitor_count(self):
+        pass  # as far as I know this is not returned in the API http_response and done directly by OpenSea
+
+    def test_param_limit_cannot_be_below_0_or_above_50(self):
+        self.assertRaises(ValueError, _AssetsEndpoint, limit="25", **self.default_asset_params)
+        self.assertRaises(ValueError, _AssetsEndpoint, limit=-1, **self.default_asset_params)
+        self.assertRaises(ValueError, _AssetsEndpoint, limit=51, **self.default_asset_params)