Repository: Chia-Network/chia-blockchain
File: chia/full_node/full_node_rpc_client.py (Python)

Code before:

def coin_record_dict_backwards_compat(coin_record: dict[str, Any]) -> dict[str, Any]:
    del coin_record["spent"]
    return coin_record


class FullNodeRpcClient(RpcClient):
    """
    Client to Chia RPC, connects to a local full node. Uses HTTP/JSON, and converts back from
    JSON into native python objects before returning. All api calls use POST requests.
    Note that this is not the same as the peer protocol, or wallet protocol (which run Chia's
    protocol on top of TCP), it's a separate protocol on top of HTTP that provides easy access
    to the full node.
    """

    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 cast(dict[str, Any], response["blockchain_state"])

    async def get_block(self, header_hash: bytes32) -> FullBlock:
        try:
            response = await self.fetch("get_block", {"header_hash": header_hash.hex()})
        except ResponseFailureError as e:
            if "not found" in str(e):
                raise ValueError(f"Block {header_hash.hex()} not found") from e
            raise e
        return FullBlock.from_json_dict(response["block"])

    async def get_blocks(self, start: int, end: int, exclude_reorged: bool = False) -> list[FullBlock]:
        response = await self.fetch(
            "get_blocks", {"start": start, "end": end, "exclude_header_hash": True, "exclude_reorged": exclude_reorged}
        )
        return [FullBlock.from_json_dict(block) for block in response["blocks"]]

    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 ResponseFailureError as e:  # Block Height not found
            if "not found" in str(e):
                return None
            raise e
        return BlockRecord.from_json_dict(response["block_record"])

    async def get_block_record(self, header_hash: bytes32) -> Optional[BlockRecord]:
        response = await self.fetch("get_block_record", {"header_hash": header_hash.hex()})
        if response["block_record"] is None:
            return None
        return BlockRecord.from_json_dict(response["block_record"])


Review comment (bug):
this looks suspicious. what if you make a request to a node by hostname and you get "host not found". Is there a good reason to not pass through the `ResponseFailureError`?
This applies to the other places here too.

Code after:

def coin_record_dict_backwards_compat(coin_record: dict[str, Any]) -> dict[str, Any]:
    del coin_record["spent"]
    return coin_record


class FullNodeRpcClient(RpcClient):
    """
    Client to Chia RPC, connects to a local full node. Uses HTTP/JSON, and converts back from
    JSON into native python objects before returning. All api calls use POST requests.
    Note that this is not the same as the peer protocol, or wallet protocol (which run Chia's
    protocol on top of TCP), it's a separate protocol on top of HTTP that provides easy access
    to the full node.
    """

    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 cast(dict[str, Any], response["blockchain_state"])

    async def get_block(self, header_hash: bytes32) -> FullBlock:
        response = await self.fetch("get_block", {"header_hash": header_hash.hex()})
        return FullBlock.from_json_dict(response["block"])

    async def get_blocks(self, start: int, end: int, exclude_reorged: bool = False) -> list[FullBlock]:
        response = await self.fetch(
            "get_blocks", {"start": start, "end": end, "exclude_header_hash": True, "exclude_reorged": exclude_reorged}
        )
        return [FullBlock.from_json_dict(block) for block in response["blocks"]]

    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 ResponseFailureError as e:  # Block Height not found
            if e.response["error"] == f"Block height {height} not found in chain":
                return None
            raise e
        return BlockRecord.from_json_dict(response["block_record"])

    async def get_block_record(self, header_hash: bytes32) -> Optional[BlockRecord]:
        response = await self.fetch("get_block_record", {"header_hash": header_hash.hex()})
        if response["block_record"] is None:
            return None
        return BlockRecord.from_json_dict(response["block_record"])

    async def get_unfinished_block_headers(self) -> list[UnfinishedHeaderBlock]:
        response = await self.fetch("get_unfinished_block_headers", {})
        return [UnfinishedHeaderBlock.from_json_dict(r) for r in response["headers"]]

    async def get_all_block(self, start: uint32, end: uint32) -> list[FullBlock]: