catnotes

SDK Python - NEM Transactions

December 23, 2023

The Symbol Python SDK supports sending NEM 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.

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

from aiohttp import ClientSession
from symbolchain.CryptoTypes import PrivateKey
from symbolchain.facade.NemFacade import NemFacade
from symbolchain.nem.Network import NetworkTimestamp

NEM_PRIVATE_KEY = '<your private key>'
NEM_PRIVATE_KEY_2 = '<your other private key>'
NEM_API_ENDPOINT = 'http://your-testnet-node:7890'

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

  • NEM_PRIVATE_KEY Should be the private key of an account that will send XEM
  • NEM_PRIVATE_KEY_2 Should be the private key of an account that will receive XEM
  • NEM_API_ENDPOINT Should be the endpoint, including protocol and port, of a node in the NEM 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 NEM_PRIVATE_KEY, you can use the NEM faucet.

Network Time

When a transaction is sent to the NEM blockchain, it must have a timestamp no later than the current network time. In addition, it must have a deadline within one day of the current network time. Finally, the deadline must not be before the timestamp.

ℹ️ Since a valid timestamp can have a wide range of values, it is not useful for determining transaction confirmation time. The block time containing the transaction is its true confirmation time, not it's timestamp value. Due to this, timestamp was removed from Symbol transactions.

In order to set these properties 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 NEM REST endpoint
        async with session.get(f'{NEM_API_ENDPOINT}/time-sync/network-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['sendTimeStamp'] / 1000))
            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 time-sync/network-time. This route is used by the NEM node's time synchronization routine, but we use it here to determine the current time. It returns a JSON object with two properties: sendTimeStamp and receiveTimeStamp. Both of these are timestamps in milliseconds. Since the NEM network uses timestamps in seconds, we pick one of these (sendTimeStamp) and divide by 1000 to get seconds. Finally, we wrap this in a NetworkTimestamp object, which represents a NEM 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 NEM, a transaction is submitted to the network by sending two things:

  1. data Serialized transaction data
  2. signature Signature of data

Assume we have a JSON string composed of these two properties. 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 POST request to a NEM REST endpoint
        async with session.post(f'{NEM_API_ENDPOINT}/transaction/announce', 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 POST request to the route transaction/announce. The POST data is the JSON string, and we make sure to set the content type to application/json. Finally, we wait for the response. If everything went well, the response should have a message property equal to SUCCESS. For now, we just return this response object.

ℹ️ In NEM, we also have a binary object format alternative to JSON. This is the format nodes use to exchange information and is indicated by using the content type of application/binary. Further discussion of this is beyond the scope of this article.

XEM

In NEM, a unit of XEM can have up to 6 decimal places. In order to easily notate a number as an amount of full XEM units, let's add a small function to multiply a number by 10^6. While tiny, this will help with readability later:

def xem(amount):
    return amount * 1000000

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 (NEM) 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,
            'timestamp': network_time.timestamp,
            'deadline': network_time.add_hours(1).timestamp,

            **transaction_descriptor
        })

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
  • timestamp Transaction timestamp; the current network time retrieved by get_network_time
  • deadline One hour after timestamp; notice that we calculate this by using add_hours, which is provided by NetworkTimestamp

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}')

Finally, 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)
        if 'SUCCESS' == push_result['message']:
            print('transaction was sent and accepted successfully')
        else:
            print('there was an error with the transaction')

        print(push_result)

Here, we check the message property of the result and print an appropriate message.

Usage

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

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

    facade = NemFacade('testnet')

    signer_key_pair = facade.KeyPair(PrivateKey(NEM_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(NEM_PRIVATE_KEY_2))
    recipient_address = facade.network.public_key_to_address(recipient_key_pair.public_key)
    print(f'recipient address {recipient_address}')

Next, call the function with an appropriate descriptor:

    await prepare_and_send_transaction(facade, signer_key_pair, {
        'type': 'transfer_transaction_v2',
        'recipient_address': recipient_address,
        'amount': xem(2),
        'fee': xem(1)
    })

In this case, we're preparing and sending a transaction that sends 2 XEM from one account (NEM_PRIVATE_KEY) to another (NEM_PRIVATE_KEY_2). In NEM, for new applications, it is recommended to use V2 transfer transactions, so we specify transfer_transaction_v2. recipient_address is derived from NEM_PRIVATE_KEY_2. The amount and fee values take advantage of our xem helper function and are expanded to 2 XEM and 1 XEM, respectively.

If you followed all the steps, you can now easily send NEM 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).