Marketplace Contract Part 5 - Universal Listings

Video Walkthrough

Watch the tutorial video for this chapter:

Step 1: Generate Two Accounts

• One for Lister, one for Buyer

from algosdk.account import generate_account

private_key, address = generate_account()
print(private_key, address)

Step 2: Load each Account with Funds

Use the following link to load funds into each account:

https://bank.testnet.algorand.network/

Step 3: Create .env

algod_token = 
algod_server = https://testnet-api.4160.nodely.dev
private_key = lqPuVOtF4p6QpsaL7iTA6mbHoy7a5+nuAPjzw7HvckiXQWUlnZQVYSCZSvUK8w+MGjh09sNsx7WybOeb0MLD/Q==
buyer_private_key = jHx+aE6G/5RKAMJK7J7R+j3SyjKhzLsrbYIrzmmhWphJsT7CpTxqdhsjWQx4wDpx2I5B+1FiBSp16O5fXd/Sqg==
asset_1 = 724395141
asset_2 = 724449087
app_id = 724455779

Step 4: Compile Universal Listing Contract

Create and compile the asset listing contract:

algokit compile py fileName.py

Use the following implementation to compile your contract:

from algopy import ARC4Contract, String, gtxn, Asset, Txn, GlobalState, itxn, Global, BoxRef, UInt64, TransactionType
from algopy.arc4 import abimethod, Address, Struct, Bool
from algopy.arc4 import UInt64 as arc4UInt64

class listingName(Struct):
    lister: Address
    counter: arc4UInt64
    
class universalTradeListing(Struct):
    asset_listed: arc4UInt64
    asset_listed_quantity: arc4UInt64
    asset_requested: arc4UInt64
    asset_amount_requested: arc4UInt64
    fulfilled: Bool
    
class listings(ARC4Contract):
    def __init__(self) -> None:
        self.listingCounter = GlobalState(arc4UInt64(0))
        
    @abimethod
    def triggerOptIn(
        self,
        asset_to_opt_into: Asset,
        fee_payment: gtxn.PaymentTransaction
    ) -> tuple[String, UInt64]:

        assert fee_payment.amount >= 101_000
        assert fee_payment.receiver == Global.current_application_address
        
        itxn.AssetTransfer(
            xfer_asset=asset_to_opt_into,
            asset_receiver=Global.current_application_address,
            fee=Global.min_txn_fee
        ).submit()
        
        return String("Successfully opted in to asset: "), asset_to_opt_into.id
        
    @abimethod
    def triggerOptIn(
        self,
        asset_to_opt_into: Asset,
        fee_payment: gtxn.PaymentTransaction
    ) -> tuple[String, UInt64]:

        assert fee_payment.amount >= 101_000
        assert fee_payment.receiver == Global.current_application_address
        
        itxn.AssetTransfer(
            xfer_asset=asset_to_opt_into,
            asset_receiver=Global.current_application_address,
            fee=Global.min_txn_fee
        ).submit()
        
        return String("Successfully opted in to asset: "), asset_to_opt_into.id
            
    @abimethod
    def postUniversalListing(
        self,
        asset_to_list: gtxn.AssetTransferTransaction,
        asset_request: arc4UInt64,
        asset_request_quantity: arc4UInt64,
        box_fee_mbr_payment: gtxn.PaymentTransaction
    ) -> String:
        
        assert box_fee_mbr_payment.amount == 31_700
        assert asset_to_list.asset_receiver == Global.current_application_address
        
        self.listingCounter.value = arc4UInt64(self.listingCounter.value.native + 1)
        
        listing_box_name_info = listingName(Address(Txn.sender), self.listingCounter.value)
        
        listing_box = BoxRef(key=listing_box_name_info.bytes)
        
        value, exists = listing_box.maybe()
        assert not exists
        
        listing_box.create(size=33)
        
        listing_box_value_info = universalTradeListing(arc4UInt64(asset_to_list.xfer_asset.id), arc4UInt64(asset_to_list.asset_amount), asset_request, arc4UInt64(asset_request_quantity.native), Bool(False))
            
        listing_box.put(listing_box_value_info.bytes)
        
        return String("Trade Request Posted")
        
    @abimethod
    def fulfillUniversalListing(
        self,
        asset_transfer_fee: gtxn.PaymentTransaction,
        asset_listed: Asset,
        listing_name: listingName,
        
    ) -> String:
        
        assert asset_transfer_fee.amount == 1000

        listing_box = BoxRef(key=listing_name.bytes)
        
        value, exists = listing_box.maybe()
        assert exists
        
        listing_value = universalTradeListing.from_bytes(value)

        if Global.group_size == 3:
            assert gtxn.Transaction(0).type == TransactionType.Payment
            payment_fulfillment_tx = gtxn.PaymentTransaction(0)
            assert payment_fulfillment_tx.receiver == Global.current_application_address
            assert listing_value.asset_listed == asset_listed.id
            assert listing_value.asset_requested == 0
            assert payment_fulfillment_tx.amount == listing_value.asset_amount_requested.native

            itxn.AssetTransfer(
                xfer_asset=asset_listed,
                asset_receiver=Txn.sender,
                asset_amount=listing_value.asset_listed_quantity.native,
                fee=Global.min_txn_fee
            ).submit()            
            
            itxn.Payment(
                receiver=listing_name.lister.native,
                amount=listing_value.asset_amount_requested.native,
                fee=Global.min_txn_fee
            ).submit()
            
            listing_box.delete()
            
            return String("Buyer paid with Algo, Listing Fulfilled!")
            
        elif Global.group_size == 5:
            assert gtxn.Transaction(2).type == TransactionType.AssetTransfer
            asset_trade_fulfillment_tx = gtxn.AssetTransferTransaction(2)
            assert asset_trade_fulfillment_tx.asset_receiver == Global.current_application_address
            assert listing_value.asset_listed == asset_listed.id
            assert listing_value.asset_requested == asset_trade_fulfillment_tx.xfer_asset.id
            assert asset_trade_fulfillment_tx.asset_amount == listing_value.asset_amount_requested
                        
            itxn.AssetTransfer(
                xfer_asset=asset_listed,
                asset_receiver=Txn.sender,
                asset_amount=listing_value.asset_listed_quantity.native,
                fee=Global.min_txn_fee
            ).submit()
                    
        listing_value.fulfilled = Bool(True)
        listing_box.put(listing_value.bytes)
        
        return String("Trade Request Fulfilled! Lister can now claim")        
                
    @abimethod
    def claimUniversalListing(
        self,
        asset_transfer_fee: gtxn.PaymentTransaction,
        asset_requested: Asset,
        listing_name: listingName,
    ) -> String:
                
        assert listing_name.lister == Address(Txn.sender)
        assert asset_transfer_fee.amount == 1000
        
        listing_box = BoxRef(key=listing_name.bytes)
        
        value, exists = listing_box.maybe()
        assert exists
        
        listing_value = universalTradeListing.from_bytes(value)
        
        assert asset_requested.id == listing_value.asset_requested
        assert listing_value.fulfilled == Bool(True)
        
        itxn.AssetTransfer(
            xfer_asset=asset_requested.id,
            asset_receiver=Txn.sender,
            asset_amount=listing_value.asset_amount_requested.native,
            fee=Global.min_txn_fee
        ).submit()
        
        listing_box.delete()
        
        return String("Trade Request Completed!")
        
    @abimethod
    def cancelUniversalListing(
        self,
        listed_asset: Asset,
        listing_name: listingName,
        transfer_fee: gtxn.PaymentTransaction
    ) -> String:
        
        assert transfer_fee.amount == 1000
        assert listing_name.lister == Address(Txn.sender)
        
        listing_box = BoxRef(key=listing_name.bytes)
        
        value, exists = listing_box.maybe()
        
        assert exists
        
        listing_value = universalTradeListing.from_bytes(value)
        
        assert listing_value.asset_listed == listed_asset.id
        
        itxn.AssetTransfer(
            xfer_asset=listed_asset.id,
            asset_amount=listing_value.asset_listed_quantity.native,
            asset_receiver=Txn.sender,
            fee=Global.min_txn_fee
        ).submit()
        
        listing_box.delete()    
        
        return String("Trade Request Cancelled!")

Ensure you have the necessary imports and class definitions for your contract.

Step 5: Create Two Arbitrary Assets for Experimenting

Follow the steps to create and configure two assets for testing purposes.

from algosdk.v2client.algod import AlgodClient
from algosdk.transaction import AssetConfigTxn, wait_for_confirmation
from algosdk.account import address_from_private_key
import os
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

private_key = os.getenv('buyer_private_key')
address = address_from_private_key(private_key)

params = algod_client.suggested_params()

asset_creation_txn = AssetConfigTxn(
    sender=address,
    sp=params,
    total=100,
    default_frozen=False,
    asset_name='Test Asset Two',
    unit_name='TA2',
    manager=address,
    reserve=address,
    strict_empty_address_check=False,
)

signed_tx = asset_creation_txn.sign(private_key)
tx_id = algod_client.send_transaction(signed_tx)
print(tx_id)

wait_for_confirmation(algod_client, tx_id)

