feat: magic eden: parametrize listing/offers max limits

blockapi/v2/api/nft/magic_eden.py

@@ -46,11 +46,21 @@ class MagicEdenApi(BlockchainApi, INftProvider, INftParser):
 
     coin_map = NotImplemented
 
-    def __init__(self, sleep_provider):
+    def __init__(self, sleep_provider, max_listings=500, max_offers=500):
         super().__init__()
-
         self._sleep_provider = sleep_provider
 
+        self.max_offers = max_offers
+        if max_listings > 15000:
+            logger.warning(
+                'Listings cap exceeding 15000 will cause an error: '
+                '"offset should be non-negative integer". '
+                'Setting maximum to 15000.'
+            )
+            self.max_listings = 15000
+        else:
+            self.max_listings = max_listings
+
     def fetch_nfts(self, address: str) -> FetchResult:
         offset = 0
         limit = 500
@@ -192,7 +202,12 @@ def fetch_offers(
                 data_len = len(results)
                 offset += limit
 
-            if data.errors or not data.data or data_len < limit or offset > 15000:
+            if (
+                data.errors
+                or not data.data
+                or data_len < limit
+                or len(items) >= self.max_offers
+            ):
                 return FetchResult(
                     status_code=data.status_code,
                     headers=data.headers,
@@ -257,7 +272,11 @@ def _get_offer_price(self, item: dict) -> str:
         return str(spot_price * (1 - seller_fee - takers_fee - lp_fee))
 
     def fetch_listings(
-        self, collection: str, cursor: Optional[str] = None
+        self,
+        collection: str,
+        cursor: Optional[str] = None,
+        sort="listPrice",
+        sort_direction="desc",
     ) -> FetchResult:
         offset = 0
         limit = 100
@@ -266,7 +285,12 @@ def fetch_listings(
         while True:
             self._sleep_provider.sleep(self.base_url, self.api_options.rate_limit)
             data = self.get_data(
-                'get_listings', slug=collection, offset=offset, limit=limit
+                'get_listings',
+                slug=collection,
+                offset=offset,
+                limit=limit,
+                sort=sort,
+                sort_direction=sort_direction,
             )
 
             if self._should_retry(data):
@@ -277,7 +301,12 @@ def fetch_listings(
                 offset += limit
 
             # note: if offset is greater than 15000, causes response "offset should be non-negative integer"
-            if data.errors or not data.data or len(data.data) < limit or offset > 15000:
+            if (
+                data.errors
+                or not data.data
+                or len(data.data) < limit
+                or len(items) > self.max_listings
+            ):
                 return FetchResult(
                     status_code=data.status_code,
                     headers=data.headers,

blockapi/v2/api/nft/opensea.py

@@ -3,7 +3,6 @@
 from decimal import Decimal
 from typing import Callable, Iterable, Optional, Tuple
 
-from blockapi.utils.num import raw_to_decimals
 from blockapi.v2.base import (
     ApiException,
     BlockchainApi,
@@ -98,6 +97,8 @@ def __init__(
         blockchain: Blockchain,
         sleep_provider: ISleepProvider = None,
         limit: Optional[int] = None,
+        max_listings=500,
+        max_offers=500,
     ):
         super().__init__(api_key)
 
@@ -110,6 +111,9 @@ def __init__(
         self._sleep_provider = sleep_provider or SleepProvider()
         self._limit = limit
 
+        self.max_listings = max_listings
+        self.max_offers = max_offers
+
     def fetch_nfts(self, address: str, cursor: Optional[str] = None) -> FetchResult:
         logger.info(f'Fetch nfts from {address}, cursor={cursor}')
         return self._coalesce(self._yield_nfts(address, cursor))
@@ -256,17 +260,32 @@ def _yield_fetch_data(
         self, fetch_method: Callable, key: str, cursor: Optional[str] = None
     ) -> Iterable[Tuple[FetchResult, Optional[str]]]:
         cursors = set()
+        page_count = 0
+        item_count = 0
 
-        count = 0
         while True:
             self._sleep_provider.sleep(self.base_url, self.api_options.rate_limit)
-            count += 1
-            logger.debug(f'Fetching page {count} of {key} from {cursor}')
+            page_count += 1
+            logger.debug(f'Fetching page {page_count} of {key} from {cursor}')
             fetched, next_cursor = fetch_method(key, cursor)
             if self._should_retry(fetched):
-                count -= 1
+                page_count -= 1
                 continue
 
+            # Count items for dynamic limiting
+            item_limit = None
+            if fetched.data:
+                current_items = 0
+                if 'offers' in fetched.data:
+                    current_items = len(fetched.data['offers'])
+                    item_limit = self.max_offers
+                elif 'listings' in fetched.data:
+                    current_items = len(fetched.data['listings'])
+                    item_limit = self.max_listings
+                # skip `'nfts' in fetched.data`, because it doesn't have a limit
+
+                item_count += current_items
+
             yield fetched, next_cursor
 
             if not next_cursor:
@@ -278,13 +297,19 @@ def _yield_fetch_data(
 
             cursors.add(cursor)
 
-            if self._limit and count >= self._limit:
+            # Check both page and item limits
+            if self._limit and page_count >= self._limit:
+                break
+
+            if item_limit and item_count >= item_limit:
                 break
 
     def _fetch_offers_page(
         self, method: str, collection: str, cursor: Optional[str] = None
     ) -> tuple[FetchResult, Optional[str]]:
-        params = dict(next=cursor) if cursor else dict()
+        params = {'limit': 100}  # the max limit allowed is 100 items per page
+        if cursor:
+            params['next'] = cursor
 
         fetched = self.get_data(
             method,
@@ -573,9 +598,11 @@ def _coalesce(fetch_results: Iterable[Tuple[FetchResult, Optional[str]]]):
         errors = []
         last = None
         last_cursor = None
+
         for item, cursor in fetch_results:
             last_cursor = cursor
             last = item
+
             if item.data:
                 data.append(item.data)