Add cache for IPFS fetching (#105)

Signed-off-by: cyc60 <[email protected]>

Signed-off-by: cyc60 <[email protected]>

Author: cyc60

oracle/oracle/common/ipfs.py

@@ -5,6 +5,7 @@
 import ipfshttpclient
 from aiohttp import ClientSession, ClientTimeout
 
+from oracle.oracle.utils import LimitedSizeDict
 from oracle.settings import (
     INFURA_IPFS_CLIENT_ENDPOINT,
     INFURA_IPFS_CLIENT_PASSWORD,
@@ -17,39 +18,51 @@
 
 timeout = ClientTimeout(total=60)
 
+CACHE_SIZE = 1024
+IPFS_CACHE = LimitedSizeDict(size_limit=CACHE_SIZE)
+
 
 @backoff.on_exception(backoff.expo, Exception, max_time=900)
 async def ipfs_fetch(ipfs_hash: str) -> Union[Dict[Any, Any], List[Dict[Any, Any]]]:
     """Tries to fetch IPFS hash from different sources."""
     _ipfs_hash = ipfs_hash.replace("ipfs://", "").replace("/ipfs/", "")
 
-    async with ClientSession(timeout=timeout) as session:
-        for endpoint in IPFS_FETCH_ENDPOINTS:
+    if IPFS_CACHE.get(_ipfs_hash):
+        return IPFS_CACHE.get(_ipfs_hash)
+
+    async def _fetch(_ipfs_hash):
+        async with ClientSession(timeout=timeout) as session:
+            for endpoint in IPFS_FETCH_ENDPOINTS:
+                try:
+                    response = await session.get(
+                        f"{endpoint.rstrip('/')}/ipfs/{_ipfs_hash}"
+                    )
+                    response.raise_for_status()
+                    return await response.json()
+                except BaseException as e:  # noqa: E722
+                    logger.exception(e)
+                    pass
+
+        if LOCAL_IPFS_CLIENT_ENDPOINT:
             try:
-                response = await session.get(
-                    f"{endpoint.rstrip('/')}/ipfs/{_ipfs_hash}"
-                )
-                response.raise_for_status()
-                return await response.json()
-            except BaseException as e:  # noqa: E722
-                logger.exception(e)
+                with ipfshttpclient.connect(LOCAL_IPFS_CLIENT_ENDPOINT) as client:
+                    return client.get_json(_ipfs_hash)
+            except ipfshttpclient.exceptions.TimeoutError:
                 pass
 
-    if LOCAL_IPFS_CLIENT_ENDPOINT:
         try:
-            with ipfshttpclient.connect(LOCAL_IPFS_CLIENT_ENDPOINT) as client:
+            with ipfshttpclient.connect(
+                INFURA_IPFS_CLIENT_ENDPOINT,
+                username=INFURA_IPFS_CLIENT_USERNAME,
+                password=INFURA_IPFS_CLIENT_PASSWORD,
+            ) as client:
                 return client.get_json(_ipfs_hash)
         except ipfshttpclient.exceptions.TimeoutError:
             pass
 
-    try:
-        with ipfshttpclient.connect(
-            INFURA_IPFS_CLIENT_ENDPOINT,
-            username=INFURA_IPFS_CLIENT_USERNAME,
-            password=INFURA_IPFS_CLIENT_PASSWORD,
-        ) as client:
-            return client.get_json(_ipfs_hash)
-    except ipfshttpclient.exceptions.TimeoutError:
-        pass
+    data = await _fetch(_ipfs_hash)
+    if data:
+        IPFS_CACHE[_ipfs_hash] = data
+        return data
 
     raise RuntimeError(f"Failed to fetch IPFS data at {_ipfs_hash}")

oracle/oracle/utils.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+from collections import OrderedDict
 from functools import wraps
 
 logger = logging.getLogger(__name__)
@@ -25,3 +26,19 @@ def wrapper(*args, **kwargs):
                 logger.exception(e)
 
     return wrapper
+
+
+class LimitedSizeDict(OrderedDict):
+    def __init__(self, *args, **kwds):
+        self.size_limit = kwds.pop("size_limit", None)
+        OrderedDict.__init__(self, *args, **kwds)
+        self._check_size_limit()
+
+    def __setitem__(self, key, value):
+        OrderedDict.__setitem__(self, key, value)
+        self._check_size_limit()
+
+    def _check_size_limit(self):
+        if self.size_limit is not None:
+            while len(self) > self.size_limit:
+                self.popitem(last=False)