kopia lustrzana https://github.com/bugout-dev/moonworm
added __init__ py and web3 py to generated directories
yhtiyar
rodzic
68e898142c
commit
953ce2e270
|
|
@ -2,7 +2,6 @@ import argparse
|
|||
import json
|
||||
import os
|
||||
|
||||
from . import manage
|
||||
from .generator import generate_contract_cli_file, generate_contract_file
|
||||
|
||||
|
||||
|
|
@ -18,34 +17,6 @@ def handle_genereate_cli(args: argparse.Namespace) -> None:
|
|||
generate_contract_cli_file(contract_abi, args.output_path)
|
||||
|
||||
|
||||
def handle_contract_show(args: argparse.Namespace) -> None:
|
||||
with open(args.abi, "r") as ifp:
|
||||
contract_abi = json.load(ifp)
|
||||
show_all = not args.functions and not args.events
|
||||
functions, events = manage.abi_show(contract_abi)
|
||||
if show_all or args.functions:
|
||||
print("Functions:")
|
||||
for function in functions:
|
||||
print(f"function {function['name']}:")
|
||||
print("\tArgs:")
|
||||
for arg in function["inputs"]:
|
||||
print(f"\t\t{arg['name']} -> {arg['type']}")
|
||||
print("")
|
||||
print("\tReturns:")
|
||||
for out in function["outputs"]:
|
||||
print(f"\t\t{out['name']} -> {out['type']}")
|
||||
print("\n")
|
||||
|
||||
if show_all or args.events:
|
||||
print("Events:")
|
||||
for event in events:
|
||||
print(f"event {event['name']}:")
|
||||
print("\tArgs:")
|
||||
for arg in event["inputs"]:
|
||||
print(f"\t\t{arg['name']} -> {arg['type']}")
|
||||
print("")
|
||||
|
||||
|
||||
def generate_argument_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Centipede: Manage your smart contract"
|
||||
|
|
@ -54,28 +25,6 @@ def generate_argument_parser() -> argparse.ArgumentParser:
|
|||
parser.set_defaults(func=lambda _: parser.print_help())
|
||||
subcommands = parser.add_subparsers()
|
||||
|
||||
contract = subcommands.add_parser("contract", description="Contract operations")
|
||||
contract.set_defaults(func=lambda _: contract.print_help())
|
||||
contract_subcommands = contract.add_subparsers()
|
||||
|
||||
def populate_contract_leaf_parsers(
|
||||
leaf_parser: argparse.ArgumentParser,
|
||||
) -> None:
|
||||
leaf_parser.add_argument(
|
||||
"-abi",
|
||||
"--abi",
|
||||
required=True,
|
||||
help=f"Path to contract abi JSON file",
|
||||
)
|
||||
|
||||
contract_show = contract_subcommands.add_parser(
|
||||
"show", description="Show contract functions and events"
|
||||
)
|
||||
populate_contract_leaf_parsers(contract_show)
|
||||
contract_show.add_argument("--functions", action="store_true")
|
||||
contract_show.add_argument("--events", action="store_true")
|
||||
contract_show.set_defaults(func=handle_contract_show)
|
||||
|
||||
generate_parser = subcommands.add_parser(
|
||||
"generate", description="Centipede code generator"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -4,14 +4,18 @@
|
|||
import argparse
|
||||
import json
|
||||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||||
import os
|
||||
|
||||
from eth_typing.evm import Address, ChecksumAddress
|
||||
import web3
|
||||
from web3 import Web3
|
||||
from web3.contract import Contract
|
||||
|
||||
with open("{abi_json}", "r") as ifp:
|
||||
CONTRACT_ABI = json.load(ifp)
|
||||
from .web3_util import *
|
||||
|
||||
abi_path = os.path.join(os.path.dirname(__file__), "abi.json")
|
||||
with open(abi_path, "r") as abi_file:
|
||||
CONTRACT_ABI = json.load(abi_file)
|
||||
|
||||
CONTRACT_FUNCTIONS = {{}}
|
||||
for abi_item in CONTRACT_ABI:
|
||||
|
|
|
|||
|
|
@ -1,12 +1,16 @@
|
|||
# Code generated by moonstream centipede : https://github.com/bugout-dev/centipede
|
||||
# Centipede version : {centipede_version}
|
||||
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
|
||||
|
||||
with open("{abi_json}", "r") as abi_file:
|
||||
from .web3_util import *
|
||||
|
||||
abi_path = os.path.join(os.path.dirname(__file__), "abi.json")
|
||||
with open(abi_path, "r") as abi_file:
|
||||
CONTRACT_ABI = json.load(abi_file)
|
||||
{contract_body}
|
||||
|
|
@ -2,6 +2,7 @@ import json
|
|||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, List, Union
|
||||
from shutil import copyfile
|
||||
|
||||
import libcst as cst
|
||||
from web3.types import ABIFunction
|
||||
|
|
@ -44,14 +45,19 @@ def make_annotation(types: list):
|
|||
)
|
||||
|
||||
|
||||
EVM_PYTHON_TYPE_MAPPINGS = {
|
||||
"uint256": make_annotation(["int"]),
|
||||
"uint8": make_annotation(["int"]),
|
||||
"uint": make_annotation(["int"]),
|
||||
"bytes4": make_annotation(["bytes"]),
|
||||
"string": make_annotation(["str"]),
|
||||
"address": make_annotation(["Address", "ChecksumAddress"]),
|
||||
}
|
||||
def python_type(evm_type: str) -> List[str]:
|
||||
if evm_type.startswith(("uint", "int")):
|
||||
return ["int"]
|
||||
elif evm_type.startswith("bytes"):
|
||||
return ["bytes"]
|
||||
elif evm_type == "string":
|
||||
return ["str"]
|
||||
elif evm_type == "address":
|
||||
return ["ChecksumAddress", "Address"]
|
||||
elif evm_type == "bool":
|
||||
return ["bool"]
|
||||
else:
|
||||
raise ValueError(f"Cannot convert to python type {evm_type}")
|
||||
|
||||
|
||||
def generate_contract_class(
|
||||
|
|
@ -108,7 +114,7 @@ def generate_contract_function(
|
|||
if param_name == "":
|
||||
param_name = f"{default_param_name}{default_counter}"
|
||||
default_counter += 1
|
||||
param_type = EVM_PYTHON_TYPE_MAPPINGS[param["type"]]
|
||||
param_type = make_annotation(python_type(param["type"]))
|
||||
param_names.append(param_name)
|
||||
func_params.append(
|
||||
cst.Param(
|
||||
|
|
@ -119,9 +125,11 @@ def generate_contract_function(
|
|||
|
||||
func_name = cst.Name(func_object["name"])
|
||||
|
||||
proxy_call_code = f"return self.contract.functions.{func_object['name']}({','.join(param_names)}).call()"
|
||||
proxy_call_code = (
|
||||
f"return self.contract.functions.{func_object['name']}({','.join(param_names)})"
|
||||
)
|
||||
func_body = cst.IndentedBlock(body=[cst.parse_statement(proxy_call_code)])
|
||||
func_returns = cst.Annotation(annotation=cst.Name(value="Any"))
|
||||
func_returns = cst.Annotation(annotation=cst.Name(value="ContractFunction"))
|
||||
|
||||
return cst.FunctionDef(
|
||||
name=func_name,
|
||||
|
|
@ -131,22 +139,38 @@ def generate_contract_function(
|
|||
)
|
||||
|
||||
|
||||
def copy_web3_util(dest_dir: str) -> None:
|
||||
dest_filepath = os.path.join(dest_dir, "web3_util.py")
|
||||
if os.path.isfile(dest_filepath):
|
||||
print(f"{dest_filepath} file already exists")
|
||||
web3_util_path = os.path.join(os.path.dirname(__file__), "web3_util.py")
|
||||
copyfile(web3_util_path, dest_filepath)
|
||||
|
||||
|
||||
def create_init_py(dest_dir: str) -> None:
|
||||
dest_filepath = os.path.join(dest_dir, "__init__.py")
|
||||
if os.path.isfile(dest_filepath):
|
||||
print(f"{dest_filepath} file already exists")
|
||||
with open(dest_filepath, "w") as ofp:
|
||||
ofp.write()
|
||||
|
||||
|
||||
def generate_contract_file(abi: Dict[str, Any], output_path: str):
|
||||
contract_body = cst.Module(body=[generate_contract_class(abi)]).code
|
||||
|
||||
JSON_FILE_PATH = os.path.join(output_path, "abi.json")
|
||||
|
||||
content = REPORTER_FILE_TEMPLATE.format(
|
||||
abi_json=JSON_FILE_PATH,
|
||||
contract_body=contract_body,
|
||||
centipede_version=CENTIPEDE_VERSION,
|
||||
)
|
||||
contract_file_path = os.path.join(output_path, "lol.py")
|
||||
contract_file_path = os.path.join(output_path, "interface.py")
|
||||
with open(contract_file_path, "w") as ofp:
|
||||
ofp.write(content)
|
||||
|
||||
JSON_FILE_PATH = os.path.join(output_path, "abi.json")
|
||||
with open(JSON_FILE_PATH, "w") as ofp:
|
||||
ofp.write(json.dumps(abi))
|
||||
copy_web3_util(output_path)
|
||||
create_init_py(output_path)
|
||||
|
||||
|
||||
def generate_argument_parser_function(abi: Dict[str, Any]) -> cst.FunctionDef:
|
||||
|
|
@ -219,12 +243,9 @@ def generate_argument_parser_function(abi: Dict[str, Any]) -> cst.FunctionDef:
|
|||
|
||||
def generate_contract_cli_file(abi: Dict[str, Any], output_path: str):
|
||||
|
||||
JSON_FILE_PATH = os.path.join(output_path, "abi.json")
|
||||
|
||||
cli_body = cst.Module(body=[generate_argument_parser_function(abi)]).code
|
||||
|
||||
content = CLI_FILE_TEMPLATE.format(
|
||||
abi_json="abi.json",
|
||||
cli_content=cli_body,
|
||||
centipede_version=CENTIPEDE_VERSION,
|
||||
)
|
||||
|
|
@ -233,5 +254,9 @@ def generate_contract_cli_file(abi: Dict[str, Any], output_path: str):
|
|||
with open(cli_file_path, "w") as ofp:
|
||||
ofp.write(content)
|
||||
|
||||
JSON_FILE_PATH = os.path.join(output_path, "abi.json")
|
||||
with open(JSON_FILE_PATH, "w") as ofp:
|
||||
ofp.write(json.dumps(abi))
|
||||
|
||||
copy_web3_util(output_path)
|
||||
create_init_py(output_path)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||||
import os
|
||||
|
||||
|
||||
|
|
@ -122,3 +122,18 @@ def read_keys_from_env() -> Tuple[ChecksumAddress, str]:
|
|||
if raw_address is None:
|
||||
raise ValueError("CENTIPEDE_ETHEREUM_ADDRESS_PRIVATE_KEY is not set")
|
||||
return (Web3.toChecksumAddress(raw_address), private_key)
|
||||
|
||||
|
||||
def cast_to_python_type(evm_type: str) -> Callable:
|
||||
if evm_type.startswith(("uint", "int")):
|
||||
return int
|
||||
elif evm_type.startswith("bytes"):
|
||||
return bytes
|
||||
elif evm_type == "string":
|
||||
return str
|
||||
elif evm_type == "address":
|
||||
return Web3.toChecksumAddress
|
||||
elif evm_type == "bool":
|
||||
return bool
|
||||
else:
|
||||
raise ValueError(f"Cannot convert to python type {evm_type}")
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue