Partial indexing, integration tests

src/demo_hic_et_nunc/dipdup.yml

@@ -3,7 +3,7 @@ package: demo_hic_et_nunc
 
 database:
   kind: sqlite
-  path: db.sqlite3
+  path: hic_et_nunc.sqlite3
 
 contracts:
   HEN_objkts: 

src/demo_quipuswap/dipdup.yml

@@ -3,7 +3,7 @@ package: demo_quipuswap
 
 database:
   kind: sqlite
-  path: db.sqlite3
+  path: quipuswap.sqlite3
 
 contracts:
   kusd_dex_mainnet:

src/demo_quipuswap/hasura_metadata.json

@@ -72,8 +72,8 @@
                     "role": "user",
                     "permission": {
                         "columns": [
-                            "tez_qty",
                             "id",
+                            "tez_qty",
                             "token_qty"
                         ],
                         "filter": {},
@@ -107,13 +107,13 @@
                     "role": "user",
                     "permission": {
                         "columns": [
-                            "slippage",
-                            "quantity",
+                            "side",
                             "level",
+                            "slippage",
                             "price",
-                            "side",
+                            "id",
                             "timestamp",
-                            "id"
+                            "quantity"
                         ],
                         "filter": {},
                         "allow_aggregations": true

src/dipdup/cli.py

@@ -20,6 +20,7 @@
 from dipdup.config import DipDupConfig, IndexTemplateConfig, LoggingConfig, OperationIndexConfig, TzktDatasourceConfig
 from dipdup.datasources.tzkt.datasource import TzktDatasource
 from dipdup.models import IndexType, State
+from dipdup.utils import tortoise_wrapper
 
 _logger = logging.getLogger(__name__)
 
@@ -69,15 +70,10 @@ async def cli(ctx, config: str, logging_config: str):
 async def run(ctx) -> None:
     config: DipDupConfig = ctx.obj.config
 
-    try:
+    url = config.database.connection_string
+    models = f'{config.package}.models'
+    async with tortoise_wrapper(url, models):
         _logger.info('Initializing database')
-        await Tortoise.init(
-            db_url=config.database.connection_string,
-            modules={
-                'models': [f'{config.package}.models'],
-                'int_models': ['dipdup.models'],
-            },
-        )
 
         connection_name, connection = next(iter(Tortoise._connections.items()))
         schema_sql = get_schema_sql(connection, False)
@@ -118,9 +114,6 @@ async def run(ctx) -> None:
         datasource_run_tasks = [asyncio.create_task(d.start()) for d in datasources.values()]
         await asyncio.gather(*datasource_run_tasks)
 
-    finally:
-        await Tortoise.close_connections()
-
 
 @cli.command(help='Initialize new dipdap')
 @click.pass_context

src/dipdup/config.py

@@ -432,6 +432,7 @@ async def initialize(self) -> None:
                         index_name=index_name,
                         index_type=IndexType.operation,
                         hash=index_hash,
+                        level=index_config.first_block - 1,
                     )
                     await state.save()
 

src/dipdup/configs/debug.yml

@@ -13,13 +13,15 @@
     SignalRCoreClient:
       formatter: brief
     dipdup.datasources.tzkt.datasource:
-      level: INFO
+      level: DEBUG
     dipdup.datasources.tzkt.cache:
-      level: INFO
+      level: DEBUG
     aiosqlite:
-      level: INFO
+      level: DEBUG
     db_client:
-      level: INFO
+      level: DEBUG
+    dipdup.models:
+      level: DEBUG
   root:
     level: DEBUG
     handlers:

src/dipdup/configs/warning.yml

@@ -0,0 +1,26 @@
+  version: 1
+  disable_existing_loggers: false
+  formatters:
+    brief:
+      format: "%(levelname)-8s %(name)-35s %(message)s"
+  handlers:
+    console:
+      level: WARNING
+      formatter: brief
+      class: logging.StreamHandler
+      stream : ext://sys.stdout
+  loggers:
+    SignalRCoreClient:
+      formatter: brief
+    dipdup.datasources.tzkt.datasource:
+      level: INFO
+    dipdup.datasources.tzkt.cache:
+      level: INFO
+    aiosqlite:
+      level: INFO
+    db_client:
+      level: INFO
+  root:
+    level: WARNING
+    handlers:
+      - console

src/dipdup/datasources/tzkt/datasource.py

@@ -109,17 +109,24 @@ def _get_client(self) -> BaseHubConnection:
 
     async def start(self):
         self._logger.info('Starting datasource')
+        rest_only = False
         for operation_index_config in self._operation_index_configs.values():
-            await self.add_subscription(operation_index_config.contract)
 
+            if operation_index_config.last_block:
+                await self.fetch_operations(operation_index_config.last_block, initial=True)
+                rest_only = True
+                continue
+
+            await self.add_subscription(operation_index_config.contract)
             latest_block = await self.get_latest_block()
             current_level = latest_block['level']
             state_level = operation_index_config.state.level
             if current_level != state_level:
                 await self.fetch_operations(current_level, initial=True)
 
-        self._logger.info('Starting websocket client')
-        await self._get_client().start()
+        if not rest_only:
+            self._logger.info('Starting websocket client')
+            await self._get_client().start()
 
     async def stop(self):
         ...
@@ -193,7 +200,7 @@ async def _process_operations(address, operations):
         for index_config in self._operation_index_configs.values():
 
             sync_event = self._sync_events[index_config.state.index_name]
-            level = index_config.state.level or 0
+            level = index_config.state.level
 
             operations = []
             offset = 0

src/dipdup/models.py

@@ -81,23 +81,23 @@ def get_merged_storage(self, storage_type: Type[StorageType]) -> StorageType:
         if self.storage is None:
             raise Exception('`storage` field missing')
 
-        storage = self.storage
-        if self.bigmaps:
-            storage = deepcopy(self.storage)
-            _logger.debug('Merging storage')
-            _logger.debug('Before: %s', storage)
-            for key, field in storage_type.__fields__.items():
-                # NOTE: TzKT could return bigmaps as object or as array of key-value objects. We need to guess this from storage.
-                # TODO: This code should be a part of datasource module.
-                if field.type_ not in (int, bool) and isinstance(storage[key], int):
-                    if hasattr(field.type_, '__fields__') and 'key' in field.type_.__fields__ and 'value' in field.type_.__fields__:
-                        storage[key] = []
+        storage = deepcopy(self.storage)
+        _logger.debug('Merging storage')
+        _logger.debug('Before: %s', storage)
+        for key, field in storage_type.__fields__.items():
+            # NOTE: TzKT could return bigmaps as object or as array of key-value objects. We need to guess this from storage.
+            # TODO: This code should be a part of datasource module.
+            if field.type_ not in (int, bool) and isinstance(storage[key], int):
+                if hasattr(field.type_, '__fields__') and 'key' in field.type_.__fields__ and 'value' in field.type_.__fields__:
+                    storage[key] = []
+                    if self.bigmaps:
                         self._merge_bigmapdiffs(storage, key, array=True)
-                    else:
-                        storage[key] = {}
+                else:
+                    storage[key] = {}
+                    if self.bigmaps:
                         self._merge_bigmapdiffs(storage, key, array=False)
 
-            _logger.debug('After: %s', storage)
+        _logger.debug('After: %s', storage)
 
         return storage_type.parse_obj(storage)
 

src/dipdup/utils.py

@@ -0,0 +1,19 @@
+from contextlib import asynccontextmanager
+from typing import Optional
+
+from tortoise import Tortoise
+
+
+@asynccontextmanager
+async def tortoise_wrapper(url: str, models: Optional[str] = None):
+    try:
+        modules = {'int_models': ['dipdup.models']}
+        if models:
+            modules['models'] = [models]
+        await Tortoise.init(
+            db_url=url,
+            modules=modules,  # type: ignore
+        )
+        yield
+    finally:
+        await Tortoise.close_connections()

tests/integration_tests/hic_et_nunc.yml

@@ -0,0 +1,46 @@
+spec_version: 0.0.1
+package: demo_hic_et_nunc
+
+database:
+  kind: sqlite
+  path: db.sqlite3
+
+contracts:
+  HEN_objkts: 
+    address: ${HEN_OBJKTS:-KT1RJ6PbjHpwc3M5rw5s2Nbmefwbuwbdxton}
+    typename: hen_objkts
+  HEN_minter: 
+    address: ${HEN_MINTER:-KT1Hkg5qeNhfwpKW4fXvq7HGZB9z2EnmCCA9}
+    typename: hen_minter
+
+datasources:
+  tzkt_mainnet:
+    kind: tzkt
+    url: ${TZKT_URL:-https://staging.api.tzkt.io}
+
+indexes:
+  hen_mainnet:
+    kind: operation
+    datasource: tzkt_mainnet
+    contract: HEN_minter
+    handlers:
+      - callback: on_mint
+        pattern:
+          - destination: HEN_minter
+            entrypoint: mint_OBJKT
+          - destination: HEN_objkts
+            entrypoint: mint
+      - callback: on_swap
+        pattern:
+          - destination: HEN_minter
+            entrypoint: swap
+      - callback: on_cancel_swap
+        pattern:
+          - destination: HEN_minter
+            entrypoint: cancel_swap
+      - callback: on_collect
+        pattern:
+          - destination: HEN_minter
+            entrypoint: collect
+    first_block: 1365000
+    last_block: 1366000