asset_id = algod_client.pending_transaction_info(tx_id)['asset-index']
asset_info = algod_client.asset_info(asset_id)
print(asset_info)

Step 6: Store Generated Asset IDs in .env

asset_1 = 123456
asset_2 = 654321

Step 7: Experiment with Opting the Contract into an Asset

If you're creating multiple listings, you need to opt the contract into each asset the first time.

from algokit_utils import ApplicationClient
from algosdk.v2client.algod import AlgodClient
from algosdk.account import address_from_private_key
from algosdk.atomic_transaction_composer import AccountTransactionSigner, TransactionWithSigner, AtomicTransactionComposer
from algosdk.transaction import PaymentTxn
from pathlib import Path
import os
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

private_key = os.getenv('private_key')
address = address_from_private_key(private_key)

app_spec = Path(__file__).parent / './listings.arc32.json'
app_id = int(os.getenv('app_id'))
signer = AccountTransactionSigner(private_key)
params = algod_client.suggested_params()

app_client = ApplicationClient(
    algod_client=algod_client,
    app_spec=app_spec,
    app_id=app_id,
    signer=signer,
    sender=address,
    suggested_params=params,
)

atc = AtomicTransactionComposer()

asset_1 = int(os.getenv('asset_1'))

fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=101_000)
wrapped_payment = TransactionWithSigner(fee_payment_tx, signer)

app_client.compose_call(atc, call_abi_method='triggerOptIn', asset_to_opt_into=asset_1, fee_payment=wrapped_payment)

results = atc.execute(algod_client, 2)

tx_ids = [results.tx_ids[i] for i in range(len(results.tx_ids))]
abi_results = [results.abi_results[i].return_value for i in range(len(results.abi_results))]

print(tx_ids)
print(abi_results)

Step 8: Make an Asset for Algo Listing Box (All listing types are supported from parts 1-4)

Create an asset listing box to manage your asset listings.

from algokit_utils import ApplicationClient
from algosdk.v2client.algod import AlgodClient
from algosdk.account import address_from_private_key
from algosdk.atomic_transaction_composer import AccountTransactionSigner, TransactionWithSigner, AtomicTransactionComposer
from algosdk.transaction import PaymentTxn, AssetTransferTxn
from pathlib import Path
from algosdk.abi import ABIType
import os
from base64 import b64decode
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

private_key = os.getenv('private_key')
address = address_from_private_key(private_key)

app_spec = Path(__file__).parent / './listings.arc32.json'
app_id = int(os.getenv('app_id'))
signer = AccountTransactionSigner(private_key)
params = algod_client.suggested_params()

app_client = ApplicationClient(
    algod_client=algod_client,
    app_spec=app_spec,
    app_id=app_id,
    signer=signer,
    sender=address,
    suggested_params=params,
)

atc = AtomicTransactionComposer()

asset_1 = int(os.getenv('asset_1'))

fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=101_000)
wrapped_payment = TransactionWithSigner(fee_payment_tx, signer)

app_client.compose_call(
    atc, 
    call_abi_method='triggerOptIn', 
    asset_to_opt_into=asset_1, 
    fee_payment=wrapped_payment)

asset_to_list_transaction = AssetTransferTxn(sender=address, sp=params, receiver=app_client.app_address, amt=1, index=asset_1)
wrapped_asset_list_transaction = TransactionWithSigner(asset_to_list_transaction, signer)

box_fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=31_700)
wrapped_box_fee_payment = TransactionWithSigner(box_fee_payment_tx, signer)

current_global_listing_counter = algod_client.application_info(app_id)['params']['global-state'][0]['value']['bytes']
uint64_coder = ABIType.from_string('(uint64)')
b64_decoded_global_listing_counter = b64decode(current_global_listing_counter)
encoded_global_listing_counter = uint64_coder.decode(b64_decoded_global_listing_counter)[0]

listing_box_counter = encoded_global_listing_counter + 1
listing_box_coder = ABIType.from_string('(address,uint64)')
users_listing_box_name = listing_box_coder.encode((address, listing_box_counter))

asset_2 = int(os.getenv('asset_2'))

app_client.compose_call(
    atc, 
    call_abi_method='postUniversalListing', 
    asset_to_list=wrapped_asset_list_transaction, 
    asset_request=0, 
    asset_request_quantity=2_000_000,
    box_fee_mbr_payment=wrapped_box_fee_payment,
    transaction_parameters={'boxes': [[app_id, users_listing_box_name]]}
)

results = atc.execute(algod_client, 2)

tx_ids = [results.tx_ids[i] for i in range(len(results.tx_ids))]
abi_results = [results.abi_results[i].return_value for i in range(len(results.abi_results))]

