Abi fetching through StarknetNode datasource

CHANGELOG.md

@@ -15,6 +15,7 @@ Releases prior to 7.0 has been removed from this file to declutter search result
 ### Other
 
 - deps: `tortoise-orm` updated to 0.24.0.
+- starknet.node: Added methods for fetching contract ABI's
 
 ## [8.2.0rc1] - 2025-01-24
 

src/dipdup/abi/cairo.py

@@ -74,7 +74,6 @@ def sn_keccak(x: str) -> str:
     return f'0x{int.from_bytes(keccak_hash, "big") & (1 << 250) - 1:x}'
 
 
-@cache
 def _loaded_abis(package: DipDupPackage) -> dict[str, Abi]:
 
     from starknet_py.abi.v2.parser import AbiParser

src/dipdup/codegen/evm.py

@@ -1,4 +1,6 @@
+from itertools import chain
 from pathlib import Path
+from typing import TYPE_CHECKING
 from typing import Any
 from typing import cast
 
@@ -13,13 +15,14 @@
 from dipdup.config.evm_transactions import EvmTransactionsHandlerConfig
 from dipdup.config.evm_transactions import EvmTransactionsIndexConfig
 from dipdup.datasources import AbiDatasource
-from dipdup.exceptions import AbiNotAvailableError
 from dipdup.exceptions import ConfigurationError
-from dipdup.exceptions import DatasourceError
 from dipdup.utils import json_dumps
 from dipdup.utils import snake_to_pascal
 from dipdup.utils import touch
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
 
 class EvmCodeGenerator(CodeGenerator):
     kind = 'evm'
@@ -60,42 +63,42 @@ async def generate_handlers(self) -> None:
         pass
 
     async def _fetch_abi(self, index_config: EvmIndexConfigU) -> None:
-        datasource_configs = tuple(c for c in index_config.datasources if isinstance(c, EvmEtherscanDatasourceConfig))
+        contracts_from_event_handlers: list[EvmContractConfig] = [
+            handler_config.contract
+            for handler_config in index_config.handlers
+            if isinstance(handler_config, EvmEventsHandlerConfig)
+        ]
+        contracts_from_transactions: list[EvmContractConfig] = [
+            handler_config.typed_contract
+            for handler_config in index_config.handlers
+            if (
+                isinstance(handler_config, EvmTransactionsHandlerConfig)
+                and handler_config.typed_contract is not None
+            )
+        ]
+        contracts: Iterable[EvmContractConfig] = chain(contracts_from_event_handlers, contracts_from_transactions)
+
+        if not contracts:
+            self._logger.debug('No contract specified. No ABI to fetch.')
+            return
 
-        contract: EvmContractConfig | None = None
+        # deduplicated (by name) Datasource list
+        datasources: list[AbiDatasource[Any]] = list(
+            {
+                datasource_config.name: cast(AbiDatasource[Any], self._datasources[datasource_config.name])
+                for datasource_config in index_config.datasources
+                if isinstance(datasource_config, EvmEtherscanDatasourceConfig)
+            }.values()
+        )
 
-        for handler_config in index_config.handlers:
-            if isinstance(handler_config, EvmEventsHandlerConfig):
-                contract = handler_config.contract
-            elif isinstance(handler_config, EvmTransactionsHandlerConfig):
-                contract = handler_config.typed_contract
-
-            if not contract:
-                continue
+        if not datasources:
+            raise ConfigurationError('No EVM ABI datasources found')
 
+        async for contract, abi_json in AbiDatasource.lookup_abi_for(contracts, using=datasources, logger=self._logger):
             abi_path = self._package.abi / contract.module_name / 'abi.json'
+
             if abi_path.exists():
                 continue
-            if not datasource_configs:
-                raise ConfigurationError('No EVM ABI datasources found')
-
-            address = contract.address or contract.abi
-            if not address:
-                raise ConfigurationError(f'`address` or `abi` must be specified for contract `{contract.module_name}`')
-
-            for datasource_config in datasource_configs:
-
-                datasource = cast(AbiDatasource[Any], self._datasources[datasource_config.name])
-                try:
-                    abi_json = await datasource.get_abi(address)
-                    break
-                except DatasourceError as e:
-                    self._logger.warning('Failed to fetch ABI from `%s`: %s', datasource_config.name, e)
-            else:
-                raise AbiNotAvailableError(
-                    address=address,
-                    typename=contract.module_name,
-                )
 
             touch(abi_path)
             abi_path.write_bytes(json_dumps(abi_json))

src/dipdup/codegen/starknet.py

@@ -1,16 +1,59 @@
 from pathlib import Path
+from typing import Any
+from typing import cast
 
 from dipdup.codegen import CodeGenerator
 from dipdup.config import HandlerConfig
+from dipdup.config.starknet import StarknetContractConfig
+from dipdup.config.starknet_events import StarknetEventsHandlerConfig
 from dipdup.config.starknet_events import StarknetEventsIndexConfig
+from dipdup.config.starknet_node import StarknetNodeDatasourceConfig
+from dipdup.datasources import AbiDatasource
+from dipdup.exceptions import ConfigurationError
+from dipdup.utils import json_dumps
 from dipdup.utils import snake_to_pascal
+from dipdup.utils import touch
 
 
 class StarknetCodeGenerator(CodeGenerator):
     kind = 'starknet'
 
-    # NOTE: For now ABIs need to be provided manually
-    async def generate_abis(self) -> None: ...
+    async def generate_abis(self) -> None:
+        for index_config in self._config.indexes.values():
+            if isinstance(index_config, StarknetEventsIndexConfig):
+                await self._fetch_abi(index_config)
+
+    async def _fetch_abi(self, index_config: StarknetEventsIndexConfig) -> None:
+        contracts: list[StarknetContractConfig] = [
+            handler_config.contract
+            for handler_config in index_config.handlers
+            if isinstance(handler_config, StarknetEventsHandlerConfig)
+        ]
+
+        if not contracts:
+            self._logger.debug('No contract specified. No ABI to fetch.')
+            return
+
+        # deduplicated (by name) Datasource list
+        datasources: list[AbiDatasource[Any]] = list(
+            {
+                datasource_config.name: cast(AbiDatasource[Any], self._datasources[datasource_config.name])
+                for datasource_config in index_config.datasources
+                if isinstance(datasource_config, StarknetNodeDatasourceConfig)
+            }.values()
+        )
+
+        if not datasources:
+            raise ConfigurationError('No Starknet ABI datasources found')
+
+        async for contract, abi_json in AbiDatasource.lookup_abi_for(contracts, using=datasources, logger=self._logger):
+            abi_path = self._package.abi / contract.module_name / 'cairo_abi.json'
+
+            if abi_path.exists():
+                continue
+
+            touch(abi_path)
+            abi_path.write_bytes(json_dumps(abi_json))
 
     async def generate_schemas(self) -> None:
         from dipdup.abi.cairo import abi_to_jsonschemas

src/dipdup/datasources/__init__.py

@@ -1,8 +1,11 @@
 import asyncio
 import time
 from abc import abstractmethod
+from collections.abc import AsyncIterator
 from collections.abc import Awaitable
 from collections.abc import Callable
+from collections.abc import Iterable
+from logging import Logger
 from typing import Any
 from typing import Generic
 from typing import TypeVar
@@ -15,6 +18,10 @@
 from dipdup.config import HttpConfig
 from dipdup.config import IndexConfig
 from dipdup.config import ResolvedHttpConfig
+from dipdup.config.evm import EvmContractConfig
+from dipdup.config.starknet import StarknetContractConfig
+from dipdup.exceptions import AbiNotAvailableError
+from dipdup.exceptions import ConfigurationError
 from dipdup.exceptions import DatasourceError
 from dipdup.exceptions import FrameworkException
 from dipdup.http import HTTPGateway
@@ -29,10 +36,12 @@
 from dipdup.utils import FormattedLogger
 
 DatasourceConfigT = TypeVar('DatasourceConfigT', bound=DatasourceConfig)
+ContractConfigT = TypeVar('ContractConfigT', bound=StarknetContractConfig | EvmContractConfig)
 
 
 EmptyCallback = Callable[[], Awaitable[None]]
 RollbackCallback = Callable[['IndexDatasource[Any]', MessageType, int, int], Awaitable[None]]
+AbiJson = dict[str, Any] | list[Any]
 
 
 class Datasource(HTTPGateway, Generic[DatasourceConfigT]):
@@ -57,8 +66,38 @@ async def run(self) -> None:
 
 
 class AbiDatasource(Datasource[DatasourceConfigT], Generic[DatasourceConfigT]):
+
     @abstractmethod
-    async def get_abi(self, address: str) -> dict[str, Any]: ...
+    async def get_abi(self, address: str) -> AbiJson: ...
+
+    @staticmethod
+    async def lookup_abi_for(
+        contracts: Iterable[ContractConfigT], using: list['AbiDatasource[DatasourceConfigT]'], logger: Logger
+    ) -> AsyncIterator[tuple[ContractConfigT, AbiJson]]:
+        """For every contract goes over each datasourse and tries to obtain abi file.
+        If no ABI exists for any of the contracts - raises error.
+        """
+        for contract in contracts:
+            abi_json = None
+
+            address = contract.address or contract.abi
+            if not address:
+                raise ConfigurationError(f'`address` or `abi` must be specified for contract `{contract.module_name}`')
+
+            for datasource in using:
+                try:
+                    abi_json = await datasource.get_abi(address=address)
+                    break
+                except DatasourceError as e:
+                    logger.warning('Failed to fetch ABI from `%s`: %s', datasource.name, e)
+
+            if abi_json is None:
+                raise AbiNotAvailableError(
+                    address=address,
+                    typename=contract.module_name,
+                )
+
+            yield (contract, abi_json)
 
 
 # FIXME: inconsistent usage

src/dipdup/datasources/evm_etherscan.py

@@ -24,7 +24,7 @@ class EvmEtherscanDatasource(AbiDatasource[EvmEtherscanDatasourceConfig]):
     async def run(self) -> None:
         pass
 
-    async def get_abi(self, address: str) -> dict[str, Any]:
+    async def get_abi(self, address: str) -> dict[str, Any] | list[Any]:
         params = {
             'module': 'contract',
             'action': 'getabi',

src/dipdup/datasources/starknet_node.py

@@ -3,13 +3,16 @@
 
 from dipdup.config import HttpConfig
 from dipdup.config.starknet_node import StarknetNodeDatasourceConfig
+from dipdup.datasources import AbiJson
 from dipdup.datasources import IndexDatasource
+from dipdup.datasources._starknetpy import StarknetpyClient
 
 if TYPE_CHECKING:
+    from starknet_py.abi.v0.shape import AbiDictList as AbiDictListV0
+    from starknet_py.abi.v1.shape import AbiDictList as AbiDictListV1
+    from starknet_py.abi.v2.shape import AbiDictList as AbiDictListV2
     from starknet_py.net.client_models import EventsChunk
 
-    from dipdup.datasources._starknetpy import StarknetpyClient
-
 
 class StarknetNodeDatasource(IndexDatasource[StarknetNodeDatasourceConfig]):
     NODE_LAST_MILE = 128
@@ -70,3 +73,19 @@ async def get_events(
             chunk_size=self._http_config.batch_size,
             continuation_token=continuation_token,
         )
+
+    async def get_abi(self, address: str) -> AbiJson:
+        from starknet_py.net.client_models import DeprecatedContractClass
+        from starknet_py.net.client_models import SierraContractClass
+
+        class_at_response = await self.starknetpy.get_class_at(address, block_number='latest')
+        # NOTE: for some reason
+        parsed_abi: AbiDictListV0 | None | AbiDictListV1 | AbiDictListV2
+        if isinstance(class_at_response, SierraContractClass):
+            parsed_abi = class_at_response.parsed_abi
+        elif isinstance(class_at_response, DeprecatedContractClass):
+            parsed_abi = class_at_response.abi
+        else:
+            raise NotImplementedError(f'Unknown response class: {class_at_response}')
+
+        return parsed_abi or []

src/dipdup/datasources/substrate_subscan.py

@@ -3,11 +3,12 @@
 
 from dipdup.config.substrate_subscan import SubstrateSubscanDatasourceConfig
 from dipdup.datasources import AbiDatasource
+from dipdup.datasources import AbiJson
 
 
 class SubstrateSubscanDatasource(AbiDatasource[SubstrateSubscanDatasourceConfig]):
     # FIXME: not used in codegen
-    async def get_abi(self, address: str) -> dict[str, Any]:
+    async def get_abi(self, address: str) -> AbiJson:
         raise NotImplementedError
 
     async def run(self) -> None:

src/dipdup/fetcher.py

@@ -42,7 +42,7 @@ def block_number(self) -> int:
 
 
 async def yield_by_level(
-    iterable: AsyncIterator[tuple[BufferT, ...]]
+    iterable: AsyncIterator[tuple[BufferT, ...]],
 ) -> AsyncGenerator[tuple[Level, tuple[BufferT, ...]], None]:
     items: tuple[BufferT, ...] = ()