etherscan: Native Balance

- Fetch Ethereum balance for debugging purposes

Author: nickninov

source-etherscan/sample_files/configured_catalog.json

@@ -26,6 +26,15 @@
       },
       "sync_mode": "incremental",
       "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "native_balance",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append"
     }
   ]
 }

source-etherscan/source_etherscan/schemas/native_balance.json

@@ -0,0 +1,44 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "NativeBalance",
+  "type": "object",
+  "additionalProperties": false,
+  "properties": {
+    "timestamp": {
+      "type": "string",
+      "format": "date-time",
+      "description": "Timestamp when the balance was fetched"
+    },
+    "wallet_address": {
+      "type": "string",
+      "description": "Ethereum wallet address"
+    },
+    "wallet_name": {
+      "type": ["string", "null"],
+      "description": "Human-readable wallet name"
+    },
+    "tags": {
+      "type": ["array", "null"],
+      "items": {
+        "type": "string"
+      },
+      "description": "Wallet tags"
+    },
+    "token_symbol": {
+      "type": "string",
+      "description": "Native token symbol (ETH)"
+    },
+    "token_decimal": {
+      "type": "integer",
+      "description": "Number of decimals for the native token"
+    },
+    "amount": {
+      "type": "string",
+      "description": "Native balance in smallest unit (wei)"
+    },
+    "chain_id": {
+      "type": "integer",
+      "description": "Blockchain chain ID"
+    }
+  }
+}

source-etherscan/source_etherscan/source.py

@@ -33,6 +33,7 @@ def __init__(self, api_key: str, wallets: list[dict], chain_id: str, **kwargs):
             }
             for wallet in self.wallets
         }
+        self.is_balance_stream = self.name.endswith('balance')
 
     def stream_slices(self, sync_mode: SyncMode, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None) -> Iterable[Optional[Mapping[str, Any]]]:
 
@@ -56,7 +57,8 @@ def stream_slices(self, sync_mode: SyncMode, cursor_field: List[str] = None, str
 
         for wallet in self.wallets:
             selected = self.historical_mapping[wallet["address"]]
-            self.logger.debug(f"{self.name} > stream_slice: Fetching data for {wallet['name']} from {selected['start_date']} to {selected['end_date']}")
+            msg = f"{self.name} > stream_slice: Fetching data for {wallet['name']}" + ("" if self.is_balance_stream else f" from {selected['start_date']} to {selected['end_date']}")
+            self.logger.info(msg)
             time.sleep(self.sleep_seconds)
             yield {
                 "address": wallet["address"],
@@ -78,11 +80,18 @@ def backoff_time(self, response: requests.Response) -> Optional[float]:
             match = re.search(r"\((\d+)\s*/", str(result))
             seconds = int(match.group(1))
 
+        elif self.is_balance_stream:
+            seconds = 1
+        
         self.logger.info(f"{self.name} > backoff_time: {seconds}s")
         return seconds
 
     def next_page_token(self, response: requests.Response):        
 
+        if self.is_balance_stream:
+            time.sleep(self.sleep_seconds)
+            return None
+
         result: Union[list[dict], str] = response.json().get("result", [])
 
         wallet_address = self.get_params(response).get("address")
@@ -123,10 +132,14 @@ def request_params(self, stream_state: Mapping[str, Any], stream_slice: Mapping[
             "module": "account",
             "offset": self.pagination_offset
         }
+        if self.is_balance_stream:
+            params.pop("sort")
+            params.pop("offset")
+
         if next_page_token:
             params["page"] = next_page_token["page"]
 
-        self.logger.debug(f"{self.name} > request_params: {params}")
+        self.logger.info(f"{self.name} > request_params: {params}")
         return params
 
     def to_datetime(self, timestamp: str) -> datetime.datetime:
@@ -204,7 +217,6 @@ def parse_response(self, response, *, stream_state: Mapping[str, Any], stream_sl
 
             yield point
 
-
 class WalletInternalTransactions(EtherscanStream):
     """
     This refers to a transfer of ETH that is carried out through a smart contract as an intermediary. 
@@ -261,7 +273,6 @@ def parse_response(self, response, *, stream_state: Mapping[str, Any], stream_sl
 
             yield point
 
-
 class WalletTokenTransactions(EtherscanStream):
     """
     Transactions of ERC-20 or ERC-721 tokens are labelled as Token Transfer transactions. 
@@ -336,6 +347,43 @@ def parse_response(self, response, *, stream_state: Mapping[str, Any], stream_sl
 
             yield point
 
+class NativeBalance(EtherscanStream):
+    """
+    Native balance (ETH) for the wallets
+
+    NOTE: Used for debugging purposes
+    """
+    primary_key = None
+    cursor_field = []
+
+    def __init__(self, api_key: str, wallets: list[dict], chain_id: str, **kwargs):
+        super().__init__(api_key, wallets, chain_id, **kwargs)
+
+    def request_params(self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None) -> MutableMapping[str, Any]:
+        params = {
+            **super().request_params(stream_state, stream_slice, next_page_token),
+            "action": "balance",
+        }
+        return params
+
+    def parse_response(self, response, *, stream_state: Mapping[str, Any], stream_slice: Optional[Mapping[str, Any]] = None, next_page_token: Optional[Mapping[str, Any]] = None):
+        
+        data: dict = response.json()
+        params = self.get_params(response)
+        wallet_address = params["address"]
+        wallet = self.wallet_info[wallet_address]
+        point = {
+            "timestamp": datetime.datetime.now(),
+            "wallet_address": wallet_address,
+            "wallet_name": wallet["name"],
+            "tags": wallet["tags"],
+            "token_symbol": "ETH",
+            "token_decimal": self.ETHEREUM_DECIMALS,
+            "amount": data["result"],
+            "chain_id": int(self.chain_id)
+        }
+        yield point
+
 class SourceEtherscan(AbstractSource):
 
     url = "https://api.etherscan.io/v2/api"
@@ -379,5 +427,6 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             WalletTransactions(**params),
             WalletInternalTransactions(**params),
             WalletTokenTransactions(**params),
+            NativeBalance(**params)
         ]
         return streams