gnosis/eth/ethereum_client.py
def batch_call_custom(
        self,
        payloads: Iterable[Dict[str, Any]],
        raise_exception: bool = True,
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> List[Optional[Any]]:
        """
        Do batch requests of multiple contract calls

        :param payloads: Iterable of Dictionaries with at least {'data': '<hex-string>',
            'output_type': <solidity-output-type>, 'to': '<checksummed-address>'}. `from` can also be provided and if
            `fn_name` is provided it will be used for debugging purposes
        :param raise_exception: If False, exception will not be raised if there's any problem and instead `None` will
            be returned as the value
        :param block_identifier: `latest` by default
        :return: List with the ABI decoded return values
        :raises: ValueError if raise_exception=True
        """
        if not payloads:
            return []

        queries = []
        for i, payload in enumerate(payloads):
            assert "data" in payload, "`data` not present"
            assert "to" in payload, "`to` not present"
            assert "output_type" in payload, "`output-type` not present"

            query_params = {"to": payload["to"], "data": payload["data"]}  # Balance of
            if "from" in payload:
                query_params["from"] = payload["from"]

            queries.append(
                {
                    "jsonrpc": "2.0",
                    "method": "eth_call",
                    "params": [
                        query_params,
                        hex(block_identifier)
                        if isinstance(block_identifier, int)
                        else block_identifier,
                    ],
                    "id": i,
                }
            )

        response = self.http_session.post(
            self.ethereum_node_url, json=queries, timeout=self.timeout
        )
        if not response.ok:
            raise ConnectionError(
                f"Error connecting to {self.ethereum_node_url}: {response.text}"
            )

        return_values: List[Optional[Any]] = []
        errors = []
        for payload, result in zip(
            payloads, sorted(response.json(), key=lambda x: x["id"])
        ):
            if "error" in result:
                fn_name = payload.get("fn_name", HexBytes(payload["data"]).hex())
                errors.append(f'`{fn_name}`: {result["error"]}')
                return_values.append(None)
            else:
                output_type = payload["output_type"]
                try:
                    decoded_values = self.w3.codec.decode_abi(
                        output_type, HexBytes(result["result"])
                    )
                    normalized_data = map_abi_data(
                        BASE_RETURN_NORMALIZERS, output_type, decoded_values
                    )
                    if len(normalized_data) == 1:
                        return_values.append(normalized_data[0])
                    else:
                        return_values.append(normalized_data)
                except (DecodingError, OverflowError):
                    fn_name = payload.get("fn_name", HexBytes(payload["data"]).hex())
                    errors.append(f"`{fn_name}`: DecodingError, cannot decode")
                    return_values.append(None)

        if errors and raise_exception:
            raise BatchCallFunctionFailed(f"Errors returned {errors}")
        else:
            return return_values

        Do batch requests of multiple contract calls

        :param payloads: Iterable of Dictionaries with at least {'data': '<hex-string>',
            'output_type': <solidity-output-type>, 'to': '<checksummed-address>'}. `from` can also be provided and if
            `fn_name` is provided it will be used for debugging purposes
        :param raise_exception: If False, exception will not be raised if there's any problem and instead `None` will
            be returned as the value
        :param block_identifier: `latest` by default
        :return: List with the ABI decoded return values
        :raises: ValueError if raise_exception=True
        
Do batch requests of multiple contract calls
Do batch requests of multiple contract calls
def batch_call_custom(
        self,
        payloads: Iterable[Dict[str, Any]],
        raise_exception: bool = True,
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> List[Optional[Any]]:
        if not payloads:
            return []
        queries = []
        for i, payload in enumerate(payloads):
            assert "data" in payload, "`data` not present"
            assert "to" in payload, "`to` not present"
            assert "output_type" in payload, "`output-type` not present"
            query_params = {"to": payload["to"], "data": payload["data"]}  
            if "from" in payload:
                query_params["from"] = payload["from"]
            queries.append(
                {
                    "jsonrpc": "2.0",
                    "method": "eth_call",
                    "params": [
                        query_params,
                        hex(block_identifier)
                        if isinstance(block_identifier, int)
                        else block_identifier,
                    ],
                    "id": i,
                }
            )
        response = self.http_session.post(
            self.ethereum_node_url, json=queries, timeout=self.timeout
        )
        if not response.ok:
            raise ConnectionError(
                f"Error connecting to {self.ethereum_node_url}: {response.text}"
            )
        return_values: List[Optional[Any]] = []
        errors = []
        for payload, result in zip(
            payloads, sorted(response.json(), key=lambda x: x["id"])
        ):
            if "error" in result:
                fn_name = payload.get("fn_name", HexBytes(payload["data"]).hex())
                errors.append(f'`{fn_name}`: {result["error"]}')
                return_values.append(None)
            else:
                output_type = payload["output_type"]
                try:
                    decoded_values = self.w3.codec.decode_abi(
                        output_type, HexBytes(result["result"])
                    )
                    normalized_data = map_abi_data(
                        BASE_RETURN_NORMALIZERS, output_type, decoded_values
                    )
                    if len(normalized_data) == 1:
                        return_values.append(normalized_data[0])
                    else:
                        return_values.append(normalized_data)
                except (DecodingError, OverflowError):
                    fn_name = payload.get("fn_name", HexBytes(payload["data"]).hex())
                    errors.append(f"`{fn_name}`: DecodingError, cannot decode")
                    return_values.append(None)
        if errors and raise_exception:
            raise BatchCallFunctionFailed(f"Errors returned {errors}")
        else:
            return return_values
def batch_call_custom ( self , payloads : Iterable [ Dict [ str , Any ] ] , raise_exception : bool = True , block_identifier : Optional [ BlockIdentifier ] = "latest" , ) -> List [ Optional [ Any ] ] : if not payloads : return [ ] queries = [ ] for i , payload in enumerate ( payloads ) : assert "data" in payload , "`data` not present" assert "to" in payload , "`to` not present" assert "output_type" in payload , "`output-type` not present" query_params = { "to" : payload [ "to" ] , "data" : payload [ "data" ] } if "from" in payload : query_params [ "from" ] = payload [ "from" ] queries . append ( { "jsonrpc" : "2.0" , "method" : "eth_call" , "params" : [ query_params , hex ( block_identifier ) if isinstance ( block_identifier , int ) else block_identifier , ] , "id" : i , } ) response = self . http_session . post ( self . ethereum_node_url , json = queries , timeout = self . timeout ) if not response . ok : raise ConnectionError ( f"Error connecting to {self.ethereum_node_url}: {response.text}" ) return_values : List [ Optional [ Any ] ] = [ ] errors = [ ] for payload , result in zip ( payloads , sorted ( response . json ( ) , key = lambda x : x [ "id" ] ) ) : if "error" in result : fn_name = payload . get ( "fn_name" , HexBytes ( payload [ "data" ] ) . hex ( ) ) errors . append ( f'`{fn_name}`: {result["error"]}' ) return_values . append ( None ) else : output_type = payload [ "output_type" ] try : decoded_values = self . w3 . codec . decode_abi ( output_type , HexBytes ( result [ "result" ] ) ) normalized_data = map_abi_data ( BASE_RETURN_NORMALIZERS , output_type , decoded_values ) if len ( normalized_data ) == 1 : return_values . append ( normalized_data [ 0 ] ) else : return_values . append ( normalized_data ) except ( DecodingError , OverflowError ) : fn_name = payload . get ( "fn_name" , HexBytes ( payload [ "data" ] ) . hex ( ) ) errors . append ( f"`{fn_name}`: DecodingError, cannot decode" ) return_values . append ( None ) if errors and raise_exception : raise BatchCallFunctionFailed ( f"Errors returned {errors}" ) else : return return_values
Do batch requests of multiple contract calls
Do batch requests of multiple contract calls
"""
        Do batch requests of multiple contract calls

        :param payloads: Iterable of Dictionaries with at least {'data': '<hex-string>',
            'output_type': <solidity-output-type>, 'to': '<checksummed-address>'}. `from` can also be provided and if
            `fn_name` is provided it will be used for debugging purposes
        :param raise_exception: If False, exception will not be raised if there's any problem and instead `None` will
            be returned as the value
        :param block_identifier: `latest` by default
        :return: List with the ABI decoded return values
        :raises: ValueError if raise_exception=True
        """ # Balance of
List with the ABI decoded return values
List with the ABI decoded return values
ValueError if raise_exception=True
ValueError if raise_exception = True
self
payloads
Iterable[Dict[str, Any]]
Iterable of Dictionaries with at least {'data': '',
'output_type': , 'to': ''}. `from` can also be provided and if
`fn_name` is provided it will be used for debugging purposes
Iterable of Dictionaries with at least { ' data ' : ' ' ' output_type ' : ' to ' : ' ' } . ` from ` can also be provided and if ` fn_name ` is provided it will be used for debugging purposes
raise_exception
bool
If False, exception will not be raised if there's any problem and instead `None` will
be returned as the value
If False exception will not be raised if there ' s any problem and instead ` None ` will be returned as the value
block_identifier
Optional[BlockIdentifier]
`latest` by default
` latest ` by default