kopia lustrzana https://github.com/bugout-dev/moonstream
Add general interface.
Andrey
rodzic
59f8f22b2e
commit
d9c80e566c
|
|
@ -1,487 +0,0 @@
|
|||
# Code generated by moonworm : https://github.com/bugout-dev/moonworm
|
||||
# Moonworm version : 0.2.4
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
|
||||
from brownie import Contract, network, project
|
||||
from brownie.network.contract import ContractContainer
|
||||
from eth_typing.evm import ChecksumAddress
|
||||
|
||||
|
||||
PROJECT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
|
||||
print("PROJECT_DIRECTORY", PROJECT_DIRECTORY)
|
||||
BUILD_DIRECTORY = os.path.join(PROJECT_DIRECTORY, "build", "contracts")
|
||||
|
||||
|
||||
def boolean_argument_type(raw_value: str) -> bool:
|
||||
TRUE_VALUES = ["1", "t", "y", "true", "yes"]
|
||||
FALSE_VALUES = ["0", "f", "n", "false", "no"]
|
||||
|
||||
if raw_value.lower() in TRUE_VALUES:
|
||||
return True
|
||||
elif raw_value.lower() in FALSE_VALUES:
|
||||
return False
|
||||
|
||||
raise ValueError(
|
||||
f"Invalid boolean argument: {raw_value}. Value must be one of: {','.join(TRUE_VALUES + FALSE_VALUES)}"
|
||||
)
|
||||
|
||||
|
||||
def bytes_argument_type(raw_value: str) -> str:
|
||||
return raw_value
|
||||
|
||||
|
||||
def get_abi_json(abi_name: str) -> List[Dict[str, Any]]:
|
||||
abi_full_path = os.path.join(BUILD_DIRECTORY, f"{abi_name}.json")
|
||||
if not os.path.isfile(abi_full_path):
|
||||
raise IOError(
|
||||
f"File does not exist: {abi_full_path}. Maybe you have to compile the smart contracts?"
|
||||
)
|
||||
|
||||
with open(abi_full_path, "r") as ifp:
|
||||
build = json.load(ifp)
|
||||
|
||||
abi_json = build.get("abi")
|
||||
if abi_json is None:
|
||||
raise ValueError(f"Could not find ABI definition in: {abi_full_path}")
|
||||
|
||||
return abi_json
|
||||
|
||||
|
||||
def contract_from_build(abi_name: str) -> ContractContainer:
|
||||
# This is workaround because brownie currently doesn't support loading the same project multiple
|
||||
# times. This causes problems when using multiple contracts from the same project in the same
|
||||
# python project.
|
||||
PROJECT = project.main.Project("moonworm", Path(PROJECT_DIRECTORY))
|
||||
|
||||
abi_full_path = os.path.join(BUILD_DIRECTORY, f"{abi_name}.json")
|
||||
if not os.path.isfile(abi_full_path):
|
||||
raise IOError(
|
||||
f"File does not exist: {abi_full_path}. Maybe you have to compile the smart contracts?"
|
||||
)
|
||||
|
||||
with open(abi_full_path, "r") as ifp:
|
||||
build = json.load(ifp)
|
||||
|
||||
return ContractContainer(PROJECT, build)
|
||||
|
||||
|
||||
class Multicall2:
|
||||
def __init__(self, contract_address: Optional[ChecksumAddress]):
|
||||
print("contract_address", contract_address)
|
||||
self.contract_name = "Multicall2"
|
||||
self.address = contract_address
|
||||
self.contract = None
|
||||
self.abi = get_abi_json("Multicall2")
|
||||
if self.address is not None:
|
||||
print(f"Using contract at {self.address}")
|
||||
print(f"Contract ABI: {self.abi}")
|
||||
self.contract: Optional[Contract] = Contract.from_abi(
|
||||
self.contract_name, self.address, self.abi
|
||||
)
|
||||
|
||||
def deploy(self, transaction_config):
|
||||
contract_class = contract_from_build(self.contract_name)
|
||||
deployed_contract = contract_class.deploy(transaction_config)
|
||||
self.address = deployed_contract.address
|
||||
self.contract = deployed_contract
|
||||
return deployed_contract.tx
|
||||
|
||||
def assert_contract_is_instantiated(self) -> None:
|
||||
if self.contract is None:
|
||||
raise Exception("contract has not been instantiated")
|
||||
|
||||
def verify_contract(self):
|
||||
self.assert_contract_is_instantiated()
|
||||
contract_class = contract_from_build(self.contract_name)
|
||||
contract_class.publish_source(self.contract)
|
||||
|
||||
def aggregate(self, calls: List, transaction_config) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.aggregate(calls, transaction_config)
|
||||
|
||||
def block_and_aggregate(self, calls: List, transaction_config) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.blockAndAggregate(calls, transaction_config)
|
||||
|
||||
def get_block_hash(
|
||||
self, block_number_arg: int, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getBlockHash.call(
|
||||
block_number_arg, block_identifier=block_number
|
||||
)
|
||||
|
||||
def get_block_number(
|
||||
self, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getBlockNumber.call(block_identifier=block_number)
|
||||
|
||||
def get_current_block_coinbase(
|
||||
self, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getCurrentBlockCoinbase.call(block_identifier=block_number)
|
||||
|
||||
def get_current_block_difficulty(
|
||||
self, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getCurrentBlockDifficulty.call(
|
||||
block_identifier=block_number
|
||||
)
|
||||
|
||||
def get_current_block_gas_limit(
|
||||
self, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getCurrentBlockGasLimit.call(block_identifier=block_number)
|
||||
|
||||
def get_current_block_timestamp(
|
||||
self, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getCurrentBlockTimestamp.call(
|
||||
block_identifier=block_number
|
||||
)
|
||||
|
||||
def get_eth_balance(
|
||||
self, addr: ChecksumAddress, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getEthBalance.call(addr, block_identifier=block_number)
|
||||
|
||||
def get_last_block_hash(
|
||||
self, block_number: Optional[Union[str, int]] = "latest"
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.getLastBlockHash.call(block_identifier=block_number)
|
||||
|
||||
def try_aggregate(
|
||||
self, require_success: bool, calls: List, transaction_config
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.tryAggregate(require_success, calls, transaction_config)
|
||||
|
||||
def try_block_and_aggregate(
|
||||
self, require_success: bool, calls: List, transaction_config
|
||||
) -> Any:
|
||||
self.assert_contract_is_instantiated()
|
||||
return self.contract.tryBlockAndAggregate(
|
||||
require_success, calls, transaction_config
|
||||
)
|
||||
|
||||
|
||||
def get_transaction_config(args: argparse.Namespace) -> Dict[str, Any]:
|
||||
signer = network.accounts.load(args.sender, args.password)
|
||||
transaction_config: Dict[str, Any] = {"from": signer}
|
||||
if args.gas_price is not None:
|
||||
transaction_config["gas_price"] = args.gas_price
|
||||
if args.max_fee_per_gas is not None:
|
||||
transaction_config["max_fee"] = args.max_fee_per_gas
|
||||
if args.max_priority_fee_per_gas is not None:
|
||||
transaction_config["priority_fee"] = args.max_priority_fee_per_gas
|
||||
if args.confirmations is not None:
|
||||
transaction_config["required_confs"] = args.confirmations
|
||||
if args.nonce is not None:
|
||||
transaction_config["nonce"] = args.nonce
|
||||
return transaction_config
|
||||
|
||||
|
||||
def add_default_arguments(parser: argparse.ArgumentParser, transact: bool) -> None:
|
||||
parser.add_argument(
|
||||
"--network", required=True, help="Name of brownie network to connect to"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--address", required=False, help="Address of deployed contract to connect to"
|
||||
)
|
||||
if not transact:
|
||||
parser.add_argument(
|
||||
"--block-number",
|
||||
required=False,
|
||||
type=int,
|
||||
help="Call at the given block number, defaults to latest",
|
||||
)
|
||||
return
|
||||
parser.add_argument(
|
||||
"--sender", required=True, help="Path to keystore file for transaction sender"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--password",
|
||||
required=False,
|
||||
help="Password to keystore file (if you do not provide it, you will be prompted for it)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--gas-price", default=None, help="Gas price at which to submit transaction"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-fee-per-gas",
|
||||
default=None,
|
||||
help="Max fee per gas for EIP1559 transactions",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-priority-fee-per-gas",
|
||||
default=None,
|
||||
help="Max priority fee per gas for EIP1559 transactions",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--confirmations",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Number of confirmations to await before considering a transaction completed",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--nonce", type=int, default=None, help="Nonce for the transaction (optional)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--value", default=None, help="Value of the transaction in wei(optional)"
|
||||
)
|
||||
parser.add_argument("--verbose", action="store_true", help="Print verbose output")
|
||||
|
||||
|
||||
def handle_deploy(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
transaction_config = get_transaction_config(args)
|
||||
contract = Multicall2(None)
|
||||
result = contract.deploy(transaction_config=transaction_config)
|
||||
print(result)
|
||||
if args.verbose:
|
||||
print(result.info())
|
||||
|
||||
|
||||
def handle_verify_contract(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.verify_contract()
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_aggregate(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
transaction_config = get_transaction_config(args)
|
||||
result = contract.aggregate(calls=args.calls, transaction_config=transaction_config)
|
||||
print(result)
|
||||
if args.verbose:
|
||||
print(result.info())
|
||||
|
||||
|
||||
def handle_block_and_aggregate(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
transaction_config = get_transaction_config(args)
|
||||
result = contract.block_and_aggregate(
|
||||
calls=args.calls, transaction_config=transaction_config
|
||||
)
|
||||
print(result)
|
||||
if args.verbose:
|
||||
print(result.info())
|
||||
|
||||
|
||||
def handle_get_block_hash(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_block_hash(
|
||||
block_number_arg=args.block_number_arg, block_number=args.block_number
|
||||
)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_block_number(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_block_number(block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_current_block_coinbase(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_current_block_coinbase(block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_current_block_difficulty(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_current_block_difficulty(block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_current_block_gas_limit(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_current_block_gas_limit(block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_current_block_timestamp(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_current_block_timestamp(block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_eth_balance(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_eth_balance(addr=args.addr, block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_get_last_block_hash(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
result = contract.get_last_block_hash(block_number=args.block_number)
|
||||
print(result)
|
||||
|
||||
|
||||
def handle_try_aggregate(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
transaction_config = get_transaction_config(args)
|
||||
result = contract.try_aggregate(
|
||||
require_success=args.require_success,
|
||||
calls=args.calls,
|
||||
transaction_config=transaction_config,
|
||||
)
|
||||
print(result)
|
||||
if args.verbose:
|
||||
print(result.info())
|
||||
|
||||
|
||||
def handle_try_block_and_aggregate(args: argparse.Namespace) -> None:
|
||||
network.connect(args.network)
|
||||
contract = Multicall2(args.address)
|
||||
transaction_config = get_transaction_config(args)
|
||||
result = contract.try_block_and_aggregate(
|
||||
require_success=args.require_success,
|
||||
calls=args.calls,
|
||||
transaction_config=transaction_config,
|
||||
)
|
||||
print(result)
|
||||
if args.verbose:
|
||||
print(result.info())
|
||||
|
||||
|
||||
def generate_cli() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="CLI for Multicall2")
|
||||
parser.set_defaults(func=lambda _: parser.print_help())
|
||||
subcommands = parser.add_subparsers()
|
||||
|
||||
deploy_parser = subcommands.add_parser("deploy")
|
||||
add_default_arguments(deploy_parser, True)
|
||||
deploy_parser.set_defaults(func=handle_deploy)
|
||||
|
||||
verify_contract_parser = subcommands.add_parser("verify-contract")
|
||||
add_default_arguments(verify_contract_parser, False)
|
||||
verify_contract_parser.set_defaults(func=handle_verify_contract)
|
||||
|
||||
aggregate_parser = subcommands.add_parser("aggregate")
|
||||
add_default_arguments(aggregate_parser, True)
|
||||
aggregate_parser.add_argument(
|
||||
"--calls", required=True, help="Type: tuple[]", nargs="+"
|
||||
)
|
||||
aggregate_parser.set_defaults(func=handle_aggregate)
|
||||
|
||||
block_and_aggregate_parser = subcommands.add_parser("block-and-aggregate")
|
||||
add_default_arguments(block_and_aggregate_parser, True)
|
||||
block_and_aggregate_parser.add_argument(
|
||||
"--calls", required=True, help="Type: tuple[]", nargs="+"
|
||||
)
|
||||
block_and_aggregate_parser.set_defaults(func=handle_block_and_aggregate)
|
||||
|
||||
get_block_hash_parser = subcommands.add_parser("get-block-hash")
|
||||
add_default_arguments(get_block_hash_parser, False)
|
||||
get_block_hash_parser.add_argument(
|
||||
"--block-number", required=True, help="Type: uint256", type=int
|
||||
)
|
||||
get_block_hash_parser.set_defaults(func=handle_get_block_hash)
|
||||
|
||||
get_block_number_parser = subcommands.add_parser("get-block-number")
|
||||
add_default_arguments(get_block_number_parser, False)
|
||||
get_block_number_parser.set_defaults(func=handle_get_block_number)
|
||||
|
||||
get_current_block_coinbase_parser = subcommands.add_parser(
|
||||
"get-current-block-coinbase"
|
||||
)
|
||||
add_default_arguments(get_current_block_coinbase_parser, False)
|
||||
get_current_block_coinbase_parser.set_defaults(
|
||||
func=handle_get_current_block_coinbase
|
||||
)
|
||||
|
||||
get_current_block_difficulty_parser = subcommands.add_parser(
|
||||
"get-current-block-difficulty"
|
||||
)
|
||||
add_default_arguments(get_current_block_difficulty_parser, False)
|
||||
get_current_block_difficulty_parser.set_defaults(
|
||||
func=handle_get_current_block_difficulty
|
||||
)
|
||||
|
||||
get_current_block_gas_limit_parser = subcommands.add_parser(
|
||||
"get-current-block-gas-limit"
|
||||
)
|
||||
add_default_arguments(get_current_block_gas_limit_parser, False)
|
||||
get_current_block_gas_limit_parser.set_defaults(
|
||||
func=handle_get_current_block_gas_limit
|
||||
)
|
||||
|
||||
get_current_block_timestamp_parser = subcommands.add_parser(
|
||||
"get-current-block-timestamp"
|
||||
)
|
||||
add_default_arguments(get_current_block_timestamp_parser, False)
|
||||
get_current_block_timestamp_parser.set_defaults(
|
||||
func=handle_get_current_block_timestamp
|
||||
)
|
||||
|
||||
get_eth_balance_parser = subcommands.add_parser("get-eth-balance")
|
||||
add_default_arguments(get_eth_balance_parser, False)
|
||||
get_eth_balance_parser.add_argument("--addr", required=True, help="Type: address")
|
||||
get_eth_balance_parser.set_defaults(func=handle_get_eth_balance)
|
||||
|
||||
get_last_block_hash_parser = subcommands.add_parser("get-last-block-hash")
|
||||
add_default_arguments(get_last_block_hash_parser, False)
|
||||
get_last_block_hash_parser.set_defaults(func=handle_get_last_block_hash)
|
||||
|
||||
try_aggregate_parser = subcommands.add_parser("try-aggregate")
|
||||
add_default_arguments(try_aggregate_parser, True)
|
||||
try_aggregate_parser.add_argument(
|
||||
"--require-success",
|
||||
required=True,
|
||||
help="Type: bool",
|
||||
type=boolean_argument_type,
|
||||
)
|
||||
try_aggregate_parser.add_argument(
|
||||
"--calls", required=True, help="Type: tuple[]", nargs="+"
|
||||
)
|
||||
try_aggregate_parser.set_defaults(func=handle_try_aggregate)
|
||||
|
||||
try_block_and_aggregate_parser = subcommands.add_parser("try-block-and-aggregate")
|
||||
add_default_arguments(try_block_and_aggregate_parser, True)
|
||||
try_block_and_aggregate_parser.add_argument(
|
||||
"--require-success",
|
||||
required=True,
|
||||
help="Type: bool",
|
||||
type=boolean_argument_type,
|
||||
)
|
||||
try_block_and_aggregate_parser.add_argument(
|
||||
"--calls", required=True, help="Type: tuple[]", nargs="+"
|
||||
)
|
||||
try_block_and_aggregate_parser.set_defaults(func=handle_try_block_and_aggregate)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = generate_cli()
|
||||
args = parser.parse_args()
|
||||
args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1 @@
|
|||
[{"inputs": [{"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "aggregate", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}, {"internalType": "bytes[]", "name": "returnData", "type": "bytes[]"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "blockAndAggregate", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}, {"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}, {"components": [{"internalType": "bool", "name": "success", "type": "bool"}, {"internalType": "bytes", "name": "returnData", "type": "bytes"}], "internalType": "struct Multicall2.Result[]", "name": "returnData", "type": "tuple[]"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}], "name": "getBlockHash", "outputs": [{"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getBlockNumber", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockCoinbase", "outputs": [{"internalType": "address", "name": "coinbase", "type": "address"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockDifficulty", "outputs": [{"internalType": "uint256", "name": "difficulty", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockGasLimit", "outputs": [{"internalType": "uint256", "name": "gaslimit", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockTimestamp", "outputs": [{"internalType": "uint256", "name": "timestamp", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"internalType": "address", "name": "addr", "type": "address"}], "name": "getEthBalance", "outputs": [{"internalType": "uint256", "name": "balance", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getLastBlockHash", "outputs": [{"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"internalType": "bool", "name": "requireSuccess", "type": "bool"}, {"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "tryAggregate", "outputs": [{"components": [{"internalType": "bool", "name": "success", "type": "bool"}, {"internalType": "bytes", "name": "returnData", "type": "bytes"}], "internalType": "struct Multicall2.Result[]", "name": "returnData", "type": "tuple[]"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"internalType": "bool", "name": "requireSuccess", "type": "bool"}, {"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "tryBlockAndAggregate", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}, {"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}, {"components": [{"internalType": "bool", "name": "success", "type": "bool"}, {"internalType": "bytes", "name": "returnData", "type": "bytes"}], "internalType": "struct Multicall2.Result[]", "name": "returnData", "type": "tuple[]"}], "stateMutability": "nonpayable", "type": "function"}]
|
||||
|
|
@ -0,0 +1,82 @@
|
|||
# Code generated by moonworm : https://github.com/bugout-dev/moonworm
|
||||
# Moonworm version : 0.2.4
|
||||
import json
|
||||
import os
|
||||
from typing import Any, Dict, Union
|
||||
|
||||
from eth_typing.evm import Address, ChecksumAddress
|
||||
from web3 import Web3
|
||||
from web3.contract import ContractFunction
|
||||
|
||||
from .web3_util import *
|
||||
|
||||
abi_path = os.path.join(os.path.dirname(__file__), "Multicall2_abi.json")
|
||||
with open(abi_path, "r") as abi_file:
|
||||
CONTRACT_ABI = json.load(abi_file)
|
||||
|
||||
|
||||
class Contract:
|
||||
def __init__(self, web3: Web3, contract_address: ChecksumAddress):
|
||||
self.web3 = web3
|
||||
self.address = contract_address
|
||||
self.contract = web3.eth.contract(address=self.address, abi=CONTRACT_ABI)
|
||||
|
||||
@staticmethod
|
||||
def constructor() -> ContractConstructor:
|
||||
return ContractConstructor()
|
||||
|
||||
def aggregate(self, calls: List) -> ContractFunction:
|
||||
return self.contract.functions.aggregate(calls)
|
||||
|
||||
def blockAndAggregate(self, calls: List) -> ContractFunction:
|
||||
return self.contract.functions.blockAndAggregate(calls)
|
||||
|
||||
def getBlockHash(self, blockNumber: int) -> ContractFunction:
|
||||
return self.contract.functions.getBlockHash(blockNumber)
|
||||
|
||||
def getBlockNumber(self) -> ContractFunction:
|
||||
return self.contract.functions.getBlockNumber()
|
||||
|
||||
def getCurrentBlockCoinbase(self) -> ContractFunction:
|
||||
return self.contract.functions.getCurrentBlockCoinbase()
|
||||
|
||||
def getCurrentBlockDifficulty(self) -> ContractFunction:
|
||||
return self.contract.functions.getCurrentBlockDifficulty()
|
||||
|
||||
def getCurrentBlockGasLimit(self) -> ContractFunction:
|
||||
return self.contract.functions.getCurrentBlockGasLimit()
|
||||
|
||||
def getCurrentBlockTimestamp(self) -> ContractFunction:
|
||||
return self.contract.functions.getCurrentBlockTimestamp()
|
||||
|
||||
def getEthBalance(self, addr: ChecksumAddress) -> ContractFunction:
|
||||
return self.contract.functions.getEthBalance(addr)
|
||||
|
||||
def getLastBlockHash(self) -> ContractFunction:
|
||||
return self.contract.functions.getLastBlockHash()
|
||||
|
||||
def tryAggregate(self, requireSuccess: bool, calls: List) -> ContractFunction:
|
||||
return self.contract.functions.tryAggregate(requireSuccess, calls)
|
||||
|
||||
def tryBlockAndAggregate(
|
||||
self, requireSuccess: bool, calls: List
|
||||
) -> ContractFunction:
|
||||
return self.contract.functions.tryBlockAndAggregate(requireSuccess, calls)
|
||||
|
||||
|
||||
def deploy(
|
||||
web3: Web3,
|
||||
contract_constructor: ContractFunction,
|
||||
contract_bytecode: str,
|
||||
deployer_address: ChecksumAddress,
|
||||
deployer_private_key: str,
|
||||
) -> Contract:
|
||||
tx_hash, contract_address = deploy_contract_from_constructor_function(
|
||||
web3,
|
||||
constructor=contract_constructor,
|
||||
contract_bytecode=contract_bytecode,
|
||||
contract_abi=CONTRACT_ABI,
|
||||
deployer=deployer_address,
|
||||
deployer_private_key=deployer_private_key,
|
||||
)
|
||||
return Contract(web3, contract_address)
|
||||
Ładowanie…
Reference in New Issue