Sentry environment, job config interval, fix shutdown, minor improv…

…ements (#93)

pyproject.toml

@@ -26,7 +26,7 @@ aiohttp = "^3.7.4"
 asyncpg = "0.23.0"
 datamodel-code-generator = "^0.11.1"
 "ruamel.yaml" = "^0.17.2"
-tortoise-orm = "0.17.4"
+tortoise-orm = "0.17.5"
 pydantic = "^1.8.1"
 aiosignalrcore = "^0.9.2"
 fcache = "^0.4.7"

src/dipdup/cli.py

@@ -17,7 +17,7 @@
 from dipdup.dipdup import DipDup
 from dipdup.exceptions import ConfigurationError, DipDupError, MigrationRequiredError
 from dipdup.hasura import HasuraGateway
-from dipdup.utils import tortoise_wrapper
+from dipdup.utils import set_decimal_context, tortoise_wrapper
 
 _logger = logging.getLogger('dipdup.cli')
 
@@ -66,9 +66,12 @@ async def cli(ctx, config: List[str], logging_config: str):
     if _config.sentry:
         sentry_sdk.init(
             dsn=_config.sentry.dsn,
+            environment=_config.sentry.environment,
             integrations=[AioHttpIntegration()],
         )
 
+    set_decimal_context(_config.package)
+
     ctx.obj = CLIContext(
         config_paths=config,
         config=_config,
@@ -143,10 +146,8 @@ async def configure_hasura(ctx, reset: bool):
     if not config.hasura:
         _logger.error('`hasura` config section is empty')
         return
-    hasura = HasuraGateway(config.package, config.hasura, cast(PostgresDatabaseConfig, config.database))
+    hasura_gateway = HasuraGateway(config.package, config.hasura, cast(PostgresDatabaseConfig, config.database))
 
     async with tortoise_wrapper(url, models):
-        try:
-            await hasura.configure(reset)
-        finally:
-            await hasura.close_session()
+        async with hasura_gateway:
+            await hasura_gateway.configure(reset)

src/dipdup/config.py

@@ -93,9 +93,12 @@ def valid_immune_tables(cls, v):
 class HTTPConfig:
     cache: Optional[bool] = None
     retry_count: Optional[int] = None
-    retry_sleep: Optional[int] = None
+    retry_sleep: Optional[float] = None
+    retry_multiplier: Optional[float] = None
     ratelimit_rate: Optional[int] = None
     ratelimit_period: Optional[int] = None
+    connection_limit: Optional[int] = None
+    batch_size: Optional[int] = None
 
     def merge(self, other: Optional['HTTPConfig']) -> None:
         if not other:
@@ -162,12 +165,13 @@ class TzktDatasourceConfig(NameMixin):
     def __hash__(self):
         return hash(self.url)
 
-    @validator('url', allow_reuse=True)
-    def valid_url(cls, v):
-        parsed_url = urlparse(v)
+    def __post_init_post_parse__(self) -> None:
+        super().__post_init_post_parse__()
+        if self.http and self.http.batch_size and self.http.batch_size > 10000:
+            raise ConfigurationError('`batch_size` must be less than 10000')
+        parsed_url = urlparse(self.url)
         if not (parsed_url.scheme and parsed_url.netloc):
-            raise ConfigurationError(f'`{v}` is not a valid datasource URL')
-        return v
+            raise ConfigurationError(f'`{self.url}` is not a valid datasource URL')
 
 
 @dataclass
@@ -666,14 +670,21 @@ def headers(self) -> Dict[str, str]:
 
 @dataclass
 class JobConfig(HandlerConfig):
-    crontab: str
+    crontab: Optional[str] = None
+    interval: Optional[int] = None
     args: Optional[Dict[str, Any]] = None
     atomic: bool = False
 
+    def __post_init_post_parse__(self):
+        if int(bool(self.crontab)) + int(bool(self.interval)) != 1:
+            raise ConfigurationError('Either `interval` or `crontab` field must be specified')
+        super().__post_init_post_parse__()
+
 
 @dataclass
 class SentryConfig:
     dsn: str
+    environment: Optional[str] = None
 
 
 @dataclass

src/dipdup/datasources/bcd/datasource.py

@@ -49,8 +49,9 @@ async def get_token(self, address: str, token_id: int) -> Optional[Dict[str, Any
     def _default_http_config(self) -> HTTPConfig:
         return HTTPConfig(
             cache=True,
-            retry_count=3,
             retry_sleep=1,
+            retry_multiplier=1.1,
             ratelimit_rate=100,
             ratelimit_period=30,
+            connection_limit=25,
         )

src/dipdup/datasources/tzkt/datasource.py

@@ -25,7 +25,6 @@
 
 OperationID = int
 
-TZKT_HTTP_REQUEST_LIMIT = 10000
 TZKT_ORIGINATIONS_REQUEST_LIMIT = 100
 OPERATION_FIELDS = (
     "type",
@@ -291,7 +290,7 @@ def __init__(
 
     @property
     def request_limit(self) -> int:
-        return TZKT_HTTP_REQUEST_LIMIT
+        return self._http_config.batch_size or 10000
 
     @property
     def level(self) -> Optional[int]:
@@ -583,10 +582,11 @@ async def subscribe_to_head(self) -> None:
     def _default_http_config(self) -> HTTPConfig:
         return HTTPConfig(
             cache=True,
-            retry_count=3,
             retry_sleep=1,
+            retry_multiplier=1.1,
             ratelimit_rate=100,
             ratelimit_period=30,
+            connection_limit=25,
         )
 
     async def _on_operation_message(self, message: List[Dict[str, Any]]) -> None: