refactoring

Author: popenta

multiversx_sdk_cli/base_transactions_controller.py

@@ -1,14 +1,32 @@
-from typing import Optional, Union
+import logging
+from typing import Any, Optional, Union
 
-from multiversx_sdk import LedgerAccount, Transaction, TransactionComputer
+from multiversx_sdk import Address, LedgerAccount, Transaction, TransactionComputer
+from multiversx_sdk.abi import (
+    AddressValue,
+    BigUIntValue,
+    BoolValue,
+    BytesValue,
+    StringValue,
+)
 
+from multiversx_sdk_cli.config import get_address_hrp
 from multiversx_sdk_cli.constants import (
+    ADDRESS_PREFIX,
     EXTRA_GAS_LIMIT_FOR_GUARDED_TRANSACTIONS,
     EXTRA_GAS_LIMIT_FOR_RELAYED_TRANSACTIONS,
+    FALSE_STR_LOWER,
+    HEX_PREFIX,
+    MAINCHAIN_ADDRESS_HRP,
+    STR_PREFIX,
+    TRUE_STR_LOWER,
 )
 from multiversx_sdk_cli.cosign_transaction import cosign_transaction
+from multiversx_sdk_cli.errors import BadUserInput
 from multiversx_sdk_cli.interfaces import IAccount
 
+logger = logging.getLogger("base_controller")
+
 
 class BaseTransactionsController:
     def __init__(self) -> None:
