lint, fix session management

src/dipdup/config/substrate.py

@@ -10,11 +10,13 @@
 from dipdup.config import Alias
 from dipdup.config import IndexConfig
 from dipdup.config import RuntimeConfig
+from dipdup.config.substrate_node import SubstrateNodeDatasourceConfig
 from dipdup.config.substrate_subscan import SubstrateSubscanDatasourceConfig
 from dipdup.config.substrate_subsquid import SubstrateSubsquidDatasourceConfig
-from dipdup.config.substrate_node import SubstrateNodeDatasourceConfig
 
-SubstrateDatasourceConfigU: TypeAlias = SubstrateSubsquidDatasourceConfig | SubstrateSubscanDatasourceConfig | SubstrateNodeDatasourceConfig
+SubstrateDatasourceConfigU: TypeAlias = (
+    SubstrateSubsquidDatasourceConfig | SubstrateSubscanDatasourceConfig | SubstrateNodeDatasourceConfig
+)
 
 
 @dataclass(config=ConfigDict(extra='forbid'), kw_only=True)

src/dipdup/datasources/__init__.py

@@ -223,7 +223,7 @@ async def _request() -> None:
 
     # TODO: probably should be defined higher
     @abstractmethod
-    async def get_head_level() -> int: ...
+    async def get_head_level(self) -> int: ...
 
 
 def create_datasource(config: DatasourceConfig) -> Datasource[Any]:

src/dipdup/datasources/substrate_node.py

@@ -7,6 +7,7 @@
 from copy import copy
 from dataclasses import dataclass
 from dataclasses import field
+from functools import partial
 from pathlib import Path
 from typing import Any
 
@@ -97,6 +98,9 @@ class SubstrateNodeDatasource(JsonRpcDatasource[SubstrateDatasourceConfigU]):
     def __init__(self, config: SubstrateDatasourceConfigU) -> None:
         from aiosubstrate.base import SubstrateInterface
 
+        # NOTE: Use our aiohttp session and limiters
+        SubstrateInterface.http_request = partial(self._jsonrpc_request, raw=True)  # type: ignore[method-assign]
+
         super().__init__(config)
         self._pending_subscription = None
         self._subscription_ids: dict[str, SubstrateNodeSubscription] = {}
@@ -219,9 +223,11 @@ async def get_block_hash(self, height: int) -> str:
 
     async def get_block_header(self, hash: str) -> BlockHeader:
         response = await self._jsonrpc_request('chain_getHeader', [hash])
-        return {'hash': hash,
-                'number': int(response['number'], 16),
-                'prevRoot': response['parentHash']}
+        return {
+            'hash': hash,
+            'number': int(response['number'], 16),
+            'prevRoot': response['parentHash'],
+        }
 
     async def get_metadata_header(self, height: int) -> MetadataHeader:
         block_hash = await self.get_block_hash(height)
@@ -245,14 +251,16 @@ async def get_events(self, block_hash: str) -> tuple[SubstrateEventDataDict, ...
         res_events: list[SubstrateEventDataDict] = []
         for event in events:
             event: dict = event.decode()
-            res_events.append({
-                'name': f'{event['module_id']}.{event['event_id']}',
-                'index': event['event_index'],
-                'extrinsicIndex': event['extrinsic_idx'],
-                'callAddress': None,
-                'args': None,
-                'decoded_args': event['attributes'],
-            })
+            res_events.append(
+                {
+                    'name': f'{event['module_id']}.{event['event_id']}',
+                    'index': event['event_index'],
+                    'extrinsicIndex': event['extrinsic_idx'],
+                    'callAddress': None,
+                    'args': None,
+                    'decoded_args': event['attributes'],
+                }
+            )
 
         return tuple(res_events)
 

src/dipdup/dipdup.py

@@ -558,7 +558,9 @@ async def _on_substrate_head(self, datasource: SubstrateNodeDatasource, head: He
         pass
 
     # TODO: fix data typing
-    async def _on_substrate_events(self, datasource: SubstrateNodeDatasource, events: tuple[dict, ...]) -> None:
+    async def _on_substrate_events(
+        self, datasource: SubstrateNodeDatasource, events: tuple[dict[str, Any], ...]
+    ) -> None:
         for index in self._indexes.values():
             if isinstance(index, SubstrateEventsIndex) and datasource in index.datasources:
                 index.push_realtime_message(events)

src/dipdup/indexes/substrate_events/fetcher.py

@@ -5,7 +5,6 @@
 from dipdup.indexes.substrate_node import SubstrateNodeFetcher
 from dipdup.indexes.substrate_subsquid import SubstrateSubsquidFetcher
 from dipdup.models.substrate import SubstrateEventData
-from dipdup.runtimes import SubstrateRuntime
 
 
 class SubstrateSubsquidEventFetcher(SubstrateSubsquidFetcher[SubstrateEventData]):
@@ -59,5 +58,4 @@ async def fetch_events(self) -> AsyncIterator[tuple[tuple[SubstrateEventData, ..
             block_hash = await self.get_random_node().get_block_hash(level)
             event_dicts = await self.get_random_node().get_events(block_hash)
             block_header = await self.get_random_node().get_block_header(block_hash)
-            yield tuple(SubstrateEventData(**event_dict, header=block_header)
-                        for event_dict in event_dicts)
+            yield tuple(SubstrateEventData(**event_dict, header=block_header) for event_dict in event_dicts)

src/dipdup/indexes/substrate_node.py

@@ -1,19 +1,12 @@
-from dipdup.datasources.substrate_node import SubstrateNodeDatasource
-import asyncio
 import logging
 import random
-from collections.abc import AsyncIterator
 from abc import ABC
-from collections import defaultdict
-from collections import deque
-from typing import Any
 from typing import Generic
 
+from dipdup.datasources.substrate_node import SubstrateNodeDatasource
 from dipdup.exceptions import FrameworkException
 from dipdup.fetcher import BufferT
 from dipdup.fetcher import DataFetcher
-from dipdup.models.substrate import SubstrateEventData
-
 
 SUBSTRATE_NODE_READAHEAD_LIMIT = 2500
 

src/dipdup/models/substrate_node.py

@@ -6,12 +6,11 @@
 from dipdup.subscriptions import Subscription
 
 
-class SubstrateNodeSubscription(ABC, Subscription):
-    ...
+class SubstrateNodeSubscription(ABC, Subscription): ...
 
 
 @dataclass(frozen=True)
 class SubstrateNodeHeadSubscription(SubstrateNodeSubscription):
     method: Literal['chain_subscribeFinalisedHeads'] = 'chain_subscribeFinalisedHeads'
-    # NOTE: used to determine which objects index require, since we can only subscribe to head 
+    # NOTE: used to determine which objects index require, since we can only subscribe to head
     fetch_events: bool = False