SDK Python - Symbol Transactions

The Symbol Python SDK supports sending Symbol transactions. In this guide, we're going to look into how to send transactions asynchronously using only the SDK and aiohttp. We'll write a function that can be used to easily send a transaction to the network given a transaction descriptor.

If you have already read the SDK Python - NEM Transactions article, the examples should look very familiar. This is intentional and an explicit design goal of the SDKs.

Setup

It's recommended that you create a new project using this guide. Once you're set up, let's add our dependencies:

pip install aiohttp
pip install symbol-sdk-python

Now create a new python file and add the following to the top:

import asyncio
import time

from aiohttp import ClientSession
from symbolchain.CryptoTypes import PrivateKey
from symbolchain.facade.SymbolFacade import SymbolFacade
from symbolchain.sc import Amount
from symbolchain.symbol.Network import NetworkTimestamp

SYMBOL_PRIVATE_KEY = '<your private key>'
SYMBOL_PRIVATE_KEY_2 = '<your other private key>'
SYMBOL_API_ENDPOINT = 'http://your-testnet-node:7890'

Make sure to set the top-level constants to appropriate values.

• SYMBOL_PRIVATE_KEY Should be the private key of an account that will send XYM
• SYMBOL_PRIVATE_KEY_2 Should be the private key of an account that will receive XYM
• SYMBOL_API_ENDPOINT Should be the endpoint, including protocol and port, of a node in the Symbol network

ℹ️ For this guide, we don't really need a second private key. You can replace that with an address instead if you prefer.

If you need to fund the SYMBOL_PRIVATE_KEY, you can use the Symbol faucet.

Network Time

When a transaction is sent to the Symbol blockchain, it must have a deadline past the current network time. In addition, it must have a deadline within six hours of its containing block. The one exception is aggregate bonded transactions, which can have a deadline up to two days in the future.

In order to set a deadline correctly, we need to query the network for the current network time. Let's create a function called get_network_time to do that for us:

async def get_network_time():
    async with ClientSession(raise_for_status=True) as session:
        # initiate a HTTP GET request to a SYMBOL REST endpoint
        async with session.get(f'{SYMBOL_API_ENDPOINT}/node/time') as response:
            # wait for the (JSON) response
            response_json = await response.json()

            # extract the network time from the json
            timestamp = NetworkTimestamp(int(response_json['communicationTimestamps']['sendTimestamp']))
            return timestamp

First, we create a ClientSession and set raise_for_status so that it will translate server errors into exceptions. Next, we send a GET request to the route node/time. This route is used by the Symbol node's time synchronization routine, but we use it here to determine the current time. It returns a JSON object with a communicationTimestamps object containing to properties: sendTimestamp and receiveTimestamp. Both of these are Symbol network timestamps in (milliseconds). We pick one of these (sendTimestamp) and wrap it in a NetworkTimestamp object, which represents a Symbol timestamp. For our purposes, it provides helper functions for adding additional time.

Push Transaction to Network

We also need to be able to push transactions to the network. In Symbol, a transaction is submitted to the network by sending a serialized transaction payload, including its signature. Assume we have a JSON string composed of this data. Let's create a function called push_transaction that will send it to the network:

async def push_transaction(json_payload):
    async with ClientSession(raise_for_status=True) as session:
        # initiate a HTTP PUT request to a SYMBOL REST endpoint
        async with session.put(f'{SYMBOL_API_ENDPOINT}/transactions', data=json_payload, headers={
            'Content-Type': 'application/json'
        }) as response:
            # wait for the (JSON) response
            return await response.json()

First, we create a ClientSession and set raise_for_status so that it will translate server errors into exceptions. Next, we send a PUT request to the route transactions. The PUT data is the JSON string, and we make sure to set the content type to application/json. Finally, we wait for the response. In Symbol, transaction processing is asynchronous, so we only get confirmation that the transaction was submitted. We need to monitor the node to determine if the transaction is accepted (moved into the unconfirmed transactions cache) and eventually confirmed (included in a block).

Wait for Status

We can monitor the transaction status by polling the network. For best results, query the transaction status from the same node to which the transaction was sent.