print(tx_ids)
print(abi_results)

Step 9: Check Respective Listing Information

Retrieve and verify the listing information from the contract.

import os
from algosdk.v2client.algod import AlgodClient
from algosdk.abi import ABIType
from base64 import b64decode
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

app_id = int(os.getenv('app_id'))

boxes = algod_client.application_boxes(app_id)['boxes']

for box in boxes:
    box_name_b64encoded = box['name']
    print(b64decode(box_name_b64encoded))
    box_value = b64decode(algod_client.application_box_by_name(app_id, b64decode(box_name_b64encoded))['value'])
    print(box_value)

    #Asset for Algo: '(uint64,uint64)'
    #Asset for Asset: '(uint64,uint64,bool)'
    #Asset for Asset Quantity: '(uint64,uint64,uint64,bool)
    #Asset Quantity for Asset Quantity: '(uint64,uint64,uint64,uint64,bool)'
    #Universal Listing: '(uint64,uint64,uint64,uint64,bool)'
    listing_box_value_coder = ABIType.from_string('(uint64,uint64)')
    print(listing_box_value_coder.decode(box_value))

Step 10: Purchase the Asset with Your Buyer Private Key

When generating a listing, the counter variable should be incremented.

counter = 1

Note; counter should match the number of listings generated so far.

When generating a listing, the counter variable should be incremented.

counter = 1

Note; counter should match the number of listings generated so far.

Execute the purchase of the asset using the buyer's private key.

from algokit_utils import ApplicationClient
from algosdk.v2client.algod import AlgodClient
from algosdk.account import address_from_private_key
from algosdk.atomic_transaction_composer import AccountTransactionSigner, TransactionWithSigner, AtomicTransactionComposer
from algosdk.transaction import PaymentTxn, AssetTransferTxn
from algosdk.abi import ABIType
from pathlib import Path
import os
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

private_key = os.getenv('buyer_private_key')
address = address_from_private_key(private_key)

app_spec = Path(__file__).parent / './listings.arc32.json'
app_id = int(os.getenv('app_id'))
signer = AccountTransactionSigner(private_key)
params = algod_client.suggested_params()

app_client = ApplicationClient(
    algod_client=algod_client,
    app_spec=app_spec,
    app_id=app_id,
    signer=signer,
    sender=address,
    suggested_params=params,
)

atc = AtomicTransactionComposer()

asset_1 = int(os.getenv('asset_1'))
asset_2 = int(os.getenv('asset_2'))

'''
User should opt into the asset if they are not prepared to receive it
'''
#asset_opt_in_transaction = AssetTransferTxn(sender=address, sp=params, receiver=address, amt=0, index=asset_1)
#wrapped_opt_in_transaction = TransactionWithSigner(asset_opt_in_transaction, signer)
#atc.add_transaction(wrapped_opt_in_transaction)

#5 tx for asset to asset payments

optin_fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=101_000)
wrapped_optin_payment = TransactionWithSigner(optin_fee_payment_tx, signer)

app_client.compose_call(
    atc, 
    call_abi_method='triggerOptIn', 
    asset_to_opt_into=asset_2, 
    fee_payment=wrapped_optin_payment)

asset_payment_tx = AssetTransferTxn(sender=address, sp=params, receiver=app_client.app_address, amt=3, index=asset_2)
wrapped_asset_payment = TransactionWithSigner(asset_payment_tx, signer)

asset_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=2_000_000)
wrapped_asset_payment = TransactionWithSigner(asset_payment_tx, signer)

atc.add_transaction(wrapped_asset_payment)

fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=1000)
wrapped_fee_payment = TransactionWithSigner(fee_payment_tx, signer)

lister_address = address_from_private_key(os.getenv('private_key'))
counter = 1 #Remember to increment counter for additional listings
listing_box_coder = ABIType.from_string('(address,uint64)')
box_name = listing_box_coder.encode((lister_address, counter))

app_client.compose_call(
    atc, 
    call_abi_method='fulfillUniversalListing', 
    asset_transfer_fee=wrapped_fee_payment,
    asset_listed=asset_1, 
    listing_name=(lister_address,counter), 
    transaction_parameters={'boxes': [[app_id, box_name]], 'accounts': [lister_address]})

results = atc.execute(algod_client, 2)

tx_ids = [results.tx_ids[i] for i in range(len(results.tx_ids))]
abi_results = [results.abi_results[i].return_value for i in range(len(results.abi_results))]

print(tx_ids)
print(abi_results)

Step 11: Complete Listing Request and Redeem Buyers Asset

