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

Code before:
        result = await self.fetch("get_aggsig_additional_data", {})
        return bytes32.from_hexstr(result["additional_data"])

    async def get_coin_records_by_hint(
        self,
        hint: bytes32,
        include_spent_coins: bool = True,
        start_height: Optional[int] = None,
        end_height: Optional[int] = None,
    ) -> list[CoinRecord]:
        d = {"hint": hint.hex(), "include_spent_coins": include_spent_coins}
        if start_height is not None:
            d["start_height"] = start_height
        if end_height is not None:
            d["end_height"] = end_height

        response = await self.fetch("get_coin_records_by_hint", d)
        return [CoinRecord.from_json_dict(coin_record_dict_backwards_compat(coin)) for coin in response["coin_records"]]

    async def get_additions_and_removals(
        self, header_hash: bytes32
    ) -> Optional[tuple[list[CoinRecord], list[CoinRecord]]]:
        try:
            response = await self.fetch("get_additions_and_removals", {"header_hash": header_hash.hex()})
        except Exception:
            return None
        removals = []
        additions = []
        for coin_record in response["removals"]:
            removals.append(CoinRecord.from_json_dict(coin_record_dict_backwards_compat(coin_record)))
        for coin_record in response["additions"]:
            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[dict[str, Any]]:
        try:
            response = await self.fetch("get_block_records", {"start": start, "end": end})
            if response["block_records"] is None:
                return []
        except Exception:
            return []
        # TODO: return block records
        return cast(list[dict[str, Any]], response["block_records"])

    async def get_block_spends(self, header_hash: bytes32) -> Optional[list[CoinSpend]]:
        try:
            response = await self.fetch("get_block_spends", {"header_hash": header_hash.hex()})
            block_spends = []
            for block_spend in response["block_spends"]:
                block_spends.append(CoinSpend.from_json_dict(block_spend))
            return block_spends

Review comment (suggestion):
I see. The RPC client swallows errors. I think a better solution would be to not catch the exception here, and let it propagate to the caller

Code after:
        result = await self.fetch("get_aggsig_additional_data", {})
        return bytes32.from_hexstr(result["additional_data"])

    async def get_coin_records_by_hint(
        self,
        hint: bytes32,
        include_spent_coins: bool = True,
        start_height: Optional[int] = None,
        end_height: Optional[int] = None,
    ) -> list[CoinRecord]:
        d = {"hint": hint.hex(), "include_spent_coins": include_spent_coins}
        if start_height is not None:
            d["start_height"] = start_height
        if end_height is not None:
            d["end_height"] = end_height

        response = await self.fetch("get_coin_records_by_hint", d)
        return [CoinRecord.from_json_dict(coin_record_dict_backwards_compat(coin)) for coin in response["coin_records"]]

    async def get_additions_and_removals(self, header_hash: bytes32) -> tuple[list[CoinRecord], list[CoinRecord]]:
        try:
            response = await self.fetch("get_additions_and_removals", {"header_hash": header_hash.hex()})
        except Exception:
            return [], []
        removals = []
        additions = []
        for coin_record in response["removals"]:
            removals.append(CoinRecord.from_json_dict(coin_record_dict_backwards_compat(coin_record)))
        for coin_record in response["additions"]:
            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[dict[str, Any]]:
        try:
            response = await self.fetch("get_block_records", {"start": start, "end": end})
            if response["block_records"] is None:
                return []
        except Exception:
            return []
        # TODO: return block records
        return cast(list[dict[str, Any]], response["block_records"])

    async def get_block_spends(self, header_hash: bytes32) -> Optional[list[CoinSpend]]:
        try:
            response = await self.fetch("get_block_spends", {"header_hash": header_hash.hex()})
            block_spends = []
            for block_spend in response["block_spends"]:
                block_spends.append(CoinSpend.from_json_dict(block_spend))
            return block_spends
        except Exception:
            return None