Fix HTTPGateway, reset Hasura on configuration failure (#118)

* Perf counter

* CLI help

* Fix ratelimiting in HTTPGateway

* Hasura reset option

* Fix bugs in retrier

* Cleanup

* Cleanup

Author: droserasprout

.gitignore

@@ -145,4 +145,6 @@ cython_debug/
 
 .noseids
 
-secrets.env
+secrets.env
+
+*.out*

src/dipdup/cli.py

@@ -160,21 +160,22 @@ def _bump_spec_version(spec_version: str):
         raise ConfigurationError('Unknown `spec_version`')
 
 
+# TODO: "cache clear"?
 @cli.command(help='Clear development request cache')
 @click.pass_context
 @cli_wrapper
 async def clear_cache(ctx):
     FileCache('dipdup', flag='cs').clear()
 
 
-@cli.group()
+@cli.group(help='Docker integration related commands')
 @click.pass_context
 @cli_wrapper
 async def docker(ctx):
     ...
 
 
-@docker.command(name='init')
+@docker.command(name='init', help='Generate Docker inventory in project directory')
 @click.option('--image', '-i', type=str, help='DipDup Docker image', default=DEFAULT_DOCKER_IMAGE)
 @click.option('--tag', '-t', type=str, help='DipDup Docker tag', default=DEFAULT_DOCKER_TAG)
 @click.option('--env-file', '-e', type=str, help='Path to env_file', default=DEFAULT_DOCKER_ENV_FILE)
@@ -185,7 +186,7 @@ async def docker_init(ctx, image: str, tag: str, env_file: str):
     await DipDupCodeGenerator(config, {}).generate_docker(image, tag, env_file)
 
 
-@cli.group()
+@cli.group(help='Hasura integration related commands')
 @click.pass_context
 @cli_wrapper
 async def hasura(ctx):
@@ -201,10 +202,15 @@ async def hasura_configure(ctx, reset: bool):
     url = config.database.connection_string
     models = f'{config.package}.models'
     if not config.hasura:
-        _logger.error('`hasura` config section is empty')
-        return
-    hasura_gateway = HasuraGateway(config.package, config.hasura, cast(PostgresDatabaseConfig, config.database))
+        raise ConfigurationError('`hasura` config section is empty')
+    if reset:
+        config.hasura.reset = True
+    hasura_gateway = HasuraGateway(
+        package=config.package,
+        hasura_config=config.hasura,
+        database_config=cast(PostgresDatabaseConfig, config.database),
+    )
 
     async with tortoise_wrapper(url, models):
         async with hasura_gateway:
-            await hasura_gateway.configure(reset)
+            await hasura_gateway.configure()

src/dipdup/config.py

@@ -7,6 +7,7 @@
 import sys
 from abc import ABC, abstractmethod
 from collections import defaultdict
+from copy import copy
 from enum import Enum
 from os import environ as env
 from os.path import dirname
@@ -101,12 +102,13 @@ class HTTPConfig:
     connection_limit: Optional[int] = None
     batch_size: Optional[int] = None
 
-    def merge(self, other: Optional['HTTPConfig']) -> None:
-        if not other:
-            return
-        for k, v in other.__dict__.items():
-            if v is not None:
-                setattr(self, k, v)
+    def merge(self, other: Optional['HTTPConfig']) -> 'HTTPConfig':
+        config = copy(self)
+        if other:
+            for k, v in other.__dict__.items():
+                if v is not None:
+                    setattr(config, k, v)
+        return config
 
 
 @dataclass
@@ -659,6 +661,7 @@ class HasuraConfig:
     camel_case: bool = False
     connection_timeout: int = 5
     rest: bool = True
+    reset: Optional[bool] = None
     http: Optional[HTTPConfig] = None
 
     @validator('url', allow_reuse=True)

src/dipdup/datasources/bcd/datasource.py

@@ -8,13 +8,22 @@
 
 
 class BcdDatasource(Datasource):
+    _default_http_config = HTTPConfig(
+        cache=True,
+        retry_sleep=1,
+        retry_multiplier=1.1,
+        ratelimit_rate=100,
+        ratelimit_period=30,
+        connection_limit=25,
+    )
+
     def __init__(
         self,
         url: str,
         network: str,
         http_config: Optional[HTTPConfig] = None,
     ) -> None:
-        super().__init__(url, http_config)
+        super().__init__(url, self._default_http_config.merge(http_config))
         self._logger = logging.getLogger('dipdup.bcd')
         self._network = network
 
@@ -45,13 +54,3 @@ async def get_token(self, address: str, token_id: int) -> Optional[Dict[str, Any
         if response:
             return response[0]
         return None
-
-    def _default_http_config(self) -> HTTPConfig:
-        return HTTPConfig(
-            cache=True,
-            retry_sleep=1,
-            retry_multiplier=1.1,
-            ratelimit_rate=100,
-            ratelimit_period=30,
-            connection_limit=25,
-        )

src/dipdup/datasources/coinbase/datasource.py

@@ -11,8 +11,16 @@
 
 
 class CoinbaseDatasource(Datasource):
+    _default_http_config = HTTPConfig(
+        cache=True,
+        retry_count=3,
+        retry_sleep=1,
+        ratelimit_rate=10,
+        ratelimit_period=1,
+    )
+
     def __init__(self, url: str = API_URL, http_config: Optional[HTTPConfig] = None) -> None:
-        super().__init__(url, http_config)
+        super().__init__(url, self._default_http_config.merge(http_config))
         self._logger = logging.getLogger('dipdup.coinbase')
 
     async def run(self) -> None:
@@ -43,15 +51,6 @@ async def get_candles(self, since: datetime, until: datetime, interval: CandleIn
             candles += [CandleData.from_json(c) for c in candles_json]
         return sorted(candles, key=lambda c: c.timestamp)
 
-    def _default_http_config(self) -> HTTPConfig:
-        return HTTPConfig(
-            cache=True,
-            retry_count=3,
-            retry_sleep=1,
-            ratelimit_rate=10,
-            ratelimit_period=1,
-        )
-
     def _split_candle_requests(self, since: datetime, until: datetime, interval: CandleInterval) -> List[Tuple[datetime, datetime]]:
         request_interval_limit = timedelta(seconds=interval.seconds * CANDLES_REQUEST_LIMIT)
         request_intervals = []

src/dipdup/datasources/datasource.py

@@ -1,7 +1,7 @@
 from abc import abstractmethod
 from enum import Enum
 from logging import Logger
-from typing import Awaitable, List, Optional, Protocol
+from typing import Awaitable, List, Protocol
 
 from pyee import AsyncIOEventEmitter  # type: ignore
 
@@ -46,7 +46,7 @@ async def run(self) -> None:
 
 
 class IndexDatasource(Datasource, AsyncIOEventEmitter):
-    def __init__(self, url: str, http_config: Optional[HTTPConfig] = None) -> None:
+    def __init__(self, url: str, http_config: HTTPConfig) -> None:
         HTTPGateway.__init__(self, url, http_config)
         AsyncIOEventEmitter.__init__(self)
 

src/dipdup/datasources/tzkt/datasource.py

@@ -288,12 +288,21 @@ class TzktDatasource(IndexDatasource):
     * Calls Matchers to match received operation groups with indexes' pattern and spawn callbacks on match
     """
 
+    _default_http_config = HTTPConfig(
+        cache=True,
+        retry_sleep=1,
+        retry_multiplier=1.1,
+        ratelimit_rate=100,
+        ratelimit_period=30,
+        connection_limit=25,
+    )
+
     def __init__(
         self,
         url: str,
         http_config: Optional[HTTPConfig] = None,
     ) -> None:
-        super().__init__(url, http_config)
+        super().__init__(url, self._default_http_config.merge(http_config))
         self._logger = logging.getLogger('dipdup.tzkt')
 
         self._transaction_subscriptions: Set[str] = set()
@@ -623,21 +632,11 @@ async def _subscribe_to_head(self) -> None:
             [],
         )
 
-    def _default_http_config(self) -> HTTPConfig:
-        return HTTPConfig(
-            cache=True,
-            retry_sleep=1,
-            retry_multiplier=1.1,
-            ratelimit_rate=100,
-            ratelimit_period=30,
-            connection_limit=25,
-        )
-
     async def _extract_message_data(self, channel: str, message: List[Any]) -> Any:
         for item in message:
             head_level = item['state']
             message_type = TzktMessageType(item['type'])
-            self._logger.debug('`%s` message: %s', channel, message_type.name) 
+            self._logger.debug('`%s` message: %s', channel, message_type.name)
 
             if message_type == TzktMessageType.STATE:
                 if self._sync_level != head_level:
@@ -657,7 +656,6 @@ async def _extract_message_data(self, channel: str, message: List[Any]) -> Any:
             else:
                 raise NotImplementedError
 
-
     async def _on_operation_message(self, message: List[Dict[str, Any]]) -> None:
         """Parse and emit raw operations from WS"""
         async for data in self._extract_message_data('operation', message):

src/dipdup/hasura.py

@@ -3,9 +3,11 @@
 import logging
 import re
 from contextlib import suppress
+from http import HTTPStatus
 from os.path import dirname, join
 from typing import Any, Dict, Iterator, List, Optional, Tuple
 
+import aiohttp
 import humps  # type: ignore
 from aiohttp import ClientConnectorError, ClientOSError
 from pydantic.dataclasses import dataclass
@@ -60,26 +62,32 @@ class HasuraError(RuntimeError):
 
 
 class HasuraGateway(HTTPGateway):
+    _default_http_config = HTTPConfig(
+        cache=False,
+        retry_count=3,
+        retry_sleep=1,
+    )
+
     def __init__(
         self,
         package: str,
         hasura_config: HasuraConfig,
         database_config: PostgresDatabaseConfig,
         http_config: Optional[HTTPConfig] = None,
     ) -> None:
-        super().__init__(hasura_config.url, http_config)
+        super().__init__(hasura_config.url, self._default_http_config.merge(http_config))
         self._logger = logging.getLogger('dipdup.hasura')
         self._package = package
         self._hasura_config = hasura_config
         self._database_config = database_config
 
-    async def configure(self, reset: bool = False) -> None:
+    async def configure(self) -> None:
         """Generate Hasura metadata and apply to instance with credentials from `hasura` config section."""
 
         self._logger.info('Configuring Hasura')
         await self._healthcheck()
 
-        if reset:
+        if self._hasura_config.reset is True:
             await self._reset_metadata()
 
         metadata = await self._fetch_metadata()
@@ -132,15 +140,6 @@ async def configure(self, reset: bool = False) -> None:
 
         self._logger.info('Hasura instance has been configured')
 
-    def _default_http_config(self) -> HTTPConfig:
-        return HTTPConfig(
-            cache=False,
-            retry_sleep=1,
-            retry_multiplier=1.1,
-            ratelimit_rate=100,
-            ratelimit_period=1,
-        )
-
     async def _hasura_request(self, endpoint: str, json: Dict[str, Any]) -> Dict[str, Any]:
         self._logger.debug('Sending `%s` request: %s', endpoint, json)
         result = await self._http.request(
@@ -188,13 +187,22 @@ async def _fetch_metadata(self) -> Dict[str, Any]:
 
     async def _replace_metadata(self, metadata: Dict[str, Any]) -> None:
         self._logger.info('Replacing metadata')
-        await self._hasura_request(
-            endpoint='query',
-            json={
-                "type": "replace_metadata",
-                "args": metadata,
-            },
-        )
+        endpoint, json = 'query', {
+            "type": "replace_metadata",
+            "args": metadata,
+        }
+        try:
+            await self._hasura_request(endpoint, json)
+        except aiohttp.ClientResponseError as e:
+            # NOTE: 400 from Hasura means we failed either to generate or to merge existing metadata.
+            # NOTE: Reset metadata and retry if not forbidden by config.
+            print(e.status, self._hasura_config.reset)
+            if e.status != HTTPStatus.BAD_REQUEST or self._hasura_config.reset is False:
+                print('raise')
+                raise
+            self._logger.warning('Failed to replace metadata, resetting')
+            await self._reset_metadata()
+            await self._hasura_request(endpoint, json)
 
     async def _get_views(self) -> List[str]:
         return [