Wallet top-up 💵

Introduction

The following guide is about how to top-up a wallet using the CosmPy library. To top-up a wallet, you need to create different wallets. In particular, if you are performing multiple transactions from a certain task_wallet, you can set an algorithm to keep that wallet address topped-up. For this use case, we will use three different wallets: wallet, authz_wallet, and task_wallet. wallet will be the main wallet address that we don't want to give full access to, therefore we will authorize authz_wallet to send a certain amount of tokens from wallet to task_wallet every time task_wallet balance falls below a certain minimum_balance threshold. This way, task_wallet can keep performing transactions using the main wallet's tokens by being topped-up by authz_wallet.

Walk-through

Aerial authorization: authorization address and authorization wallet

1. Let's start by creating a Python script for this: touch aerial_authz.py

2. First of all, we need to import the necessary classes:

   import argparse
   from datetime import datetime, timedelta
    
   from google.protobuf import any_pb2, timestamp_pb2
    
   from cosmpy.aerial.client import LedgerClient, NetworkConfig
   from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
   from cosmpy.aerial.faucet import FaucetApi
   from cosmpy.aerial.tx import Transaction
   from cosmpy.aerial.wallet import LocalWallet
   from cosmpy.protos.cosmos.authz.v1beta1.authz_pb2 import Grant
   from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgGrant
   from cosmpy.protos.cosmos.bank.v1beta1.authz_pb2 import SendAuthorization
   from cosmpy.protos.cosmos.base.v1beta1.coin_pb2 import Coin

3. We then proceed and define a _parse_commandline() function:

   def _parse_commandline():
       parser = argparse.ArgumentParser()
       parser.add_argument(
           "authz_address",
           help="address that will be granted authorization to send tokens from wallet",
       )
       parser.add_argument(
           "total_authz_time",
           type=int,
           nargs="?",
           default=10,
           help="authorization time for authz_address in minutes",
       )
       parser.add_argument(
           "spend_limit",
           type=int,
           nargs="?",
           default=1000000000000000000,
           help="maximum tokens that authz_wallet will be able to spend from wallet",
       )
    
       return parser.parse_args()

   The _parse_commandline() function is using the argparse module to define a command-line interface for this script. It expects and processes three command-line arguments:

   1. authz_address: this is a required argument. It expects the user to provide an address that will be granted authorization to send tokens from a wallet.

   2. total_authz_time: this is an optional argument. If provided, it should be an integer representing the authorization time for the authz_address in minutes. If not provided, it defaults to 10 minutes.

   3. spend_limit: this is another optional argument. If provided, it should be an integer representing the maximum tokens that the authz_wallet will be able to spend from the wallet. If not provided, it defaults to 1000000000000000000.

   The help parameter provides a description of what each argument is for, which can be helpful for users who might not be familiar with the script. After defining these arguments, the function uses parser.parse_args() to process the command-line arguments provided by the user and return them as an object containing the values provided for authz_address, total_authz_time, and spend_limit.