async def wait_for_transaction_status(transaction_hash, desired_status):
    async with ClientSession(raise_for_status=False) as session:
        for _ in range(6):
            # query the status of the transaction
            async with session.get(f'{SYMBOL_API_ENDPOINT}/transactionStatus/{transaction_hash}') as response:
                # wait for the (JSON) response
                response_json = await response.json()

                # check if the transaction has transitioned
                if 200 == response.status:
                    status = response_json['group']
                    print(f'transaction {transaction_hash} has status "{status}"')
                    if desired_status == status:
                        print(f'transaction {transaction_hash} has transitioned to {desired_status}')
                        return

                    if 'failed' == status:
                        print(f'transaction {transaction_hash} failed validation: {response_json["code"]}')
                        break
                else:
                    print(f'transaction {transaction_hash} has unknown status')

            # if not, wait 20s before trying again
            time.sleep(20)

        # fail if the transaction didn't transition to the desired status after 2m
        raise RuntimeError(f'transaction {transaction_hash} did not transition to {desired_status} in alloted time period')

First, we create a ClientSession and do not set raise_for_status because we want to manually handle server errors. Next, we send a GET request to the route transactionStatus, including the hash of the transaction we want to query. If the request succeeds, we inspect the response to determine the transaction's status. If the status is the desired status or failed, we stop polling. Otherwise, we sleep for five seconds before trying again.

Get (Confirmed) Transaction From Network (By Hash)

After a transaction is confirmed, we can query it by hash.

async def get_transaction_by_hash(transaction_hash):
    async with ClientSession(raise_for_status=True) as session:
        # initiate a HTTP GET request to a SYMBOL REST endpoint
        async with session.get(f'{SYMBOL_API_ENDPOINT}/transactions/confirmed/{transaction_hash}') as response:
            response_json = await response.json()
            return response_json

First, we create a ClientSession and set raise_for_status so that it will translate server errors into exceptions. Next, we send a GET request to the route transactions/confirmed. The hash of the transaction to query is provided as a query string parameter. Upon success, the result is a JSON transaction and metadata pair.

Currency Mosaic ID

In Symbol, networks can have different base mosaics. When writing generic code, it's generally a best practice to query the network for its currency mosaic id. This will allow the code to work seamlessly with different networks; for example, mainnet and testnet.

Symbol REST nodes expose the entire config-network.properties file. Let's query that configuration and extract the currency mosaic id:

async def get_currency_mosaic_id():
    async with ClientSession(raise_for_status=True) as session:
        # initiate a HTTP GET request to a SYMBOL REST endpoint
        async with session.get(f'{SYMBOL_API_ENDPOINT}/network/properties') as response:
            # wait for the (JSON) response
            response_json = await response.json()

            # extract the currency mosaic id from the json
            formatted_currency_mosaic_id = response_json['chain']['currencyMosaicId']
            return int(formatted_currency_mosaic_id.replace('\'', ''), 16)

