Box Storage in Algorand Python
Introduction
Box storage is akin to a key-value database that smart contracts can manipulate on-chain.
An application can create any number of boxes, with each box storing up to 32k bytes of data.
Algorand Python has recently introduced three new abstractions for working with box storage: Box, BoxMap, and BoxRef.
In this article we'll leverage BoxMap
to manage users and assets for a PokΓ©mon-inspired game.
Let's start with the basic imports and an outline for the contract class:
from algopy import Account, ARC4Contract, BoxMap, Bytes, Global, Txn, UInt64, arc4, gtxn, op
class Game(ARC4Contract):
def __init__(self) -> None:
...
BoxMap Basics
BoxMap
allows us to group a set of boxes with common key and value types.
Let's define the information we want to store about each user and create a mapping from their account to their profile box:
class User(arc4.Struct):
registered_at: arc4.UInt64
name: arc4.String
balance: arc4.UInt64
class Game(ARC4Contract):
def __init__(self) -> None:
self.user = BoxMap(Account, User)
The first argument to BoxMap
is the key type (Account
) and the second is the value type (User
).
BoxMap(key_type=Account, value_type=User)
.There is also an optional third argument to BoxMap
called key_prefix
.
The prefix is used as a way to logically separate one set of boxes from another.
If no argument is passed, the key prefix defaults to the variable name (here it's 'user').
User Registration
Next we need a method to handle user registration:
class Game(ARC4Contract):
def __init__(self) -> None:
self.user = BoxMap(Account, User)
@arc4.abimethod
def register(self, name: arc4.String) -> User:
"""Registers a user and returns their profile information.
Args:
name (arc4.String): The user's name.
Returns:
User: The user's profile information.
"""
if Txn.sender not in self.user:
self.user[Txn.sender] = User(
registered_at=arc4.UInt64(Global.latest_timestamp), name=name, balance=arc4.UInt64(0)
)
return self.user[Txn.sender]
The register
method begins by checking whether a user box already exists for the transaction sender's account.
If it doesn't exist, a new box is created.
This condition is important to avoid overwriting existing information for the user (such as their balance) if they have already previously registered.
Finally, the user's profile is read from box storage and returned.
To test this:
def test_register(algod_client: AlgodClient, app_client: GameClient, account: Account) -> None:
"""Tests the `register` method."""
box_abi = abi.ABIType.from_string("(uint64,string,uint64)")
box_name = b"user" + decode_address(account.address)
# Test application call return value
user = app_client.register(
name="Alice",
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
).return_value
assert isinstance(user.registered_at, int) and user.registered_at > 0
assert user.name == "Alice"
assert isinstance(user.balance, int) and user.balance == 0
# Test box value fetched from Algod
box_value = b64decode(algod_client.application_box_by_name(app_client.app_id, box_name)["value"])
registered_at, name, balance = box_abi.decode(box_value)
assert isinstance(registered_at, int) and registered_at == user.registered_at
assert name == "Alice"
assert isinstance(balance, int) and balance == 0
We fist construct an ABI type using the algosdk
library:
abi.ABIType.from_string("(uint64,string,uint64)")
The types here line up to the User
struct in the smart contract:
class User(arc4.Struct):
registered_at: arc4.UInt64
name: arc4.String
balance: arc4.UInt64
Then we construct the box name to pass to the boxes
array in the application call, making sure to add the key prefix from the BoxMap
:
box_name = b"user" + decode_address(account.address)
Then we can do some basic tests on the response:
assert isinstance(user.registered_at, int) and user.registered_at > 0
assert user.name == "Alice"
assert isinstance(user.balance, int) and user.balance == 0
I chose to test both the User
returned from the application call, and the value of the application box (fetched from Algod).
For completeness, it would be good to also check that the register method is idempotent. I'll leave that up to you.
Funding Accounts
Users need to be able to fund their accounts to purchase in-game assets.
Let's add a new method to enable this:
@arc4.abimethod
def fund_account(self, payment: gtxn.PaymentTransaction) -> arc4.UInt64:
"""Funds a user's account.
Args:
payment (gtxn.PaymentTransaction): The payment transaction.
Returns:
arc4.UInt64: The user's updated balance.
"""
assert (
payment.receiver == Global.current_application_address
), "Payment receiver must be the application address"
assert payment.sender in self.user, "User must be registered"
self.user[payment.sender].balance = arc4.UInt64(self.user[payment.sender].balance.native + payment.amount)
return self.user[payment.sender].balance
The payment is made as part of a group transaction.
The two assert
statements check that the payment receiver is the application address, and that the sender of the payment is a registered user.
If those checks pass, the balance in the user's profile box is updated.
User
is an arc4.Struct
, so each attribute has to be an ARC-4 type.
To operate on the balance
numerically, we can access its representation as a UInt64
through the native
property.
The end result is converted back to an arc4.UInt64
and updated in box storage.
To test this:
def test_fund_account(algod_client: AlgodClient, app_client: GameClient) -> None:
"""Tests the `fund_account` method."""
# Generate new account
account = get_account(app_client.algod_client, "test")
app_client.signer = AccountTransactionSigner(account.private_key)
box_abi = abi.ABIType.from_string("(uint64,string,uint64)")
box_name = b"user" + decode_address(account.address)
# Register a new user
user = app_client.register(
name="Bob",
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
).return_value
# Store balance before funding
balance_before = user.balance
# Construct payment transaction
ptxn = PaymentTxn(
sender=account.address,
sp=algod_client.suggested_params(),
receiver=app_client.app_address,
amt=10_000,
)
# Fund the user's account
balance_returned = app_client.fund_account(
payment=TransactionWithSigner(ptxn, AccountTransactionSigner(account.private_key)),
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
).return_value
# Test the value returned from the app call
assert balance_before + 10_000 == balance_returned
# Parse user's box from Algod
box_value = b64decode(algod_client.application_box_by_name(app_client.app_id, box_name)["value"])
_, _, box_balance = box_abi.decode(box_value)
# Test box value balance
assert balance_before + 10_000 == box_balance
I have again chosen to test the balance returned from the application call and the balance stored in the box separately, as it's possible the two could diverge if my code is wrong :)
Creating Game Assets
Let's define a new type for in-game assets:
class GameAsset(arc4.Struct):
name: arc4.String
description: arc4.String
price: arc4.UInt64
A box key can only be up to 64 bytes long, so we'll use the hash of the asset name as the key:
from typing import TypeAlias
Hash: TypeAlias = Bytes
class Game(ARC4Contract):
def __init__(self) -> None:
self.user = BoxMap(Account, User)
self.asset = BoxMap(Hash, GameAsset)
TypeAlias
is optional, but it can be a good way to improve readability and type hints.Now we can define a method to insert a new game asset or update an existing record in box storage:
@arc4.abimethod
def admin_upsert_asset(self, asset: GameAsset) -> None:
"""Updates or inserts a game asset.
Args:
asset (GameAsset): The game asset information.
"""
assert Txn.sender == Global.creator_address, "Only the creator can call this method"
self.asset[op.sha256(asset.name.bytes)] = asset.copy()
The only 'gotcha' here is needing to call asset.copy()
, because it's a reference to a mutable data type.
To test the admin_upsert_asset
method:
def test_admin_upsert_asset(algod_client: AlgodClient, app_client: GameClient, account: Account) -> None:
"""Tests the `admin_upsert_asset` method."""
# Switch back to creator account
app_client.signer = AccountTransactionSigner(account.private_key)
for asset in (
("POKEBALL", "Catches Pokemon", 200),
("POTION", "Restores 20 HP", 300),
("BICYCLE", "Allows you to travel faster", 1_000_000),
):
name, _, _ = asset
box_name = b"asset" + sha256(abi.StringType().encode(name)).digest()
# Call app client
app_client.admin_upsert_asset(
asset=asset,
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
)
# Test box value fetched from Algod
box_value = b64decode(algod_client.application_box_by_name(app_client.app_id, box_name)["value"])
box_abi = abi.ABIType.from_string("(string,string,uint64)")
# Test box value balance
assert asset == tuple(box_abi.decode(box_value))
I should test that both inserting and updating works, and that only the creator account can call the method.
But I'm lazy.
Owning Assets
Now we have an interesting decision to make about data modelling.
What's the best way to represent the one-to-many relationship between users and assets?
We could either store all user-assets in a single box (the user profile), or create one box per user-asset.
The advantage of the first option is that some of the application calls will only need to reference a single box in the boxes
array, and the smart contract can easily iterate over the user's assets:
UserAsset: TypeAlias = arc4.Tuple[arc4.DynamicBytes, arc4.UInt64] # (asset_id, quantity)
class User(arc4.Struct):
registered_at: arc4.UInt64
name: arc4.String
balance: arc4.UInt64
assets: arc4.DynamicArray[UserAsset]
The major downside is that each box has fixed storage capacity, limiting the number of distinct assets a user can own.
Updating and deleting elements is also O(n), which is costly in a smart contract.
The alternative is to create one box per user-asset.
This will scale to support any number of assets per user, and updating and deleting elements is O(1).
I think it's the better option for this game.
Let's create one last BoxMap
to represent distinct user-assets:
Hash: TypeAlias = Bytes
Quantity: TypeAlias = UInt64
class Game(ARC4Contract):
def __init__(self) -> None:
self.user = BoxMap(Account, User)
self.asset = BoxMap(Hash, GameAsset)
self.user_asset = BoxMap(Hash, Quantity)
The key will be hash(<account> + hash(<asset name>))
, because it saves computing the second hash in the contract.
The value is the quantity of the asset that the user owns.
To buy an asset:
@arc4.abimethod
def buy_asset(self, asset_id: Hash, quantity: Quantity) -> None:
"""Buys a game asset.
Args:
asset_id (Hash): The hash of the asset name.
quantity (Quantity): The quantity to purchase.
"""
assert Txn.sender in self.user, "User must be registered"
assert asset_id in self.asset, "Invalid asset ID"
user_balance = self.user[Txn.sender].balance.native
asset_price = self.asset[asset_id].price.native
assert user_balance >= (total := asset_price * quantity), "Insufficient funds"
# Update user balance
self.user[Txn.sender].balance = arc4.UInt64(user_balance - total)
# Insert or update user-asset box
user_asset_id = op.sha256(Txn.sender.bytes + asset_id)
if user_asset_id in self.user_asset:
self.user_asset[user_asset_id] += quantity
else:
self.user_asset[user_asset_id] = quantity
We first check that the user's in-game balance is sufficient for them to make the purchase.
:=
.Then we update the user's balance in the profile box.
If the user already has a box for this asset, we add the purchased quantity to the existing amount.
Otherwise, we create a new user-asset box and set the purchased quantity as the box value.
To test this method:
def test_buy_asset(algod_client: AlgodClient, app_client: GameClient, account: Account) -> None:
"""Tests the `buy_asset` method."""
box = lambda name: b64decode(algod_client.application_box_by_name(app_client.app_id, name)["value"])
# Generate new account
account = get_account(app_client.algod_client, "test_buyer")
app_client.signer = AccountTransactionSigner(account.private_key)
# Register new user
user_box_name = b"user" + decode_address(account.address)
app_client.register(
name="Ash",
transaction_parameters=TransactionParameters(boxes=[(0, user_box_name)]),
)
# Get asset price from box storage
asset_name = "POKEBALL"
asset_box_name = b"asset" + (asset_id := sha256(abi.StringType().encode(asset_name)).digest())
_, _, asset_price = abi.ABIType.from_string("(string,string,uint64)").decode(box(asset_box_name))
# Construct payment transaction to fund the user's game account
ptxn = PaymentTxn(
sender=account.address,
sp=algod_client.suggested_params(),
receiver=app_client.app_address,
amt=asset_price * 2,
)
# Fund the user's account
app_client.fund_account(
payment=TransactionWithSigner(ptxn, AccountTransactionSigner(account.private_key)),
transaction_parameters=TransactionParameters(boxes=[(0, b"user" + decode_address(account.address))]),
)
# Get user balance before buying asset
user_box_abi = abi.ABIType.from_string("(uint64,string,uint64)")
_, _, balance_before = user_box_abi.decode(box(user_box_name))
user_asset_box_name = b"user_asset" + sha256(decode_address(account.address) + asset_id).digest()
# Get user-asset quantity before buying
try:
quantity_before = abi.UintType(64).decode(box(user_asset_box_name))
except AlgodHTTPError:
quantity_before = 0
buy = lambda: app_client.buy_asset(
asset_id=asset_id,
quantity=1,
transaction_parameters=TransactionParameters(
boxes=[
(0, asset_box_name),
(0, user_box_name),
(0, user_asset_box_name),
]
),
)
# Buy one unit of asset
buy()
# Buy another unit of asset
buy()
# Get user balance after buying two units of the asset
_, _, balance_after = user_box_abi.decode(box(user_box_name))
# Test user balance in profile box
assert balance_before - (asset_price * 2) == balance_after
# Test user-asset box value
quantity_after = abi.UintType(64).decode(box(user_asset_box_name))
assert quantity_after - 2 == quantity_before
I've chosen to create and register a new account, funding it with enough money to buy two units of the POKEBALL asset.
The test buys one unit first, then a second unit in a subsequent transaction.
The point of doing these separately is to check that both insertion and updating works in this section of the buy_asset
method:
if user_asset_id in self.user_asset:
self.user_asset[user_asset_id] = self.user_asset[user_asset_id] + quantity
else:
self.user_asset[user_asset_id] = quantity
The test then checks that the user's in-game balance has been updated correctly in their profile box.
Finally, the user-asset box is tested to ensure that the new value is 2 greater than the starting quantity.
Ideas for Further Development
If you're looking for ideas on how to improve or extend the contract with new features, here are a few that could be useful:
Handling box storage costs (each player could pay for their own storage)
Allowing users to sell assets back to the application
Enabling users to trade/exchange assets with one another
Creating an ASA to represent in-game currency
Asset tokenisation (fungible or non-fungible)
The Complete Contract Code
from typing import TypeAlias
from algopy import Account, ARC4Contract, BoxMap, Bytes, Global, Txn, UInt64, arc4, gtxn, op
Hash: TypeAlias = Bytes
Quantity: TypeAlias = UInt64
class User(arc4.Struct):
registered_at: arc4.UInt64
name: arc4.String
balance: arc4.UInt64
class GameAsset(arc4.Struct):
name: arc4.String
description: arc4.String
price: arc4.UInt64
class Game(ARC4Contract):
def __init__(self) -> None:
self.user = BoxMap(Account, User)
self.asset = BoxMap(Hash, GameAsset)
self.user_asset = BoxMap(Hash, Quantity)
@arc4.abimethod
def register(self, name: arc4.String) -> User:
"""Registers a user and returns their profile information.
Args:
name (arc4.String): The user's name.
Returns:
User: The user's profile information.
"""
if Txn.sender not in self.user:
self.user[Txn.sender] = User(
registered_at=arc4.UInt64(Global.latest_timestamp), name=name, balance=arc4.UInt64(0)
)
return self.user[Txn.sender]
@arc4.abimethod
def fund_account(self, payment: gtxn.PaymentTransaction) -> arc4.UInt64:
"""Funds a user's account.
Args:
payment (gtxn.PaymentTransaction): The payment transaction.
Returns:
arc4.UInt64: The user's updated balance.
"""
assert (
payment.receiver == Global.current_application_address
), "Payment receiver must be the application address"
assert payment.sender in self.user, "User must be registered"
self.user[payment.sender].balance = arc4.UInt64(self.user[payment.sender].balance.native + payment.amount)
return self.user[payment.sender].balance
@arc4.abimethod
def buy_asset(self, asset_id: Hash, quantity: Quantity) -> None:
"""Buys a game asset.
Args:
asset_id (Hash): The hash of the asset name.
quantity (Quantity): The quantity to purchase.
"""
assert Txn.sender in self.user, "User must be registered"
assert asset_id in self.asset, "Invalid asset ID"
user_balance = self.user[Txn.sender].balance.native
asset_price = self.asset[asset_id].price.native
assert user_balance >= (total := asset_price * quantity), "Insufficient funds"
# Update user balance
self.user[Txn.sender].balance = arc4.UInt64(user_balance - total)
# Insert or update user-asset box
user_asset_id = op.sha256(Txn.sender.bytes + asset_id)
if user_asset_id in self.user_asset:
self.user_asset[user_asset_id] += quantity
else:
self.user_asset[user_asset_id] = quantity
@arc4.abimethod
def admin_upsert_asset(self, asset: GameAsset) -> None:
"""Updates or inserts a game asset.
Args:
asset (GameAsset): The game asset information.
"""
assert Txn.sender == Global.creator_address, "Only the creator can call this method"
self.asset[op.sha256(asset.name.bytes)] = asset.copy()
Tests
These tests are not exhaustive. Please only use them as a starting point for your own tests, or to see how off-chain code and application calls can be constructed.
from base64 import b64decode
from hashlib import sha256
import algokit_utils
import pytest
from algokit_utils import (
Account,
TransactionParameters,
TransferParameters,
get_account,
transfer,
)
from algokit_utils.config import config
from algosdk import abi
from algosdk.atomic_transaction_composer import (
AccountTransactionSigner,
TransactionWithSigner,
)
from algosdk.encoding import decode_address
from algosdk.error import AlgodHTTPError
from algosdk.transaction import PaymentTxn
from algosdk.v2client.algod import AlgodClient
from algosdk.v2client.indexer import IndexerClient
from smart_contracts.artifacts.box_map.client import GameClient
@pytest.fixture(scope="session")
def app_client(account: Account, algod_client: AlgodClient, indexer_client: IndexerClient) -> GameClient:
config.configure(
debug=True,
# trace_all=True,
)
client = GameClient(
algod_client,
creator=account,
indexer_client=indexer_client,
)
client.deploy(
on_schema_break=algokit_utils.OnSchemaBreak.AppendApp,
on_update=algokit_utils.OnUpdate.AppendApp,
)
transfer(
algod_client,
TransferParameters(
from_account=account,
to_address=client.app_address,
micro_algos=100_000_000,
),
)
return client
def test_register(algod_client: AlgodClient, app_client: GameClient, account: Account) -> None:
"""Tests the `register` method."""
box_abi = abi.ABIType.from_string("(uint64,string,uint64)")
box_name = b"user" + decode_address(account.address)
# Test application call return value
user = app_client.register(
name="Alice",
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
).return_value
assert isinstance(user.registered_at, int) and user.registered_at > 0
assert user.name == "Alice"
assert isinstance(user.balance, int) and user.balance == 0
# Test box value fetched from Algod
box_value = b64decode(algod_client.application_box_by_name(app_client.app_id, box_name)["value"])
registered_at, name, balance = box_abi.decode(box_value)
assert isinstance(registered_at, int) and registered_at == user.registered_at
assert name == "Alice"
assert isinstance(balance, int) and balance == 0
def test_fund_account(algod_client: AlgodClient, app_client: GameClient) -> None:
"""Tests the `fund_account` method."""
# Generate new account
account = get_account(app_client.algod_client, "test")
app_client.signer = AccountTransactionSigner(account.private_key)
box_abi = abi.ABIType.from_string("(uint64,string,uint64)")
box_name = b"user" + decode_address(account.address)
# Register a new user
user = app_client.register(
name="Bob",
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
).return_value
# Store balance before funding
balance_before = user.balance
# Construct payment transaction
ptxn = PaymentTxn(
sender=account.address,
sp=algod_client.suggested_params(),
receiver=app_client.app_address,
amt=10_000,
)
# Fund the user's account
balance_returned = app_client.fund_account(
payment=TransactionWithSigner(ptxn, AccountTransactionSigner(account.private_key)),
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
).return_value
# Test the value returned from the app call
assert balance_before + 10_000 == balance_returned
# Parse user's box from Algod
box_value = b64decode(algod_client.application_box_by_name(app_client.app_id, box_name)["value"])
_, _, box_balance = box_abi.decode(box_value)
# Test box value balance
assert balance_before + 10_000 == box_balance
def test_admin_upsert_asset(algod_client: AlgodClient, app_client: GameClient, account: Account) -> None:
"""Tests the `admin_upsert_asset` method."""
# Switch back to creator account
app_client.signer = AccountTransactionSigner(account.private_key)
for asset in (
("POKEBALL", "Catches Pokemon", 200),
("POTION", "Restores 20 HP", 300),
("BICYCLE", "Allows you to travel faster", 1_000_000),
):
name, _, _ = asset
box_name = b"asset" + sha256(abi.StringType().encode(name)).digest()
# Call app client
app_client.admin_upsert_asset(
asset=asset,
transaction_parameters=TransactionParameters(boxes=[(0, box_name)]),
)
# Test box value fetched from Algod
box_value = b64decode(algod_client.application_box_by_name(app_client.app_id, box_name)["value"])
box_abi = abi.ABIType.from_string("(string,string,uint64)")
# Test box value balance
assert asset == tuple(box_abi.decode(box_value))
def test_buy_asset(algod_client: AlgodClient, app_client: GameClient, account: Account) -> None:
"""Tests the `buy_asset` method."""
box = lambda name: b64decode(algod_client.application_box_by_name(app_client.app_id, name)["value"])
# Generate new account
account = get_account(app_client.algod_client, "test_buyer")
app_client.signer = AccountTransactionSigner(account.private_key)
# Register new user
user_box_name = b"user" + decode_address(account.address)
app_client.register(
name="Ash",
transaction_parameters=TransactionParameters(boxes=[(0, user_box_name)]),
)
# Get asset price from box storage
asset_name = "POKEBALL"
asset_box_name = b"asset" + (asset_id := sha256(abi.StringType().encode(asset_name)).digest())
_, _, asset_price = abi.ABIType.from_string("(string,string,uint64)").decode(box(asset_box_name))
# Construct payment transaction to fund the user's game account
ptxn = PaymentTxn(
sender=account.address,
sp=algod_client.suggested_params(),
receiver=app_client.app_address,
amt=asset_price * 2,
)
# Fund the user's account
app_client.fund_account(
payment=TransactionWithSigner(ptxn, AccountTransactionSigner(account.private_key)),
transaction_parameters=TransactionParameters(boxes=[(0, b"user" + decode_address(account.address))]),
)
# Get user balance before buying asset
user_box_abi = abi.ABIType.from_string("(uint64,string,uint64)")
_, _, balance_before = user_box_abi.decode(box(user_box_name))
user_asset_box_name = b"user_asset" + sha256(decode_address(account.address) + asset_id).digest()
# Get user-asset quantity before buying
try:
quantity_before = abi.UintType(64).decode(box(user_asset_box_name))
except AlgodHTTPError:
quantity_before = 0
buy = lambda: app_client.buy_asset(
asset_id=asset_id,
quantity=1,
transaction_parameters=TransactionParameters(
boxes=[
(0, asset_box_name),
(0, user_box_name),
(0, user_asset_box_name),
]
),
)
# Buy one unit of asset
buy()
# Buy another unit of asset
buy()
# Get user balance after buying two units of the asset
_, _, balance_after = user_box_abi.decode(box(user_box_name))
# Test user balance in profile box
assert balance_before - (asset_price * 2) == balance_after
# Test user-asset box value
quantity_after = abi.UintType(64).decode(box(user_asset_box_name))
assert quantity_after - 2 == quantity_before
Subscribe to my newsletter
Read articles from Alexander Codes directly inside your inbox. Subscribe to the newsletter, and don't miss out.
Written by
Alexander Codes
Alexander Codes
Data Engineer β¨οΈ | Pythonista π | Blogger π You can support my work at https://app.nf.domains/name/alexandercodes.algo