Implemented Assets Endpoint class

Author: dehidehidehi

.gitignore

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

open_sea_v1/__init__.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/__init__.py

@@ -0,0 +1,145 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from requests import Response
+
+from open_sea_v1.endpoints.endpoint_enums import AssetsOrderBy, OpenseaApiEndpoints
+from open_sea_v1.endpoints.endpoint_base_client import OpenSeaClient
+from open_sea_v1.endpoints.endpoint_abc import BaseOpenSeaEndpoint
+from open_sea_v1.responses.asset_obj import Asset
+
+
+@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[Asset]:
+        self._validate_response_property()
+        assets_json = self._response.json()['assets']
+        assets = [Asset(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.list():
+            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_abc.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_assets.py

@@ -0,0 +1,94 @@
+from enum import Enum
+
+OPENSEA_API_V1 = "https://api.opensea.io/api/v1/"
+OPENSEA_LISTINGS_V1 = "https://api.opensea.io/wyvern/v1/"
+
+
+class ExtendedStrEnum(str, Enum):
+
+    @classmethod
+    def list(cls) -> list[str]:
+        return list(map(lambda c: c.value, cls))
+
+
+class OpenseaApiEndpoints(ExtendedStrEnum):
+    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"
+
+
+class AssetsOrderBy(ExtendedStrEnum):
+    TOKEN_ID = "token_id"
+    SALE_DATE = "sale_date"
+    SALE_COUNT = "sale_count"
+    VISITOR_COUNT = "visitor_count"
+    SALE_PRICE = "sale_price"
+
+
+class Asset(ExtendedStrEnum):
+    TOKEN_ID = "token_id"
+    NUM_SALES = "num_sales"
+    BACKGROUND_COLOR = "background_color"
+    IMAGE_URL = "image_url"
+    IMAGE_PREVIEW_URL = "image_preview_url"
+    IMAGE_THUMBNAIL_URL = "image_thumbnail_url"
+    IMAGE_ORIGINAL_URL = "image_original_url"
+    ANIMATION_URL = "animation_url"
+    ANIMATION_ORIGINAL_URL = "animation_original_url"
+    NAME = "name"
+    DESCRIPTION = "description"
+    EXTERNAL_LINK = "external_link"
+    ASSET_CONTRACT_DICT = "asset_contract"
+    PERMALINK = "permalink"
+    COLLECTION_DICT = "collection"
+    DECIMALS = "decimals"
+    TOKEN_METADATA = "token_metadata"
+    OWNER_DICT = "owner"
+    SELL_ORDERS = "sell_orders"
+    CREATOR_DICT = "creator"
+    TRAITS_DICT = "traits"
+    LAST_SALE_DICT = "last_sale"
+    TOP_BID = "top_bid"
+    LISTING_DATE = "listing_date"
+    IS_PRESALE = "is_presale"
+    TRANSFER_FEE_PAYMENT_TOKEN = "transfer_fee_payment_token"
+    TRANSFER_FEE = "transfer_fee"
+
+
+class AssetTraits(ExtendedStrEnum):
+    TRAIT_TYPE = "trait_type"
+    VALUE = "value"
+    DISPLAY_TYPE = "display_type"
+
+
+class AssetContract(ExtendedStrEnum):
+    ADDRESS = "address"
+    NAME = "name"
+    SYMBOL = "symbol"
+    IMAGE_URL = "image_url"
+    DESCRIPTION = "description"
+    EXTERNAL_LINK = "external_link"
+
+
+class AssetOwner(ExtendedStrEnum):
+    ADDRESS = 'address'
+    CONFIG = 'config'
+    PROFILE_IMG_URL = 'profile_img_url'
+    USER = 'user'
+
+
+class AssetLastSale(ExtendedStrEnum):
+    ASSET = 'asset'
+    ASSET_BUNDLE = 'asset_bundle'
+    EVENT_TYPE = 'event_type'
+    EVENT_TIMESTAMP = 'event_timestamp'
+    AUCTION_TYPE = 'auction_type'
+    TOTAL_PRICE = 'total_price'
+    PAYMENT_TOKEN_DICT = 'payment_token'
+    TRANSACTION_DICT = 'transaction'
+    CREATED_DATE = 'created_date'
+    QUANTITY = 'quantity'