Metadata interface for TzKT (#213)

* Use orjson

* metadata_interface flag

* Builtin hooks fix

* docs

* Changelog

* CLI, init cursor

* fix metadata interface

* network fiels

* text token_id, fix decimal precision

* Fix changelog

* Additional config checks

Author: droserasprout

CHANGELOG.md

@@ -6,6 +6,8 @@ Please use [this](https://docs.gitlab.com/ee/development/changelog.html) documen
 
 ### Added
 
+* metadata: Added `metadata_interface` feature flag to expose metadata in TzKT format.
+* tzkt: Added missing fields to the `HeadBlockData` model.
 * tzkt: Added `iter_...` methods to iterate over item batches.
 
 ### Fixed
@@ -21,6 +23,10 @@ Please use [this](https://docs.gitlab.com/ee/development/changelog.html) documen
 
 * bcd: Removed `bcd` datasource and config section.
 
+### Performance
+
+* dipdup: Use fast `orjson` library instead of built-in `json` where possible.
+
 ## 4.2.6 - 2022-02-25
 
 ### Fixed

poetry.lock

@@ -30,6 +30,7 @@ asyncclick = "^8.0.1"
 asyncpg = "^0.24.0"
 datamodel-code-generator = "^0.11.18"
 fcache = "^0.4.7"
+orjson = "^3.6.6"
 pydantic = "^1.8.1"
 pyhumps = "^3.0.2"
 pysignalr = "^0.1.1"

pyproject.toml

@@ -195,19 +195,22 @@ async def cli(ctx, config: List[str], env_file: List[str], logging_config: str):
 @click.option('--postpone-jobs', is_flag=True, help='Do not start job scheduler until all indexes are synchronized')
 @click.option('--early-realtime', is_flag=True, help='Establish a realtime connection before all indexes are synchronized')
 @click.option('--merge-subscriptions', is_flag=True, help='Subscribe to all operations/big map diffs during realtime indexing')
+@click.option('--metadata-interface', is_flag=True, help='Enable metadata interface')
 @click.pass_context
 @cli_wrapper
 async def run(
     ctx,
     postpone_jobs: bool,
     early_realtime: bool,
     merge_subscriptions: bool,
+    metadata_interface: bool,
 ) -> None:
     config: DipDupConfig = ctx.obj.config
     config.initialize()
     config.advanced.postpone_jobs |= postpone_jobs
     config.advanced.early_realtime |= early_realtime
     config.advanced.merge_subscriptions |= merge_subscriptions
+    config.advanced.metadata_interface |= metadata_interface
 
     set_decimal_context(config.package)
 

src/dipdup/cli.py

@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 import re
@@ -16,6 +15,7 @@
 from typing import List
 from typing import cast
 
+import orjson as json
 from jinja2 import Template
 
 from dipdup import __version__
@@ -182,7 +182,7 @@ async def fetch_schemas(self) -> None:
                         storage_schema_path = join(contract_schemas_path, 'storage.json')
                         storage_schema = preprocess_storage_jsonschema(contract_schemas['storageSchema'])
 
-                        write(storage_schema_path, json.dumps(storage_schema, indent=4, sort_keys=True))
+                        write(storage_schema_path, json.dumps(storage_schema, option=json.OPT_INDENT_2))
 
                         if not isinstance(operation_pattern_config, OperationHandlerTransactionPatternConfig):
                             continue
@@ -200,7 +200,7 @@ async def fetch_schemas(self) -> None:
 
                         entrypoint = entrypoint.replace('.', '_').lstrip('_')
                         entrypoint_schema_path = join(parameter_schemas_path, f'{entrypoint}.json')
-                        written = write(entrypoint_schema_path, json.dumps(entrypoint_schema, indent=4))
+                        written = write(entrypoint_schema_path, json.dumps(entrypoint_schema, option=json.OPT_INDENT_2))
                         if not written and contract_config.typename is not None:
                             with open(entrypoint_schema_path, 'r') as file:
                                 existing_schema = json.loads(file.read())
@@ -229,11 +229,11 @@ async def fetch_schemas(self) -> None:
                     big_map_path = big_map_handler_config.path.replace('.', '_')
                     big_map_key_schema = big_map_schema['keySchema']
                     big_map_key_schema_path = join(big_map_schemas_path, f'{big_map_path}_key.json')
-                    write(big_map_key_schema_path, json.dumps(big_map_key_schema, indent=4))
+                    write(big_map_key_schema_path, json.dumps(big_map_key_schema, option=json.OPT_INDENT_2))
 
                     big_map_value_schema = big_map_schema['valueSchema']
                     big_map_value_schema_path = join(big_map_schemas_path, f'{big_map_path}_value.json')
-                    write(big_map_value_schema_path, json.dumps(big_map_value_schema, indent=4))
+                    write(big_map_value_schema_path, json.dumps(big_map_value_schema, option=json.OPT_INDENT_2))
 
             elif isinstance(index_config, HeadIndexConfig):
                 pass

src/dipdup/codegen.py

@@ -942,10 +942,12 @@ def iter_imports(self, package: str) -> Iterator[Tuple[str, str]]:
 class AdvancedConfig:
     reindex: Dict[ReindexingReasonC, ReindexingAction] = field(default_factory=dict)
     scheduler: Optional[Dict[str, Any]] = None
+    # TODO: Drop in major version
     oneshot: bool = False
     postpone_jobs: bool = False
     early_realtime: bool = False
     merge_subscriptions: bool = False
+    metadata_interface: bool = False
 
 
 @dataclass
@@ -1137,9 +1139,15 @@ def initialize(self, skip_imports: bool = False) -> None:
             self._imports_resolved.add(index_config.name)
 
     def _validate(self) -> None:
-        # NOTE: Hasura
-        if isinstance(self.database, SqliteDatabaseConfig) and self.hasura:
-            raise ConfigurationError('SQLite database engine is not supported by Hasura')
+        # NOTE: Hasura and metadata interface
+        if self.hasura:
+            if isinstance(self.database, SqliteDatabaseConfig):
+                raise ConfigurationError('SQLite database engine is not supported by Hasura')
+            if self.advanced.metadata_interface and self.hasura.camel_case:
+                raise ConfigurationError('`metadata_interface` flag is incompatible with `camel_case` one')
+        else:
+            if self.advanced.metadata_interface:
+                raise ConfigurationError('`metadata_interface` flag requires `hasura` section to be present')
 
         # NOTE: Reserved hooks
         for name, hook_config in self.hooks.items():

src/dipdup/config.py

@@ -53,9 +53,11 @@
 from dipdup.exceptions import InitializationRequiredError
 from dipdup.exceptions import ReindexingRequiredError
 from dipdup.models import Contract
+from dipdup.models import ContractMetadata
 from dipdup.models import Index
 from dipdup.models import ReindexingReason
 from dipdup.models import Schema
+from dipdup.models import TokenMetadata
 from dipdup.utils import FormattedLogger
 from dipdup.utils import slowdown
 from dipdup.utils.database import execute_sql_scripts
@@ -68,6 +70,31 @@
 pending_hooks: Deque[Awaitable[None]] = deque()
 
 
+class MetadataCursor:
+    _contract = 0
+    _token = 0
+
+    def __new__(cls):
+        raise NotImplementedError
+
+    @classmethod
+    async def initialize(cls) -> None:
+        if last_contract := await ContractMetadata.filter().order_by('-update_id').first():
+            cls._contract = last_contract.update_id
+        if last_token := await TokenMetadata.filter().order_by('-update_id').first():
+            cls._token = last_token.update_id
+
+    @classmethod
+    def contract(cls) -> int:
+        cls._contract += 1
+        return cls._contract
+
+    @classmethod
+    def token(cls) -> int:
+        cls._token += 1
+        return cls._token
+
+
 # TODO: Dataclasses are cool, everyone loves them. Resolve issue with pydantic serialization.
 class DipDupContext:
     def __init__(
@@ -216,6 +243,41 @@ async def spawn_index(self, name: str, state: Optional[Index] = None) -> None:
         # NOTE: IndexDispatcher will handle further initialization when it's time
         pending_indexes.append(index)
 
+    async def update_contract_metadata(
+        self,
+        network: str,
+        address: str,
+        metadata: Dict[str, Any],
+    ) -> None:
+        if not self.config.advanced.metadata_interface:
+            return
+        update_id = MetadataCursor.contract()
+        await ContractMetadata.update_or_create(
+            network=network,
+            contract=address,
+            defaults={'metadata': metadata, 'update_id': update_id},
+        )
+
+    async def update_token_metadata(
+        self,
+        network: str,
+        address: str,
+        token_id: str,
+        metadata: Dict[str, Any],
+    ) -> None:
+        if not self.config.advanced.metadata_interface:
+            return
+        if not all(str.isdigit(c) for c in token_id):
+            raise ValueError('`token_id` must be a number')
+
+        update_id = MetadataCursor.token()
+        await TokenMetadata.update_or_create(
+            network=network,
+            contract=address,
+            token_id=token_id,
+            defaults={'metadata': metadata, 'update_id': update_id},
+        )
+
     def _get_datasource(self, name: str, type_: Type[DatasourceT]) -> DatasourceT:
         datasource = self.datasources.get(name)
         if not datasource:

src/dipdup/context.py

@@ -52,11 +52,18 @@ def __init__(self, url: str, http_config: HTTPConfig, merge_subscriptions: bool
         self._on_rollback: Set[RollbackCallbackT] = set()
         self._subscriptions: SubscriptionManager = SubscriptionManager(merge_subscriptions)
         self._subscriptions.add(HeadSubscription())
+        self._network: Optional[str] = None
 
     @property
     def name(self) -> str:
         return self._http._url
 
+    @property
+    def network(self) -> str:
+        if not self._network:
+            raise RuntimeError('Network is not set')
+        return self._network
+
     @abstractmethod
     async def subscribe(self) -> None:
         ...
@@ -89,6 +96,11 @@ async def emit_rollback(self, from_level: int, to_level: int) -> None:
         for fn in self._on_rollback:
             await fn(self, from_level, to_level)
 
+    def set_network(self, network: str) -> None:
+        if self._network:
+            raise RuntimeError('Network is already set')
+        self._network = network
+
     def set_sync_level(self, subscription: Optional[Subscription], level: int) -> None:
         self._subscriptions.set_sync_level(subscription, level)
 

src/dipdup/datasources/datasource.py

@@ -962,10 +962,13 @@ def convert_block(cls, block_json: Dict[str, Any]) -> BlockData:
     def convert_head_block(cls, head_block_json: Dict[str, Any]) -> HeadBlockData:
         """Convert raw head block message from WS/REST into dataclass"""
         return HeadBlockData(
+            chain=head_block_json['chain'],
+            chain_id=head_block_json['chainId'],
             cycle=head_block_json['cycle'],
             level=head_block_json['level'],
             hash=head_block_json['hash'],
             protocol=head_block_json['protocol'],
+            next_protocol=head_block_json['nextProtocol'],
             timestamp=cls._parse_timestamp(head_block_json['timestamp']),
             voting_epoch=head_block_json['votingEpoch'],
             voting_period=head_block_json['votingPeriod'],
@@ -980,6 +983,7 @@ def convert_head_block(cls, head_block_json: Dict[str, Any]) -> HeadBlockData:
             quote_jpy=Decimal(head_block_json['quoteJpy']),
             quote_krw=Decimal(head_block_json['quoteKrw']),
             quote_eth=Decimal(head_block_json['quoteEth']),
+            quote_gbp=Decimal(head_block_json['quoteGbp']),
         )
 
     @classmethod

src/dipdup/datasources/tzkt/datasource.py

@@ -36,6 +36,7 @@
 from dipdup.config import default_hooks
 from dipdup.context import CallbackManager
 from dipdup.context import DipDupContext
+from dipdup.context import MetadataCursor
 from dipdup.context import pending_indexes
 from dipdup.datasources.coinbase.datasource import CoinbaseDatasource
 from dipdup.datasources.datasource import Datasource
@@ -351,6 +352,9 @@ async def run(self) -> None:
             await self._initialize_datasources()
             await self._set_up_hasura(stack)
 
+            if advanced_config.metadata_interface:
+                await MetadataCursor.initialize()
+
             if self._config.oneshot:
                 start_scheduler_event, spawn_datasources_event = Event(), Event()
             else:
@@ -417,7 +421,8 @@ async def _initialize_schema(self) -> None:
         # TODO: Fix Tortoise ORM to raise more specific exception
         except KeyError:
             try:
-                # NOTE: A small migration, ReindexingReason became ReversedEnum
+                # TODO: Drop with major version bump
+                # NOTE: A small migration, ReindexingReason became ReversedEnum in 3.1.0
                 for item in ReindexingReason:
                     await conn.execute_script(f'UPDATE dipdup_schema SET reindex = "{item.name}" WHERE reindex = "{item.value}"')
 
@@ -491,9 +496,11 @@ async def _initialize_datasources(self) -> None:
             if not isinstance(datasource, TzktDatasource):
                 continue
 
+            head_block = await datasource.get_head_block()
+            datasource.set_network(head_block.chain)
             datasource.set_sync_level(
                 subscription=None,
-                level=(await datasource.get_head_block()).level,
+                level=head_block.level,
             )
 
             db_head = await Head.filter(name=datasource.name).first()