from typing import (
Any,
Dict,
Sequence,
)
from toolz import (
assoc,
dissoc,
)
from eth_account._utils.validation import (
is_rlp_structured_access_list,
is_rpc_structured_access_list,
)
def set_transaction_type_if_needed(transaction_dict: Dict[str, Any]) -> Dict[str, Any]:
if 'type' not in transaction_dict:
if 'gasPrice' in transaction_dict and 'accessList' in transaction_dict:
transaction_dict = assoc(transaction_dict, 'type', '0x1')
elif 'maxFeePerGas' in transaction_dict and 'maxPriorityFeePerGas' in transaction_dict:
transaction_dict = assoc(transaction_dict, 'type', '0x2')
return transaction_dict
def transaction_rpc_to_rlp_structure(dictionary: Dict[str, Any]) -> Dict[str, Any]:
access_list = dictionary.get('accessList')
if access_list:
dictionary = dissoc(dictionary, 'accessList')
rlp_structured_access_list = _access_list_rpc_to_rlp_structure(access_list)
dictionary = assoc(dictionary, 'accessList', rlp_structured_access_list)
return dictionary
def _access_list_rpc_to_rlp_structure(access_list: Sequence) -> Sequence:
if not is_rpc_structured_access_list(access_list):
raise ValueError("provided object not formatted as JSON-RPC-structured access list")
rlp_structured_access_list = []
for d in access_list:
rlp_structured_access_list.append(
(
d['address'], tuple(_ for _ in d['storageKeys']) )
)
return tuple(rlp_structured_access_list)
def transaction_rlp_to_rpc_structure(dictionary: Dict[str, Any]) -> Dict[str, Any]:
access_list = dictionary.get('accessList')
if access_list:
dictionary = dissoc(dictionary, 'accessList')
rpc_structured_access_list = _access_list_rlp_to_rpc_structure(access_list)
dictionary = assoc(dictionary, 'accessList', rpc_structured_access_list)
return dictionary
def _access_list_rlp_to_rpc_structure(access_list: Sequence) -> Sequence:
if not is_rlp_structured_access_list(access_list):
raise ValueError("provided object not formatted as rlp-structured access list")
rpc_structured_access_list = []
for t in access_list:
rpc_structured_access_list.append(
{
'address': t[0],
'storageKeys': t[1]
}
)
return tuple(rpc_structured_access_list)