from algokit_utils import ApplicationClient
from algosdk.v2client.algod import AlgodClient
from algosdk.account import address_from_private_key
from algosdk.atomic_transaction_composer import AccountTransactionSigner, TransactionWithSigner, AtomicTransactionComposer
from algosdk.transaction import PaymentTxn, AssetTransferTxn
from pathlib import Path
from algosdk.util import algos_to_microalgos
from algosdk.abi import ABIType
import os
from base64 import b64decode
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

private_key = os.getenv('private_key')
address = address_from_private_key(private_key)

app_spec = Path(__file__).parent / './listings.arc32.json'
app_id = int(os.getenv('app_id'))
signer = AccountTransactionSigner(private_key)
params = algod_client.suggested_params()

app_client = ApplicationClient(
    algod_client=algod_client,
    app_spec=app_spec,
    app_id=app_id,
    signer=signer,
    sender=address,
    suggested_params=params,
)

atc = AtomicTransactionComposer()

asset_1 = int(os.getenv('asset_1'))
asset_2 = int(os.getenv('asset_2'))

'''
User should opt into the asset if they are not prepared to receive it
'''
#asset_opt_in_transaction = AssetTransferTxn(sender=address, sp=params, receiver=address, amt=0, index=asset_2)
#wrapped_opt_in_transaction = TransactionWithSigner(asset_opt_in_transaction, signer)
#atc.add_transaction(wrapped_opt_in_transaction)

fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=1000)
wrapped_fee_payment = TransactionWithSigner(fee_payment_tx, signer)

lister_address = address_from_private_key(os.getenv('private_key'))
counter = 1 #Remember to increment counter for additional listings
listing_box_coder = ABIType.from_string('(address,uint64)')
box_name = listing_box_coder.encode((lister_address, counter))

app_client.compose_call(
    atc, 
    call_abi_method='claimUniversalListing', 
    asset_transfer_fee=wrapped_fee_payment,
    asset_requested=asset_2,
    listing_name=(lister_address,counter), 
    transaction_parameters={'boxes': [[app_id, box_name]]})

results = atc.execute(algod_client, 2)

tx_ids = [results.tx_ids[i] for i in range(len(results.tx_ids))]
abi_results = [results.abi_results[i].return_value for i in range(len(results.abi_results))]

print(tx_ids)
print(abi_results)

Step 12: Experiment with Cancelling a Listing Request

If you fulfilled the previous request already, you will need to generate a new listing again with Step 8, and increment the "counter" variable from 1 to 2. For example:

counter = 2

from algokit_utils import ApplicationClient
from algosdk.v2client.algod import AlgodClient
from algosdk.account import address_from_private_key
from algosdk.atomic_transaction_composer import AccountTransactionSigner, TransactionWithSigner, AtomicTransactionComposer
from algosdk.transaction import PaymentTxn
from algosdk.abi import ABIType
from pathlib import Path
import os
from dotenv import load_dotenv

load_dotenv()

algod_token = os.getenv('algod_token')
algod_server = os.getenv('algod_server')
algod_client = AlgodClient(algod_token, algod_server)

private_key = os.getenv('private_key')
address = address_from_private_key(private_key)

app_spec = Path(__file__).parent / './listings.arc32.json'
app_id = int(os.getenv('app_id'))
signer = AccountTransactionSigner(private_key)
params = algod_client.suggested_params()

app_client = ApplicationClient(
    algod_client=algod_client,
    app_spec=app_spec,
    app_id=app_id,
    signer=signer,
    sender=address,
    suggested_params=params,
)

atc = AtomicTransactionComposer()

asset_1 = int(os.getenv('asset_1'))

fee_payment_tx = PaymentTxn(sender=address, sp=params, receiver=app_client.app_address, amt=1000)
wrapped_fee_payment = TransactionWithSigner(fee_payment_tx, signer)

lister_address = address_from_private_key(os.getenv('private_key'))
counter = 1 #Remember to increment counter for additional listings

listing_box_coder = ABIType.from_string('(address,uint64)')
box_name = listing_box_coder.encode((lister_address, counter))

app_client.compose_call(
    atc, 
    call_abi_method='cancelUniversalListing', 
    listed_asset=asset_1, 
    listing_name=(lister_address,counter), 
    transfer_fee=wrapped_fee_payment, 
    transaction_parameters={'boxes': [[app_id, box_name]]})

results = atc.execute(algod_client, 2)

tx_ids = [results.tx_ids[i] for i in range(len(results.tx_ids))]
abi_results = [results.abi_results[i].return_value for i in range(len(results.abi_results))]

print(tx_ids)
print(abi_results)