opensea: filter listings by original address

integrations/center.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from typing import Any, Dict, Generator, List, Optional, Union
+import traceback
 
 from urllib.parse import urlencode
 import requests
@@ -192,7 +193,11 @@ def fetch_nft_search(search_str: str) -> Generator[Union[NFTCollection, NFTAsset
         url = f"{API_URL}/{network}/search?{q}"
         timing.log('search_begin')
         response = requests.get(url, headers=HEADERS)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except Exception:
+            traceback.print_exc()
+            break
         timing.log('search_done')
         obj = response.json()
         for r in obj['results']:
@@ -264,7 +269,11 @@ def fetch_nft_search_collection_by_trait(network: str, address: str, trait_name:
         ))
         url = f"{API_URL}/{network}/{address}/assets/searchByTraits?{q}"
         response = requests.post(url, headers=headers, json=payload)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except Exception:
+            traceback.print_exc()
+            break
         timing.log('search_done')
         obj = response.json()
         if not obj['items']:
@@ -331,7 +340,11 @@ def fetch_nft_collection_assets(network: str, address: str) -> NFTCollectionAsse
         ]}
         url = f"{API_URL}/{network}/assets"
         response = requests.post(url, headers=HEADERS, json=payload)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except Exception:
+            traceback.print_exc()
+            break
         obj = response.json()
         for item in obj:
             if not item:
@@ -389,7 +402,11 @@ def fetch_nft_collection_assets_for_sale(network: str, address: str, _skip_timin
         ]}
         url = f"{API_URL}/{network}/assets"
         response = requests.post(url, headers=HEADERS, json=payload)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except Exception:
+            traceback.print_exc()
+            break
         if not _skip_timing:
             timing.log('search_done')
         obj = response.json()
@@ -436,7 +453,11 @@ def fetch_nft_collection_traits(network: str, address: str) -> NFTCollectionTrai
         ))
         url = f"{API_URL}/{network}/{address}/traits?{q}"
         response = requests.get(url, headers=HEADERS)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except Exception:
+            traceback.print_exc()
+            break
         obj = response.json()
         for item in obj['items']:
             total = 0
@@ -475,7 +496,11 @@ def fetch_nft_collection_trait_values(network: str, address: str, trait: str) ->
         ))
         url = f"{API_URL}/{network}/{address}/traits/{trait}?{q}"
         response = requests.get(url, headers=HEADERS)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except Exception:
+            traceback.print_exc()
+            break
         obj = response.json()
         total = 0
         for item in obj['items']:

integrations/opensea.py

@@ -53,9 +53,16 @@ class NFTListing:
 def fetch_contract(address: str) -> NFTContract:
     """Fetch data about a contract (collection)."""
     url = f"{API_V1_URL}/asset_contract/{address}"
-    response = requests.get(url, headers=HEADERS)
-    response.raise_for_status()
-    obj = response.json()
+
+    def _exec_request():
+        response = requests.get(url, headers=HEADERS)
+        response.raise_for_status()
+        return response.json()
+
+    obj = retry_on_exceptions_with_backoff(
+        _exec_request,
+        [ErrorToRetry(requests.exceptions.HTTPError, _should_retry_exception)],
+    )
     collection = obj["collection"]
     return NFTContract(
         chain=obj["chain_identifier"],
@@ -82,9 +89,16 @@ def fetch_listings(address: str, token_id: str) -> List[NFTListing]:
             **(dict(next=next_cursor) if next_cursor else {})
         ))
         url = f"{API_V2_URL}/orders/{chain}/seaport/listings?{q}"
-        response = requests.get(url, headers=HEADERS)
-        response.raise_for_status()
-        obj = response.json()
+
+        def _exec_request():
+            response = requests.get(url, headers=HEADERS)
+            response.raise_for_status()
+            return response.json()
+
+        obj = retry_on_exceptions_with_backoff(
+            _exec_request,
+            [ErrorToRetry(requests.exceptions.HTTPError, _should_retry_exception)],
+        )
         for item in obj['orders']:
             offer = item["protocol_data"]["parameters"]["offer"][0]
             price_value = int(item["current_price"])
@@ -104,25 +118,40 @@ def fetch_listings(address: str, token_id: str) -> List[NFTListing]:
     return ret
 
 