4. We can the define our main() function:

   def main():
       """Run main."""
       args = _parse_commandline()
    
       wallet = LocalWallet.generate()
    
       authz_address = args.authz_address
    
       ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
       faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
    
       total_authz_time = args.total_authz_time
       wallet_balance = ledger.query_bank_balance(wallet.address())
    
       amount = args.spend_limit
    
       while wallet_balance < (amount):
           print("Providing wealth to wallet...")
           faucet_api.get_wealth(wallet.address())
           wallet_balance = ledger.query_bank_balance(wallet.address())
    
       spend_amount = Coin(amount=str(amount), denom="atestfet")
    
       # Authorize authz_wallet to send tokens from wallet
       authz_any = any_pb2.Any()
       authz_any.Pack(
           SendAuthorization(spend_limit=[spend_amount]),
           "",
       )
    
       expiry = timestamp_pb2.Timestamp()
       expiry.FromDatetime(datetime.now() + timedelta(seconds=total_authz_time * 60))
       grant = Grant(authorization=authz_any, expiration=expiry)
    
       msg = MsgGrant(
           granter=str(wallet.address()),
           grantee=authz_address,
           grant=grant,
       )
    
       tx = Transaction()
       tx.add_message(msg)
    
       tx = prepare_and_broadcast_basic_transaction(ledger, tx, wallet)
       tx.wait_to_complete()
    
    
   if __name__ == "__main__":
       main()

   In the first line we define a variable args using _parse_commandline() function defined previously to retrieve the command-line arguments authz_address, total_authz_time, and spend_limit. The values are stored in the args variable. We then generate a new wallet using the generate() method of the LocalWallet class, and then set the authz_address variable to retrieve the authz_address from the command-line arguments previously defined. This is the address that will be granted authorization to send tokens from the wallet. We then initialize a ledger object using the LedgerClient class and configure it to connect to the Fetch.ai stable testnet. We also initialize a faucet_api object using the FaucetApi class to provide access to a faucet API on the Fetch.ai stable testnet.

   total_authz_time retrieves the total authorization time (in minutes) from the command-line arguments. We proceed by checking the balance of the wallet by querying the ledger, using the query_bank_balance() method. We can then define a loop that continues until the wallet balance is greater than the specified spend amount (amount). Within the loop, it requests additional tokens from the faucet using faucet_api.get_wealth() and updates the wallet balance.

   Below, we define the spend_amount variable using a Coin object representing the spend amount. In this case, it's specified in the "atestfet" denomination. We then constructs an authorization object (authz_any) using SendAuthorization. It sets an expiration time for the authorization, and creates an instance of MsgGrant message type, specifying the granter (the wallet's address), grantee (the authz_address), and the grant (the authorization object). A new transaction (tx) is finally created, and msg is added to it. The transaction is then prepared and broadcasted using prepare_and_broadcast_basic_transaction(). Finally, the script waits for the transaction to complete.

5. Save the script.

The overall script should be as follows.

import argparse
from datetime import datetime, timedelta
 
from google.protobuf import any_pb2, timestamp_pb2
 
from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
from cosmpy.aerial.faucet import FaucetApi
from cosmpy.aerial.tx import Transaction
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.protos.cosmos.authz.v1beta1.authz_pb2 import Grant
from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgGrant
from cosmpy.protos.cosmos.bank.v1beta1.authz_pb2 import SendAuthorization
from cosmpy.protos.cosmos.base.v1beta1.coin_pb2 import Coin
 
 
def _parse_commandline():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "authz_address",
        help="address that will be granted authorization to send tokens from wallet",
    )
    parser.add_argument(
        "total_authz_time",
        type=int,
        nargs="?",
        default=10,
        help="authorization time for authz_address in minutes",
    )
    parser.add_argument(
        "spend_limit",
        type=int,
        nargs="?",
        default=1000000000000000000,
        help="maximum tokens that authz_wallet will be able to spend from wallet",
    )
 
    return parser.parse_args()
 
 
def main():
    """Run main."""
    args = _parse_commandline()
 
    wallet = LocalWallet.generate()
 
    authz_address = args.authz_address
 
    ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
    faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
 
    total_authz_time = args.total_authz_time
    wallet_balance = ledger.query_bank_balance(wallet.address())
 
    amount = args.spend_limit
 
    while wallet_balance < (amount):
        print("Providing wealth to wallet...")
        faucet_api.get_wealth(wallet.address())
        wallet_balance = ledger.query_bank_balance(wallet.address())
 
    spend_amount = Coin(amount=str(amount), denom="atestfet")
 
    # Authorize authz_wallet to send tokens from wallet
    authz_any = any_pb2.Any()
    authz_any.Pack(
        SendAuthorization(spend_limit=[spend_amount]),
        "",
    )
 
    expiry = timestamp_pb2.Timestamp()
    expiry.FromDatetime(datetime.now() + timedelta(seconds=total_authz_time * 60))
    grant = Grant(authorization=authz_any, expiration=expiry)
 
    msg = MsgGrant(
        granter=str(wallet.address()),
        grantee=authz_address,
        grant=grant,
    )
 
    tx = Transaction()
    tx.add_message(msg)
 
    tx = prepare_and_broadcast_basic_transaction(ledger, tx, wallet)
    tx.wait_to_complete()
 
 
if __name__ == "__main__":
    main()

Aerial top-up