First, we create a ClientSession and set raise_for_status so that it will translate server errors into exceptions. Next, we send a GET request to the route network/properties. It returns a JSON object with the full network configuration. From that, we extract the currencyMosaicId setting. Finally, we need to remove its thousands separators ('''), and convert it from a hex string to an integer before returning.,

Transaction Descriptor

In the Symbol SDK, a dictionary of transaction properties is called a transaction descriptor. The SDK accepts this descriptor and returns a transaction object.

Let's write a function that accepts three arguments:

• facade Symbol SDK facade for interacting with a (Symbol) network
• signer_key_pair Key pair of the transaction sender that will be used to sign the transaction
• transaction_descriptor Transaction descriptor composed of the desired transaction properties

async def prepare_and_send_transaction(facade, signer_key_pair, transaction_descriptor):

First, we need to query the network time using the get_network_time function we wrote earlier:

    async with ClientSession(raise_for_status=True) as session:
        # get the current network time from the network
        network_time = await get_network_time(session)

ℹ️ Notice that we changed get_network_time to accept a ClientSession instead of create one. Where possible, it is recommended to reuse a ClientSession object because each session encapsulates a connection pool. Taking advantage of this connection pooling can give your application a performance boost.

Second, we need to create a transaction object from the transaction_descriptor argument. We can do this by using the facade's transaction factory create method:

        # create the transaction
        transaction = facade.transaction_factory.create({
            'signer_public_key': signer_key_pair.public_key,
            'deadline': network_time.add_hours(1).timestamp,

            **transaction_descriptor
        })
        transaction.fee = Amount(100 * transaction.size)

In addition, we set some properties that are common across all transactions:

• signer_public_key Public key of the transaction signer; derived from signer_key_pair
• deadline One hour after timestamp; notice that we calculate this by using add_hours, which is provided by NetworkTimestamp
• fee Symbol transaction fees are cost-per-byte fees; here we're setting the fee assuming a fee multiplier of 100

Third, we need to sign the transaction and construct an appropriate transaction payload:

        # sign the transaction and attach its signature
        signature = facade.sign_transaction(signer_key_pair, transaction)
        json_payload = facade.transaction_factory.attach_signature(transaction, signature)

sign_transaction signs transaction with signer_key_pair and returns the resulting signature. attach_signature constructs a transaction payload, composed of both serialized transaction data and signature, that can be sent directly to the network.

Fourth, as an optional step, we calculate the transaction hash by calling hash_transaction:

        # hash the transaction (this is dependent on the signature)
        transaction_hash = facade.hash_transaction(transaction)
        print(f'transaction hash {transaction_hash}')

Fifth, we send the transaction to the network using the push_transaction function we created earlier:

        # send the transaction to the network
        push_result = await push_transaction(session, json_payload)
        print('transaction was sent')

Here, we only know that the transaction was send, but not whether or not it was accepted and/or confirmed.

Finally, let's query the transaction status and wait for it to be confirmed:

    # wait for the transaction to be confirmed
    await wait_for_transaction_status(transaction_hash, 'confirmed')
    return transaction_hash

We simply need to call the helper function we wrote earlier and wait for the transaction to be confirmed.

Usage

Let's see how we can use the function we just wrote to send a transaction to the network!

First, create a SymbolFacade for the test network and load two accounts:

    facade = SymbolFacade('testnet')

    signer_key_pair = facade.KeyPair(PrivateKey(SYMBOL_PRIVATE_KEY))
    signer_address = facade.network.public_key_to_address(signer_key_pair.public_key)
    print(f'signer address {signer_address}')

    recipient_key_pair = facade.KeyPair(PrivateKey(SYMBOL_PRIVATE_KEY_2))
    recipient_address = facade.network.public_key_to_address(recipient_key_pair.public_key)
    print(f'recipient address {recipient_address}')

Next, query the network's currency mosaic id:

    currency_mosaic_id = await get_currency_mosaic_id()
    print(f'currency mosaic id {currency_mosaic_id:16X}')

Finally, call the function with an appropriate descriptor:

    await prepare_and_send_transaction(facade, signer_key_pair, {
        'type': 'transfer_transaction_v1',
        'recipient_address': recipient_address,
        'mosaics': [
            {'mosaic_id': currency_mosaic_id, 'amount': 2_000000}
        ]
    })

In this case, we're preparing and sending a transaction that sends 2 XEM from one account (SYMBOL_PRIVATE_KEY) to another (SYMBOL_PRIVATE_KEY_2). This is a transfer transaction, so we specify type as transfer_transaction_v1. recipient_address is derived from SYMBOL_PRIVATE_KEY_2. The mosaics array allows us to send multiple mosaics at once, but, here, we're only sending two units of the currency mosaic id (XYM).

If you followed all the steps, you can now easily send Symbol transactions to the network by calling prepare_and_send_transaction! The code is very DRY (don't repeat yourself). In order to send different transactions, you only have to change the transaction descriptor passed to prepare_and_send_transaction. In addition, the common fields are all set correctly in one place, and don't need to clutter your code (or descriptors).

Bonus - Symbol Connector

If you're wondering how you'll be able to discover all of the Symbol REST endpoints on your own, there's a separate library symbol-lightapi that will help. This library is not yet comprehensive - we're always open to PRs - and is more of a work in progress at the moment. Nonetheless, it can be used to replace many of the helper functions we're written above. Let's take a quick look at what it offers.

First, we need to install the symbol-lightapi package via pip:

pip install symbol-lightapi

We can use this package to create a connector around a Symbol node. With this connector, we can query Symbol network properties. In fact, get_network_time and get_currency_mosaic_id can be completely replaced by function calls on this connector:

from symbollightapi.connector.SymbolConnector import SymbolConnector


async def query_network_with_connector():
    connector = SymbolConnector(SYMBOL_API_ENDPOINT)

    currency_mosaic_id = await connector.currency_mosaic_id()
    print(f'currency_mosaic_id: {currency_mosaic_id}')

    network_timestamp = await connector.network_time()
    print(f' network_timestamp: {network_timestamp}')

In addition, we can use the connector to send transactions to the network:

    try:
        await connector.announce_transaction(transaction)
    except NodeException:
        pass  # ignore 202

ℹ️ At the moment, there is a bug where HTTP 202 responses are treated as errors. Once this is fixed, the try/except block can and should be removed.

Changelog

• 2024-11-26: update prepare_and_send_transaction to return the hash of the sent transaction
• 2024-11-26: add get_transaction_by_hash in Get (Confirmed) Transaction From Network (By Hash)