-def fetch_all_listings(slug: str) -> List[NFTListing]:
+def fetch_all_listings(address: str) -> List[NFTListing]:
     """Fetch all listings for a collection."""
     # NOTE: a given token ID might have more than one listing
+    contract = fetch_contract(address)
+    slug = contract.slug
     limit = PAGE_LIMIT
     next_cursor = None
     ret = []
-    # Arbitary limit to optimize for latency, based on hueristics related to observed number of NFTs listed for blue-chip collections. 
+    # Arbitary limit to optimize for latency, based on hueristics related to observed number of NFTs listed for blue-chip collections.
     max_results = 300
-    while len(ret) < max_results:
+    max_queries = 5
+    queries = 0
+    while len(ret) < max_results and queries < max_queries:
+        queries += 1
         q = urlencode(dict(
             limit=limit,
             **(dict(next=next_cursor) if next_cursor else {})
         ))
         url = f"{API_V2_URL}/listings/collection/{slug}/all?{q}"
-        response = requests.get(url, headers=HEADERS)
-        response.raise_for_status()
-        obj = response.json()
+
+        def _exec_request():
+            response = requests.get(url, headers=HEADERS)
+            response.raise_for_status()
+            return response.json()
+
+        obj = retry_on_exceptions_with_backoff(
+            _exec_request,
+            [ErrorToRetry(requests.exceptions.HTTPError, _should_retry_exception)],
+        )
         for item in obj['listings']:
             offer = item["protocol_data"]["parameters"]["offer"][0]
+            item_address = offer["token"]
+            if item_address != address:
+                continue
             current_price = item["price"]["current"]
             currency = current_price["currency"]
             price_value = int(current_price['value'])
@@ -132,7 +161,7 @@ def fetch_all_listings(slug: str) -> List[NFTListing]:
                 price_str = f"{price_value / 10 ** current_price['decimals']} {currency}"
             listing = NFTListing(
                 chain=item["chain"],
-                address=offer["token"],
+                address=item_address,
                 token_id=offer["identifierOrCriteria"],
                 price_str=price_str,
                 price_value=price_value,
@@ -145,34 +174,19 @@ def fetch_all_listings(slug: str) -> List[NFTListing]:
 
 
 def fetch_asset_listing_prices_with_retries(address: str, token_id: str) -> Optional[Dict[str, Union[str, int]]]:
-
-    def _get_listing_prices():
-        listings = fetch_listings(address, token_id)
-        for listing in listings:
-            return dict(price_str=listing.price_str, price_value=listing.price_value)
-        return None
-
-    return retry_on_exceptions_with_backoff(
-        _get_listing_prices,
-        [ErrorToRetry(requests.exceptions.HTTPError, _should_retry_exception)],
-    )
+    listings = fetch_listings(address, token_id)
+    for listing in listings:
+        return dict(price_str=listing.price_str, price_value=listing.price_value)
+    return None
 
 
 def fetch_contract_listing_prices_with_retries(address: str) -> Dict[str, Dict[str, Union[str, int]]]:
-
-    def _get_listing_prices():
-        contract = fetch_contract(address)
-        listings = fetch_all_listings(contract.slug)
-        ret = {}
-        for listing in listings:
-            if listing.token_id not in ret or ret[listing.token_id].price_value > listing.price_value:
-                ret[listing.token_id] = listing
-        return {token_id: dict(price_str=listing.price_str, price_value=listing.price_value) for token_id, listing in ret.items()}
-
-    return retry_on_exceptions_with_backoff(
-        _get_listing_prices,
-        [ErrorToRetry(requests.exceptions.HTTPError, _should_retry_exception)],
-    )
+    listings = fetch_all_listings(address)
+    ret = {}
+    for listing in listings:
+        if listing.token_id not in ret or ret[listing.token_id].price_value > listing.price_value:
+            ret[listing.token_id] = listing
+    return {token_id: dict(price_str=listing.price_str, price_value=listing.price_value) for token_id, listing in ret.items()}
 
 
 def _should_retry_exception(exception):

utils/common.py

@@ -165,7 +165,7 @@ def wrapped_fn(*args, **kwargs):
             return str(e)
         except Exception as e:
             traceback.print_exc()
-            return f'Got exception evaluating {fn.__name__}(args={args}, kwargs={kwargs}): {e}'
+            return "An error occurred. Please try again."
 
     @functools.wraps(fn)
     def wrapped_generator_fn(*args, **kwargs):