@@ -84,3 +102,45 @@ def _sign_guarded_transaction_if_guardian(
     def _sign_relayed_transaction_if_relayer(self, transaction: Transaction, relayer: Union[IAccount, None]):
         if relayer and transaction.relayer:
             transaction.relayer_signature = relayer.sign_transaction(transaction)
+
+    def _prepare_args_for_factory(self, arguments: list[str]) -> list[Any]:
+        args: list[Any] = []
+
+        for arg in arguments:
+            if arg.startswith(HEX_PREFIX):
+                args.append(BytesValue(self._hex_to_bytes(arg)))
+            elif arg.isnumeric():
+                args.append(BigUIntValue(int(arg)))
+            elif arg.startswith(ADDRESS_PREFIX):
+                args.append(AddressValue.new_from_address(Address.new_from_bech32(arg[len(ADDRESS_PREFIX) :])))
+            elif arg.startswith(MAINCHAIN_ADDRESS_HRP):
+                # this flow will be removed in the future
+                logger.warning(
+                    "Address argument has no prefix. This flow will be removed in the future. Please provide each address using the `addr:` prefix. (e.g. --arguments addr:erd1...)"
+                )
+                args.append(AddressValue.new_from_address(Address.new_from_bech32(arg)))
+            elif arg.startswith(get_address_hrp()):
+                args.append(AddressValue.new_from_address(Address.new_from_bech32(arg)))
+            elif arg.lower() == FALSE_STR_LOWER:
+                args.append(BoolValue(False))
+            elif arg.lower() == TRUE_STR_LOWER:
+                args.append(BoolValue(True))
+            elif arg.startswith(STR_PREFIX):
+                args.append(StringValue(arg[len(STR_PREFIX) :]))
+            else:
+                raise BadUserInput(
+                    f"Unknown argument type for argument: `{arg}`. Use `mxpy contract <sub-command> --help` to check all supported arguments"
+                )
+
+        return args
+
+    def _hex_to_bytes(self, arg: str):
+        argument = arg[len(HEX_PREFIX) :]
+        argument = argument.upper()
+        argument = self.ensure_even_length(argument)
+        return bytes.fromhex(argument)
+
+    def ensure_even_length(self, string: str) -> str:
+        if len(string) % 2 == 1:
+            return "0" + string
+        return string

multiversx_sdk_cli/cli_contracts.py

@@ -412,6 +412,10 @@ def call(args: Any):
     arguments, should_prepare_args = _get_contract_arguments(args)
     contract_address = Address.new_from_bech32(args.contract)
 
+    token_transfers = None
+    if args.token_transfers:
+        token_transfers = cli_shared.prepare_token_transfers(args.token_transfers)
+
     tx = contract.prepare_execute_transaction(
         caller=sender,
         contract=contract_address,
@@ -420,7 +424,7 @@ def call(args: Any):
         should_prepare_args=should_prepare_args,
         gas_limit=int(args.gas_limit),
         value=int(args.value),
-        transfers=args.token_transfers,
+        token_transfers=token_transfers,
         nonce=sender.nonce,
         version=int(args.version),
         options=int(args.options),

multiversx_sdk_cli/cli_multisig.py

@@ -22,9 +22,6 @@
     SendAsyncCall,
     SendTransferExecuteEgld,
     SendTransferExecuteEsdt,
-    Token,
-    TokenComputer,
-    TokenTransfer,
     Transaction,
     TransactionsFactoryConfig,
 )
@@ -1189,7 +1186,7 @@ def deposit(args: Any):
 
     token_transfers = args.token_transfers or None
     if token_transfers:
-        token_transfers = _prepare_token_transfers(token_transfers)
+        token_transfers = cli_shared.prepare_token_transfers(token_transfers)
 
     tx = multisig.prepare_deposit_transaction(
         owner=sender,
@@ -1444,7 +1441,7 @@ def transfer_and_execute(args: Any):
         gas_price=int(args.gas_price),
         version=int(args.version),
         options=int(args.options),
-        should_prepare_args=should_prepare_args,
+        should_prepare_args_for_factory=should_prepare_args,
         guardian_and_relayer_data=guardian_and_relayer_data,
         opt_gas_limit=opt_gas_limit,
         function=function,
@@ -1485,7 +1482,7 @@ def transfer_and_execute_esdt(args: Any):
     function = args.function if args.function else None
     contract_abi = Abi.load(Path(args.contract_abi)) if args.contract_abi else None
     arguments, should_prepare_args = _get_contract_arguments(args)
-    token_transfers = _prepare_token_transfers(args.token_transfers)
+    token_transfers = cli_shared.prepare_token_transfers(args.token_transfers)
 
     tx = multisig.prepare_transfer_execute_esdt_transaction(
         owner=sender,
@@ -1496,7 +1493,7 @@ def transfer_and_execute_esdt(args: Any):
         gas_price=int(args.gas_price),
         version=int(args.version),
         options=int(args.options),
-        should_prepare_args=should_prepare_args,
+        should_prepare_args_for_factory=should_prepare_args,
         guardian_and_relayer_data=guardian_and_relayer_data,
         token_transfers=token_transfers,
         opt_gas_limit=opt_gas_limit,
@@ -1534,7 +1531,7 @@ def async_call(args: Any):
 
     token_transfers = args.token_transfers or None
     if token_transfers:
-        token_transfers = _prepare_token_transfers(args.token_transfers)
+        token_transfers = cli_shared.prepare_token_transfers(args.token_transfers)
 
     tx = multisig.prepare_async_call_transaction(
         owner=sender,
@@ -1545,7 +1542,7 @@ def async_call(args: Any):
         gas_price=int(args.gas_price),
         version=int(args.version),
         options=int(args.options),
-        should_prepare_args=should_prepare_args,
+        should_prepare_args_for_factory=should_prepare_args,
         guardian_and_relayer_data=guardian_and_relayer_data,
         native_token_amount=int(args.value),
         token_transfers=token_transfers,
@@ -1593,7 +1590,7 @@ def deploy_from_source(args: Any):
         readable=args.metadata_readable,
         payable=args.metadata_payable,
         payable_by_sc=args.metadata_payable_by_sc,
-        should_prepare_args=should_prepare_args,
+        should_prepare_args_for_factory=should_prepare_args,
         guardian_and_relayer_data=guardian_and_relayer_data,
         native_token_amount=int(args.value),
         abi=contract_abi,
@@ -1640,7 +1637,7 @@ def upgrade_from_source(args: Any):
         readable=args.metadata_readable,
         payable=args.metadata_payable,
         payable_by_sc=args.metadata_payable_by_sc,
-        should_prepare_args=should_prepare_args,
+        should_prepare_args_for_factory=should_prepare_args,
         guardian_and_relayer_data=guardian_and_relayer_data,
         native_token_amount=int(args.value),
         abi=contract_abi,
@@ -2247,22 +2244,6 @@ def _get_contract_arguments(args: Any) -> tuple[list[Any], bool]:
         return args.arguments, True
 
 
-def _prepare_token_transfers(transfers: list[str]) -> list[TokenTransfer]:
-    token_computer = TokenComputer()
-    token_transfers: list[TokenTransfer] = []
-
-    for i in range(0, len(transfers) - 1, 2):
-        identifier = transfers[i]
-        amount = int(transfers[i + 1])
-        nonce = token_computer.extract_nonce_from_extended_identifier(identifier)
-
-        token = Token(identifier, nonce)
-        transfer = TokenTransfer(token, amount)
-        token_transfers.append(transfer)
-
-    return token_transfers
-
-
 def _send_or_simulate(tx: Transaction, contract_address: Address, args: Any):
     output_builder = cli_shared.send_or_simulate(tx, args, dump_output=False)
     output_builder.set_contract_address(contract_address)

multiversx_sdk_cli/cli_shared.py

@@ -11,6 +11,9 @@
     ApiNetworkProvider,
     LedgerAccount,
     ProxyNetworkProvider,
+    Token,
+    TokenComputer,
+    TokenTransfer,
     Transaction,
 )
 
@@ -617,3 +620,20 @@ def prepare_guardian_relayer_data(args: Any) -> GuardianRelayerData:
         relayer=relayer,
         relayer_address=relayer_address,
     )
+
+
+def prepare_token_transfers(transfers: list[str]) -> list[TokenTransfer]:
+    """Converts a list of token transfers as received from the CLI to a list of TokenTransfer objects."""
+    token_computer = TokenComputer()
+    token_transfers: list[TokenTransfer] = []
+
+    for i in range(0, len(transfers) - 1, 2):
+        identifier = transfers[i]
+        amount = int(transfers[i + 1])
+        nonce = token_computer.extract_nonce_from_extended_identifier(identifier)
+
+        token = Token(identifier, nonce)
+        transfer = TokenTransfer(token, amount)
+        token_transfers.append(transfer)
+
+    return token_transfers

multiversx_sdk_cli/cli_transactions.py

@@ -2,14 +2,7 @@
 from pathlib import Path
 from typing import Any
 
-from multiversx_sdk import (
-    Address,
-    ProxyNetworkProvider,
-    Token,
-    TokenComputer,
-    TokenTransfer,
-    TransactionComputer,
-)
+from multiversx_sdk import Address, ProxyNetworkProvider, TransactionComputer
 
 from multiversx_sdk_cli import cli_shared, utils
 from multiversx_sdk_cli.args_validation import (
@@ -133,7 +126,7 @@ def create_transaction(args: Any):
     gas_limit = int(args.gas_limit) if args.gas_limit else 0
 
     transfers = getattr(args, "token_transfers", None)
-    transfers = prepare_token_transfers(transfers) if transfers else None
+    transfers = cli_shared.prepare_token_transfers(transfers) if transfers else None
 
     chain_id = cli_shared.get_chain_id(args.chain, args.proxy)
     tx_controller = TransactionsController(chain_id)
@@ -155,22 +148,6 @@ def create_transaction(args: Any):
     cli_shared.send_or_simulate(tx, args)
 
 
-def prepare_token_transfers(transfers: list[Any]) -> list[TokenTransfer]:
-    token_computer = TokenComputer()
-    token_transfers: list[TokenTransfer] = []
-
-    for i in range(0, len(transfers) - 1, 2):
-        identifier = transfers[i]
-        amount = int(transfers[i + 1])
-        nonce = token_computer.extract_nonce_from_extended_identifier(identifier)
-
-        token = Token(identifier, nonce)
-        transfer = TokenTransfer(token, amount)
-        token_transfers.append(transfer)
-
-    return token_transfers
-
-
 def send_transaction(args: Any):
     validate_proxy_argument(args)
 

multiversx_sdk_cli/constants.py

@@ -19,3 +19,10 @@
 TCS_SERVICE_ID = "MultiversXTCSService"
 EXTRA_GAS_LIMIT_FOR_GUARDED_TRANSACTIONS = 50_000
 EXTRA_GAS_LIMIT_FOR_RELAYED_TRANSACTIONS = 50_000
+
+HEX_PREFIX = "0x"
+FALSE_STR_LOWER = "false"
+TRUE_STR_LOWER = "true"
+STR_PREFIX = "str:"
+ADDRESS_PREFIX = "addr:"
+MAINCHAIN_ADDRESS_HRP = "erd"