We are now ready to write a Python script which automates the process of topping-up the designated wallet (task_wallet) from the main wallet (wallet) after authorization from authz_wallet, whenever the balance of task_wallet falls below a certain threshold (minimum_balance).

1. Let's create a Python script for this and name it: touch aerial_topup.py

2. Let's then import the needed modules such as argparse for command-line argument parsing and modules from the cosmpy library for blockchain interaction:

   import argparse
   import time
   from google.protobuf import any_pb2
   from cosmpy.aerial.client import LedgerClient, NetworkConfig
   from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
   from cosmpy.aerial.faucet import FaucetApi
   from cosmpy.aerial.tx import Transaction
   from cosmpy.aerial.wallet import LocalWallet
   from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgExec
   from cosmpy.protos.cosmos.bank.v1beta1.tx_pb2 import MsgSend
   from cosmpy.protos.cosmos.base.v1beta1.coin_pb2 import Coin

3. We then define a _parse_commandline() function that sets up command-line arguments:

   def _parse_commandline():
       parser = argparse.ArgumentParser()
       parser.add_argument("wallet_address", help="main wallet address")
       parser.add_argument(
           "task_wallet_address", help="wallet address that will perform transactions"
       )
       parser.add_argument(
           "top_up_amount",
           type=int,
           nargs="?",
           default=10000000000000000,
           help="top-up amount from wallet address to task_wallet address",
       )
       parser.add_argument(
           "minimum_balance",
           type=int,
           nargs="?",
           default=1000000000000000,
           help="minimum task_wallet address balance that will trigger top-up",
       )
       parser.add_argument(
           "interval_time",
           type=int,
           nargs="?",
           default=5,
           help="interval time in seconds to query task_wallet's balance",
       )
    
       return parser.parse_args()

   Above we defined different arguments including the addresses of the main wallet (wallet_address) and the task wallet (task_wallet_address), the top-up amount from wallet_address to task_wallet_address (top_up_amount), the minimum balance for task_wallet_address (minimum_balance), and the interval time in seconds to query task_wallet_address's balance (interval_time).

   After these arguments are defined, the function uses parser.parse_args() to process the command-line arguments provided by the user. The values are then returned as an object, where each attribute corresponds to the name of the argument. This allows the script to access and utilize these values during execution.

4. We are now ready to define our main() function:

   def main():
       """Run main."""
       ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
       args = _parse_commandline()
    
       wallet_address = args.wallet_address
    
       task_wallet_address = args.task_wallet_address
    
       # Use aerial_authz.py to authorize authz_wallet address to send tokens from wallet
       authz_wallet = LocalWallet.generate()
       faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
    
       wallet_balance = ledger.query_bank_balance(authz_wallet.address())
    
       while wallet_balance < (10**18):
           print("Providing wealth to wallet...")
           faucet_api.get_wealth(authz_wallet.address())
           wallet_balance = ledger.query_bank_balance(authz_wallet.address())
    
       ledger = LedgerClient(NetworkConfig.latest_stable_testnet())
    
       # Top-up amount
       amount = args.top_up_amount
       top_up_amount = Coin(amount=str(amount), denom="atestfet")
    
       # Minimum balance for task_wallet
       minimum_balance = args.minimum_balance
    
       # Interval to query task_wallet's balance
       interval_time = args.interval_time
    
       while True:
    
           wallet_balance = ledger.query_bank_balance(wallet_address)
    
           if wallet_balance < amount:
               print("Wallet doesn't have enough balance to top-up task_wallet")
               break
    
           task_wallet_balance = ledger.query_bank_balance(task_wallet_address)
    
           if task_wallet_balance < minimum_balance:
    
               print("topping up task wallet")
               # Top-up task_wallet
               msg = any_pb2.Any()
               msg.Pack(
                   MsgSend(
                       from_address=wallet_address,
                       to_address=task_wallet_address,
                       amount=[top_up_amount],
                   ),
                   "",
               )
    
               tx = Transaction()
               tx.add_message(MsgExec(grantee=str(authz_wallet.address()), msgs=[msg]))
    
               tx = prepare_and_broadcast_basic_transaction(ledger, tx, authz_wallet)
               tx.wait_to_complete()
    
           time.sleep(interval_time)
    
    
   if __name__ == "__main__":
       main()

   Here we defined the main() function which orchestrates all of the operations. It first initializes a ledger object to interact with the blockchain using the LedgerClient() class. It then parses command-line arguments using _parse_commandline() and stores them in the args variable. The function then retrieves wallet addresses for wallet and task_wallet from args. The function then uses aerial_authz.py script previously created above to authorize authz_wallet address to send tokens from wallet. If the balance of authz_wallet is below 10**18, it uses a faucet API to provide wealth to the wallet until it reaches this threshold. Within the script, we then re-initialize the ledger object with the latest stable testnet configuration. We then proceed to set the top-up amount, the minimum balance, and interval timer thresholds from args. The script then enters an infinite loop (while True) in which it queries the balance of the main wallet. Checks if the main wallet has enough balance to top-up task_wallet. Queries the balance of task_wallet: if its balance falls below the specified minimum, it initiates a top-up by first creating a message to send tokens from wallet_address to task_wallet_address, then constructing a transaction (tx) with the authorization and message. It then prepares, broadcasts the transaction, and waits for a specified interval before repeating the process.

