Caching TzKT request proxy (#39)

Author: droserasprout

poetry.lock

@@ -30,6 +30,7 @@ datamodel-code-generator = "^0.11.1"
 tortoise-orm = "^0.17.1"
 pydantic = "^1.8.1"
 aiosignalrcore = "^0.9.2"
+fcache = "^0.4.7"
 
 [tool.poetry.dev-dependencies]
 black = "^20.8b1"

pyproject.toml

@@ -7,6 +7,7 @@
 from typing import List
 
 import click
+from fcache.cache import FileCache  # type: ignore
 
 from dipdup import __version__
 from dipdup.config import DipDupConfig, LoggingConfig
@@ -70,3 +71,10 @@ async def init(ctx):
     config: DipDupConfig = ctx.obj.config
     dipdup = DipDup(config)
     await dipdup.init()
+
+
+@cli.command(help='Clear development request cache')
+@click.pass_context
+@click_async
+async def clear_cache(ctx):
+    FileCache('dipdup', flag='cs').clear()

src/dipdup/cli.py

@@ -50,7 +50,7 @@ async def get(
         contract_config: ContractConfig,
     ) -> Dict[str, Any]:
         if datasource_config not in self._datasources:
-            self._datasources[datasource_config] = TzktDatasource(datasource_config.url)
+            self._datasources[datasource_config] = TzktDatasource(datasource_config.url, True)
             self._schemas[datasource_config] = {}
         if contract_config.address not in self._schemas[datasource_config]:
             self._logger.info('Fetching schemas for contract `%s`', contract_config.address)

src/dipdup/codegen.py

@@ -20,6 +20,7 @@
 )
 from dipdup.datasources.tzkt.cache import BigMapCache, OperationCache
 from dipdup.datasources.tzkt.enums import TzktMessageType
