Implemented CollectionsEndpoint.

Added extra doc strings to events and assets endpoints.

Author: dehidehidehi

open_sea_v1/endpoints/assets.py

@@ -22,7 +22,7 @@ class AssetsOrderBy(str, Enum):
 @dataclass
 class AssetsEndpoint(BaseClient, BaseEndpoint):
     """
-    Opensea API Assets Endpoint
+    Opensea API Assets Endpoint.
 
     Parameters
     ----------

open_sea_v1/endpoints/collections.py

@@ -0,0 +1,68 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from open_sea_v1.endpoints.abc import BaseEndpoint
+from open_sea_v1.endpoints.client import BaseClient, ClientParams
+from open_sea_v1.endpoints.urls import EndpointURLS
+from open_sea_v1.responses.collection import CollectionResponse
+
+
+@dataclass
+class CollectionsEndpoint(BaseClient, BaseEndpoint):
+    """
+    From OpenSea documentation:
+    ----------
+    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.
+
+    Maintainer observations:
+    ----------
+    You cannot specify the collection name for the data you wish to retrieve.
+    The only way to retrieve such data, is by getting ALL collections from OpenSea, then
+    iterating over them.
+
+
+    Parameters
+    ----------
+    client_params:
+        ClientParams instance.
+
+    asset_owner: Optional[str]
+        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.
+
+    :return: Parsed JSON
+    """
+
+    client_params: ClientParams = None
+    asset_owner: Optional[str] = None
+
+    def __post_init__(self):
+        self._validate_request_params()
+
+    @property
+    def url(self):
+        return EndpointURLS.COLLECTIONS.value
+
+    @property
+    def parsed_http_response(self) -> list[CollectionResponse]:
+        resp_json = self._http_response.json()
+        collections_json = resp_json if isinstance(resp_json, list) else resp_json['collections']
+        collections = [CollectionResponse(collection_json) for collection_json in collections_json]
+        return collections
+
+    def _get_request(self, **kwargs):
+        params = dict(
+            asset_owner=self.asset_owner,
+            offset=self.client_params.offset,
+            limit=self.client_params.limit,
+        )
+        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:
+        if self.asset_owner is not None and not isinstance(self.asset_owner, str):
+            raise TypeError(f'{self.asset_owner=} must be a str instance representing a wallet address.')

open_sea_v1/endpoints/events.py

@@ -36,7 +36,7 @@ class AuctionType(ExtendedStrEnum):
 @dataclass
 class EventsEndpoint(BaseClient, BaseEndpoint):
     """
-    Opensea API Events Endpoint
+    The events endpoint provides a list of events that occur on the assets that OpenSea tracks.
 
     Parameters
     ----------

open_sea_v1/endpoints/tests/test_collections.py

@@ -0,0 +1,33 @@
+from unittest import TestCase
+
+from open_sea_v1.endpoints.client import ClientParams
+from open_sea_v1.endpoints.collections import CollectionsEndpoint
+from open_sea_v1.responses.collection import CollectionResponse
+
+
+class TestCollectionsEndpoint(TestCase):
+
+    def setUp(self) -> None:
+        self.asset_owner = "0x5ca12f79e4d33b0bd153b40df59f6db9ee03482e"  # punk
+        self.endpoint_kwargs = dict(
+            client_params=ClientParams(limit=2, max_pages=1),
+        )
+
+    @staticmethod
+    def create_and_get(**kwargs) -> list[CollectionResponse]:
+        collections_client = CollectionsEndpoint(**kwargs)
+        collections_client._get_request()
+        return collections_client.parsed_http_response
+
+    def test_collection_resp_works(self):
+        basic_collections_resp = self.create_and_get(**self.endpoint_kwargs)
+        self.assertTrue(basic_collections_resp)
+
+        self.endpoint_kwargs |= dict(asset_owner=self.asset_owner)
+        owner_asset_collections_resp = self.create_and_get(**self.endpoint_kwargs)
+        self.assertTrue(owner_asset_collections_resp)
+
+        self.assertNotEqual(
+            {c.slug for c in basic_collections_resp},
+            {c.slug for c in owner_asset_collections_resp},
+        )