5. Save the script.

The overall script should be as follows:

import argparse
import time
 
from google.protobuf import any_pb2
 
from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
from cosmpy.aerial.faucet import FaucetApi
from cosmpy.aerial.tx import Transaction
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgExec
from cosmpy.protos.cosmos.bank.v1beta1.tx_pb2 import MsgSend
from cosmpy.protos.cosmos.base.v1beta1.coin_pb2 import Coin
 
def _parse_commandline():
    parser = argparse.ArgumentParser()
    parser.add_argument("wallet_address", help="main wallet address")
    parser.add_argument(
        "task_wallet_address", help="wallet address that will perform transactions"
    )
    parser.add_argument(
        "top_up_amount",
        type=int,
        nargs="?",
        default=10000000000000000,
        help="top-up amount from wallet address to task_wallet address",
    )
    parser.add_argument(
        "minimum_balance",
        type=int,
        nargs="?",
        default=1000000000000000,
        help="minimum task_wallet address balance that will trigger top-up",
    )
    parser.add_argument(
        "interval_time",
        type=int,
        nargs="?",
        default=5,
        help="interval time in seconds to query task_wallet's balance",
    )
 
    return parser.parse_args()
 
def main():
    """Run main."""
    ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
    args = _parse_commandline()
 
    wallet_address = args.wallet_address
 
    task_wallet_address = args.task_wallet_address
 
    # Use aerial_authz.py to authorize authz_wallet address to send tokens from wallet
    authz_wallet = LocalWallet.generate()
    faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
 
    wallet_balance = ledger.query_bank_balance(authz_wallet.address())
 
    while wallet_balance < (10**18):
        print("Providing wealth to wallet...")
        faucet_api.get_wealth(authz_wallet.address())
        wallet_balance = ledger.query_bank_balance(authz_wallet.address())
 
    ledger = LedgerClient(NetworkConfig.latest_stable_testnet())
 
    # Top-up amount
    amount = args.top_up_amount
    top_up_amount = Coin(amount=str(amount), denom="atestfet")
 
    # Minimum balance for task_wallet
    minimum_balance = args.minimum_balance
 
    # Interval to query task_wallet's balance
    interval_time = args.interval_time
 
    while True:
 
        wallet_balance = ledger.query_bank_balance(wallet_address)
 
        if wallet_balance < amount:
            print("Wallet doesn't have enough balance to top-up task_wallet")
            break
 
        task_wallet_balance = ledger.query_bank_balance(task_wallet_address)
 
        if task_wallet_balance < minimum_balance:
 
            print("topping up task wallet")
            # Top-up task_wallet
            msg = any_pb2.Any()
            msg.Pack(
                MsgSend(
                    from_address=wallet_address,
                    to_address=task_wallet_address,
                    amount=[top_up_amount],
                ),
                "",
            )
 
            tx = Transaction()
            tx.add_message(MsgExec(grantee=str(authz_wallet.address()), msgs=[msg]))
 
            tx = prepare_and_broadcast_basic_transaction(ledger, tx, authz_wallet)
            tx.wait_to_complete()
 
        time.sleep(interval_time)
 
if __name__ == "__main__":
    main()