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
-
Let's start by creating a Python script for this:
touch aerial_authz.py
-
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
-
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 theargparse
module to define a command-line interface for this script. It expects and processes three command-line arguments:-
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. -
total_authz_time
: this is an optional argument. If provided, it should be an integer representing the authorization time for theauthz_address
in minutes. If not provided, it defaults to10
minutes. -
spend_limit
: this is another optional argument. If provided, it should be an integer representing the maximum tokens that theauthz_wallet
will be able to spend from the wallet. If not provided, it defaults to1000000000000000000
.
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 usesparser.parse_args()
to process the command-line arguments provided by the user and return them as an object containing the values provided forauthz_address
,total_authz_time
, andspend_limit
. -
-
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 argumentsauthz_address
,total_authz_time
, andspend_limit
. The values are stored in theargs
variable. We then generate a new wallet using thegenerate()
method of theLocalWallet
class, and then set theauthz_address
variable to retrieve theauthz_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 aledger
object using theLedgerClient
class and configure it to connect to the Fetch.ai stable testnet. We also initialize afaucet_api
object using theFaucetApi
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 thequery_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 usingfaucet_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
) usingSendAuthorization
. It sets an expiration time for the authorization, and creates an instance ofMsgGrant
message type, specifying thegranter
(the wallet's address),grantee
(theauthz_address
), and thegrant
(the authorization object). A new transaction (tx
) is finally created, andmsg
is added to it. The transaction is then prepared and broadcasted usingprepare_and_broadcast_basic_transaction()
. Finally, the script waits for the transaction to complete. -
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
).
-
Let's create a Python script for this and name it:
touch aerial_topup.py
-
Let's then import the needed modules such as
argparse
for command-line argument parsing and modules from thecosmpy
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
-
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 fromwallet_address
totask_wallet_address
(top_up_amount
), the minimum balance fortask_wallet_address
(minimum_balance
), and the interval time in seconds to querytask_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. -
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 aledger
object to interact with the blockchain using theLedgerClient()
class. It then parses command-line arguments using_parse_commandline()
and stores them in theargs
variable. The function then retrieves wallet addresses forwallet
andtask_wallet
fromargs
. The function then usesaerial_authz.py
script previously created above to authorizeauthz_wallet
address to send tokens fromwallet
. If the balance ofauthz_wallet
is below10**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 fromargs
. The script then enters an infinite loop (while True
) in which it queries the balance of the mainwallet
. Checks if the main wallet has enough balance to top-uptask_wallet
. Queries the balance oftask_wallet
: if its balance falls below the specified minimum, it initiates a top-up by first creating a message to send tokens fromwallet_address
totask_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. -
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()