hint chia.rpc.full_node_rpc_client (#14705)

hint chia.rpc.rull_node_rpc_client

chia/cmds/netspace_funcs.py

@@ -4,7 +4,7 @@ from typing import Optional
 
 from chia.cmds.cmds_util import get_any_service_client
 from chia.rpc.full_node_rpc_client import FullNodeRpcClient
-from chia.util.byte_types import hexstr_to_bytes
+from chia.types.blockchain_format.sized_bytes import bytes32
 from chia.util.misc import format_bytes
 
 
@@ -24,7 +24,7 @@ async def netstorge_async(rpc_port: Optional[int], delta_block_height: str, star
 
                     newer_block_height = blockchain_state["peak"].height
                 else:
-                    newer_block = await client.get_block_record(hexstr_to_bytes(start))
+                    newer_block = await client.get_block_record(bytes32.from_hexstr(start))
                     if newer_block is None:
                         print("Block header hash", start, "not found.")
                         return None
@@ -39,7 +39,6 @@ async def netstorge_async(rpc_port: Optional[int], delta_block_height: str, star
                 network_space_bytes_estimate = await client.get_network_space(
                     newer_block_header.header_hash, older_block_header.header_hash
                 )
-                assert network_space_bytes_estimate is not None
                 print(
                     "Older Block\n"
                     f"Block Height: {older_block_header.height}\n"

chia/cmds/show.py

@@ -34,9 +34,7 @@ from chia.cmds.show_funcs import show_async
 @click.option(
     "-r", "--remove-connection", help="Remove a Node by the first 8 characters of NodeID", type=str, default=""
 )
-@click.option(
-    "-bh", "--block-header-hash-by-height", help="Look up a block header hash by block height", type=str, default=""
-)
+@click.option("-bh", "--block-header-hash-by-height", help="Look up a block header hash by block height", type=int)
 @click.option("-b", "--block-by-header-hash", help="Look up a block by block header hash", type=str, default="")
 @click.pass_context
 def show_cmd(
@@ -48,7 +46,7 @@ def show_cmd(
     connections: bool,
     add_connection: str,
     remove_connection: str,
-    block_header_hash_by_height: str,
+    block_header_hash_by_height: Optional[int],
     block_by_header_hash: str,
 ) -> None:
     import asyncio

chia/cmds/show_funcs.py

@@ -99,10 +99,9 @@ async def print_block_from_hash(
     from chia.types.blockchain_format.sized_bytes import bytes32
     from chia.types.full_block import FullBlock
     from chia.util.bech32m import encode_puzzle_hash
-    from chia.util.byte_types import hexstr_to_bytes
 
-    block: Optional[BlockRecord] = await node_client.get_block_record(hexstr_to_bytes(block_by_header_hash))
-    full_block: Optional[FullBlock] = await node_client.get_block(hexstr_to_bytes(block_by_header_hash))
+    block: Optional[BlockRecord] = await node_client.get_block_record(bytes32.from_hexstr(block_by_header_hash))
+    full_block: Optional[FullBlock] = await node_client.get_block(bytes32.from_hexstr(block_by_header_hash))
     # Would like to have a verbose flag for this
     if block is not None:
         assert full_block is not None
@@ -191,7 +190,7 @@ async def show_async(
     root_path: Path,
     print_fee_info_flag: bool,
     print_state: bool,
-    block_header_hash_by_height: str,
+    block_header_hash_by_height: Optional[int],
     block_by_header_hash: str,
 ) -> None:
     from chia.cmds.cmds_util import get_any_service_client
@@ -206,7 +205,7 @@ async def show_async(
             if print_fee_info_flag:
                 await print_fee_info(node_client)
             # Get Block Information
-            if block_header_hash_by_height != "":
+            if block_header_hash_by_height is not None:
                 block_header = await node_client.get_block_record_by_height(block_header_hash_by_height)
                 if block_header is not None:
                     print(f"Header hash of block {block_header_hash_by_height}: {block_header.header_hash.hex()}")

chia/rpc/full_node_rpc_client.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, cast
 
 from chia.consensus.block_record import BlockRecord
 from chia.full_node.signage_point import SignagePoint
@@ -13,10 +13,10 @@ from chia.types.full_block import FullBlock
 from chia.types.spend_bundle import SpendBundle
 from chia.types.unfinished_header_block import UnfinishedHeaderBlock
 from chia.util.byte_types import hexstr_to_bytes
-from chia.util.ints import uint32, uint64
+from chia.util.ints import uint32
 
 
-def coin_record_dict_backwards_compat(coin_record: Dict[str, Any]):
+def coin_record_dict_backwards_compat(coin_record: Dict[str, Any]) -> Dict[str, Any]:
     del coin_record["spent"]
     return coin_record
 
@@ -30,13 +30,13 @@ class FullNodeRpcClient(RpcClient):
     to the full node.
     """
 
-    async def get_blockchain_state(self) -> Dict:
+    async def get_blockchain_state(self) -> Dict[str, Any]:
         response = await self.fetch("get_blockchain_state", {})
         if response["blockchain_state"]["peak"] is not None:
             response["blockchain_state"]["peak"] = BlockRecord.from_json_dict(response["blockchain_state"]["peak"])
-        return response["blockchain_state"]
+        return cast(Dict[str, Any], response["blockchain_state"])
 
-    async def get_block(self, header_hash) -> Optional[FullBlock]:
+    async def get_block(self, header_hash: bytes32) -> Optional[FullBlock]:
         try:
             response = await self.fetch("get_block", {"header_hash": header_hash.hex()})
         except Exception:
@@ -49,14 +49,14 @@ class FullNodeRpcClient(RpcClient):
         )
         return [FullBlock.from_json_dict(block) for block in response["blocks"]]
 
-    async def get_block_record_by_height(self, height) -> Optional[BlockRecord]:
+    async def get_block_record_by_height(self, height: int) -> Optional[BlockRecord]:
         try:
             response = await self.fetch("get_block_record_by_height", {"height": height})
         except Exception:
             return None
         return BlockRecord.from_json_dict(response["block_record"])
 
-    async def get_block_record(self, header_hash) -> Optional[BlockRecord]:
+    async def get_block_record(self, header_hash: bytes32) -> Optional[BlockRecord]:
         try:
             response = await self.fetch("get_block_record", {"header_hash": header_hash.hex()})
             if response["block_record"] is None:
@@ -73,20 +73,16 @@ class FullNodeRpcClient(RpcClient):
         response = await self.fetch("get_blocks", {"start": start, "end": end, "exclude_header_hash": True})
         return [FullBlock.from_json_dict(r) for r in response["blocks"]]
 
-    async def get_network_space(
-        self, newer_block_header_hash: bytes32, older_block_header_hash: bytes32
-    ) -> Optional[uint64]:
-        try:
-            network_space_bytes_estimate = await self.fetch(
-                "get_network_space",
-                {
-                    "newer_block_header_hash": newer_block_header_hash.hex(),
-                    "older_block_header_hash": older_block_header_hash.hex(),
-                },
-            )
-        except Exception:
-            return None
-        return network_space_bytes_estimate["space"]
+    async def get_network_space(self, newer_block_header_hash: bytes32, older_block_header_hash: bytes32) -> int:
+        network_space_bytes_estimate = await self.fetch(
+            "get_network_space",
+            {
+                "newer_block_header_hash": newer_block_header_hash.hex(),
+                "older_block_header_hash": older_block_header_hash.hex(),
+            },
+        )
+
+        return cast(int, network_space_bytes_estimate["space"])
 
     async def get_coin_record_by_name(self, coin_id: bytes32) -> Optional[CoinRecord]:
         try:
@@ -102,7 +98,7 @@ class FullNodeRpcClient(RpcClient):
         include_spent_coins: bool = True,
         start_height: Optional[int] = None,
         end_height: Optional[int] = None,
-    ) -> List:
+    ) -> List[CoinRecord]:
         names_hex = [name.hex() for name in names]
         d = {"names": names_hex, "include_spent_coins": include_spent_coins}
         if start_height is not None:
@@ -119,7 +115,7 @@ class FullNodeRpcClient(RpcClient):
         include_spent_coins: bool = True,
         start_height: Optional[int] = None,
         end_height: Optional[int] = None,
-    ) -> List:
+    ) -> List[CoinRecord]:
         d = {"puzzle_hash": puzzle_hash.hex(), "include_spent_coins": include_spent_coins}
         if start_height is not None:
             d["start_height"] = start_height
@@ -135,7 +131,7 @@ class FullNodeRpcClient(RpcClient):
         include_spent_coins: bool = True,
         start_height: Optional[int] = None,
         end_height: Optional[int] = None,
-    ) -> List:
+    ) -> List[CoinRecord]:
         puzzle_hashes_hex = [ph.hex() for ph in puzzle_hashes]
         d = {"puzzle_hashes": puzzle_hashes_hex, "include_spent_coins": include_spent_coins}
         if start_height is not None:
@@ -152,7 +148,7 @@ class FullNodeRpcClient(RpcClient):
         include_spent_coins: bool = True,
         start_height: Optional[int] = None,
         end_height: Optional[int] = None,
-    ) -> List:
+    ) -> List[CoinRecord]:
         parent_ids_hex = [pid.hex() for pid in parent_ids]
         d = {"parent_ids": parent_ids_hex, "include_spent_coins": include_spent_coins}
         if start_height is not None:
@@ -169,7 +165,7 @@ class FullNodeRpcClient(RpcClient):
         include_spent_coins: bool = True,
         start_height: Optional[int] = None,
         end_height: Optional[int] = None,
-    ) -> List:
+    ) -> List[CoinRecord]:
         d = {"hint": hint.hex(), "include_spent_coins": include_spent_coins}
         if start_height is not None:
             d["start_height"] = start_height
@@ -192,7 +188,7 @@ class FullNodeRpcClient(RpcClient):
             additions.append(CoinRecord.from_json_dict(coin_record_dict_backwards_compat(coin_record)))
         return additions, removals
 
-    async def get_block_records(self, start: int, end: int) -> List:
+    async def get_block_records(self, start: int, end: int) -> List[Dict[str, Any]]:
         try:
             response = await self.fetch("get_block_records", {"start": start, "end": end})
             if response["block_records"] is None:
@@ -200,7 +196,7 @@ class FullNodeRpcClient(RpcClient):
         except Exception:
             return []
         # TODO: return block records
-        return response["block_records"]
+        return cast(List[Dict[str, Any]], response["block_records"])
 
     async def get_block_spends(self, header_hash: bytes32) -> Optional[List[CoinSpend]]:
         try:
@@ -212,7 +208,7 @@ class FullNodeRpcClient(RpcClient):
         except Exception:
             return None
 
-    async def push_tx(self, spend_bundle: SpendBundle):
+    async def push_tx(self, spend_bundle: SpendBundle) -> Dict[str, Any]:
         return await self.fetch("push_tx", {"spend_bundle": spend_bundle.to_json_dict()})
 
     async def get_puzzle_and_solution(self, coin_id: bytes32, height: uint32) -> Optional[CoinSpend]:
@@ -226,19 +222,23 @@ class FullNodeRpcClient(RpcClient):
         response = await self.fetch("get_all_mempool_tx_ids", {})
         return [bytes32(hexstr_to_bytes(tx_id_hex)) for tx_id_hex in response["tx_ids"]]
 
-    async def get_all_mempool_items(self) -> Dict[bytes32, Dict]:
-        response: Dict = await self.fetch("get_all_mempool_items", {})
-        converted: Dict[bytes32, Dict] = {}
+    async def get_all_mempool_items(self) -> Dict[bytes32, Dict[str, Any]]:
+        response = await self.fetch("get_all_mempool_items", {})
+        converted: Dict[bytes32, Dict[str, Any]] = {}
         for tx_id_hex, item in response["mempool_items"].items():
             converted[bytes32(hexstr_to_bytes(tx_id_hex))] = item
         return converted
 
-    async def get_mempool_item_by_tx_id(self, tx_id: bytes32, include_pending: bool = False) -> Optional[Dict]:
+    async def get_mempool_item_by_tx_id(
+        self,
+        tx_id: bytes32,
+        include_pending: bool = False,
+    ) -> Optional[Dict[str, Any]]:
         try:
             response = await self.fetch(
                 "get_mempool_item_by_tx_id", {"tx_id": tx_id.hex(), "include_pending": include_pending}
             )
-            return response["mempool_item"]
+            return cast(Dict[str, Any], response["mempool_item"])
         except Exception:
             return None