+from dipdup.datasources.tzkt.proxy import TzktRequestProxy
 from dipdup.models import (
     BigMapAction,
     BigMapContext,
@@ -69,7 +70,7 @@
 
 
 class TzktDatasource:
-    def __init__(self, url: str):
+    def __init__(self, url: str, cache: bool):
         super().__init__()
         self._url = url.rstrip('/')
         self._logger = logging.getLogger(__name__)
@@ -85,6 +86,7 @@ def __init__(self, url: str):
         self._operation_cache = OperationCache()
         self._big_map_cache = BigMapCache()
         self._rollback_fn: Optional[Callable[[int, int], Awaitable[None]]] = None
+        self._proxy = TzktRequestProxy(cache)
 
     async def add_index(self, index_name: str, index_config: Union[OperationIndexConfig, BigMapIndexConfig, BlockIndexConfig]):
         self._logger.info('Adding index `%s`', index_name)
@@ -203,7 +205,7 @@ async def subscribe_to_big_maps(self, address: Address, path: Path) -> None:
     async def _fetch_operations(self, addresses: List[str], offset: int, first_level: int, last_level: int) -> List[Dict[str, Any]]:
         self._logger.info('Fetching levels %s-%s with offset %s', first_level, last_level, offset)
 
-        async with http_request(
+        operations = await self._proxy.http_request(
             'get',
             url=f'{self._url}/v1/operations/transactions',
             params={
@@ -215,10 +217,9 @@ async def _fetch_operations(self, addresses: List[str], offset: int, first_level
                 "select": ','.join(OPERATION_FIELDS),
                 "status": "applied",
             },
-        ) as resp:
-            operations = await resp.json()
+        )
 
-        async with http_request(
+        target_operations = await self._proxy.http_request(
             'get',
             url=f'{self._url}/v1/operations/transactions',
             params={
@@ -230,8 +231,7 @@ async def _fetch_operations(self, addresses: List[str], offset: int, first_level
                 "select": ','.join(OPERATION_FIELDS),
                 "status": "applied",
             },
-        ) as resp:
-            target_operations = await resp.json()
+        )
 
         sender_operation_keys = {op['id'] for op in operations}
         for op in target_operations:
@@ -297,7 +297,7 @@ async def _fetch_big_maps(
     ) -> List[Dict[str, Any]]:
         self._logger.info('Fetching levels %s-%s with offset %s', first_level, last_level, offset)
 
-        async with http_request(
+        big_maps = await self._proxy.http_request(
             'get',
             url=f'{self._url}/v1/bigmaps/updates',
             params={
@@ -308,8 +308,7 @@ async def _fetch_big_maps(
                 "level.gt": first_level,
                 "level.le": last_level,
             },
-        ) as resp:
-            big_maps = await resp.json()
+        )
 
         self._logger.info('%s big map updates fetched', len(big_maps))
         self._logger.debug(big_maps)
@@ -369,11 +368,10 @@ async def _process_level_big_maps(big_maps):
 
     async def fetch_jsonschemas(self, address: str) -> Dict[str, Any]:
         self._logger.info('Fetching jsonschemas for address `%s', address)
-        async with http_request(
+        jsonschemas = await self._proxy.http_request(
             'get',
             url=f'{self._url}/v1/contracts/{address}/interface',
-        ) as response:
-            jsonschemas = await response.json()
+        )
         self._logger.debug(jsonschemas)
         return jsonschemas
 
@@ -597,10 +595,10 @@ def convert_big_map(cls, big_map_json: Dict[str, Any]) -> BigMapData:
 
     async def get_latest_block(self) -> Dict[str, Any]:
         self._logger.info('Fetching latest block')
-        async with http_request(
+        block = await self._proxy.http_request(
             'get',
             url=f'{self._url}/v1/head',
-        ) as resp:
-            block = await resp.json()
+            skip_cache=True,
+        )
         self._logger.debug(block)
         return block

src/dipdup/datasources/tzkt/datasource.py

@@ -0,0 +1,26 @@
+import hashlib
+import logging
+import pickle
+
+from fcache.cache import FileCache  # type: ignore
+
+from dipdup.utils import http_request
+
+
+class TzktRequestProxy:
+    def __init__(self, cache: bool = False) -> None:
+        self._logger = logging.getLogger(__name__)
+        self._cache = FileCache('dipdup', flag='cs') if cache else None
+
+    async def http_request(self, method: str, skip_cache: bool = False, **kwargs):
+        if self._cache is not None and not skip_cache:
+            key = hashlib.sha256(pickle.dumps([method, kwargs])).hexdigest()
+            try:
+                return self._cache[key]
+            except KeyError:
+                response = await http_request(method, **kwargs)
+                self._cache[key] = response
+                return response
+        else:
+            response = await http_request(method, **kwargs)
+            return response

src/dipdup/datasources/tzkt/proxy.py

@@ -10,7 +10,14 @@
 
 import dipdup.codegen as codegen
 from dipdup import __version__
-from dipdup.config import ROLLBACK_HANDLER, DipDupConfig, IndexTemplateConfig, PostgresDatabaseConfig, TzktDatasourceConfig
+from dipdup.config import (
+    ROLLBACK_HANDLER,
+    DipDupConfig,
+    IndexTemplateConfig,
+    PostgresDatabaseConfig,
+    SqliteDatabaseConfig,
+    TzktDatasourceConfig,
+)
 from dipdup.datasources.tzkt.datasource import TzktDatasource
 from dipdup.hasura import configure_hasura
 from dipdup.models import IndexType, State
@@ -31,6 +38,7 @@ async def init(self) -> None:
 
     async def run(self) -> None:
         url = self._config.database.connection_string
+        cache = isinstance(self._config.database, SqliteDatabaseConfig)
         models = f'{self._config.package}.models'
         rollback_fn = getattr(importlib.import_module(f'{self._config.package}.handlers.{ROLLBACK_HANDLER}'), ROLLBACK_HANDLER)
 
@@ -48,7 +56,7 @@ async def run(self) -> None:
                 self._logger.info('Processing index `%s`', index_name)
                 if isinstance(index_config.datasource, TzktDatasourceConfig):
                     if index_config.datasource_config not in datasources:
-                        datasources[index_config.datasource_config] = TzktDatasource(index_config.datasource_config.url)
+                        datasources[index_config.datasource_config] = TzktDatasource(index_config.datasource_config.url, cache)
                         datasources[index_config.datasource_config].set_rollback_fn(rollback_fn)
                     await datasources[index_config.datasource_config].add_index(index_name, index_config)
                 else:

src/dipdup/dipdup.py

@@ -148,9 +148,8 @@ async def configure_hasura(config: DipDupConfig):
     _logger.info('Waiting for Hasura instance to be healthy')
     for _ in range(60):
         with suppress(ClientConnectorError, ClientOSError):
-            async with http_request('get', url=f'{url}/healthz') as response:
-                if response.status == 200:
-                    break
+            await http_request('get', url=f'{url}/healthz')
+            break
         await asyncio.sleep(1)
     else:
         _logger.error('Hasura instance not responding for 60 seconds')
@@ -161,7 +160,7 @@ async def configure_hasura(config: DipDupConfig):
         headers['X-Hasura-Admin-Secret'] = config.hasura.admin_secret
 
     _logger.info('Fetching existing metadata')
-    async with http_request(
+    existing_hasura_metadata = await http_request(
         'post',
         url=f'{url}/v1/query',
         data=json.dumps(
@@ -171,8 +170,7 @@ async def configure_hasura(config: DipDupConfig):
             },
         ),
         headers=headers,
-    ) as response:
-        existing_hasura_metadata = await response.json()
+    )
 
     _logger.info('Merging existing metadata')
     hasura_metadata_tables = [table['table'] for table in hasura_metadata['tables']]
@@ -181,7 +179,7 @@ async def configure_hasura(config: DipDupConfig):
             hasura_metadata['tables'].append(table)
 
     _logger.info('Sending replace metadata request')
-    async with http_request(
+    result = await http_request(
         'post',
         url=f'{url}/v1/query',
         data=json.dumps(
@@ -191,7 +189,6 @@ async def configure_hasura(config: DipDupConfig):
             },
         ),
         headers=headers,
-    ) as response:
-        result = await response.json()
-        if not result.get('message') == 'success':
-            _logger.error('Can\'t configure Hasura instance: %s', result)
+    )
+    if not result.get('message') == 'success':
+        _logger.error('Can\'t configure Hasura instance: %s', result)

src/dipdup/hasura.py

@@ -46,7 +46,6 @@ async def tortoise_wrapper(url: str, models: Optional[str] = None):
         await Tortoise.close_connections()
 
 
-@asynccontextmanager
 async def http_request(method: str, **kwargs):
     async with aiohttp.ClientSession() as session:
         headers = {
@@ -60,7 +59,7 @@ async def http_request(method: str, **kwargs):
         ) as response:
             request_string = kwargs['url'] + '?' + '&'.join([f'{key}={value}' for key, value in kwargs.get('params', {}).items()])
             _logger.debug('Calling `%s`', request_string)
-            yield response
+            return await response.json()
 
 
 async def reindex():

src/dipdup/utils.py

@@ -35,7 +35,7 @@ async def asyncSetUp(self):
         )
         self.index_config.state = State(index_name='test', index_type=IndexType.operation, hash='')
         self.index_config.handlers[0].pattern[0].parameter_type_cls = CollectParameter
-        self.datasource = TzktDatasource('tzkt.test')
+        self.datasource = TzktDatasource('tzkt.test', True)
         await self.datasource.add_index('test', self.index_config)
 
     async def test_convert_operation(self):