Tutorials

Are you new to IOTA in Python? Don’t worry, we got you covered! With the walkthrough examples of this section, you will be a master of PyOTA.

In each section below, a code snippet will be shown and discussed in detail to help you understand how to carry out specific tasks with PyOTA.

The example scripts displayed here can also be found under examples/tutorials/ directory in the repository. Run them in a Python environment that has PyOTA installed. See Install PyOTA for more info.

If you feel that something is missing or not clear, please post your questions and suggestions in the PyOTA Bug Tracker.

Let’s get to it then!

1. Hello Node

In this example, you will learn how to:

  • Import the iota package into your application.

  • Instantiate an API object for communication with the IOTA network.

  • Request information about the IOTA node you are connected to.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# Import neccessary modules
from iota import Iota
from pprint import pprint

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443')

# Request information about the node
response = api.get_node_info()

# Using pprint instead of print for a nicer looking result in the console
pprint(response)

Discussion

1
2
3
# Import neccessary modules
from iota import Iota
from pprint import pprint

First things first, we need to import in our application the modules we intend to use. PyOTA provide the iota package, therefore, whenever you need something from the library, you need to import it from there.

Notice, how we import the Iota object, that defines a so-called extended API object. We will use this to send and receive data from the network. Read more about API objects at PyOTA API Classes.

We also import the pprint method that prettifies the output before printing it to the console.

5
6
# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443')

Next, we declare an API object. Since this object handles the communication, we need to specify an IOTA node to connect to in the form of an URI. Note, that the library will parse this string and will throw an exception if it is not a valid one.

8
9
# Request information about the node
response = api.get_node_info()

Then we can call the Iota.get_node_info() method of the API object to get some basic info about the node.

11
12
# Using pprint instead of print for a nicer looking result in the console
pprint(response)

Finally, we print out the response. It is important to note, that all API methods return a python dictionary. Refer to the method’s documentation to determine what exactly is in the response dict. Here for example, we could list the features of the node:

pprint(response['features'])

2. Send Data

In this example, you will learn how to:

  • Encode data to be stored on the Tangle.

  • Generate a random IOTA address that doesn’t belong to anyone.

  • Create a zero-value transaction with custom payload.

  • Send a transaction to the network.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from iota import Iota, TryteString, Address, Tag, ProposedTransaction
from pprint import pprint

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443', devnet=True)

# Prepare custom data
my_data = TryteString.from_unicode('Hello from the Tangle!')

# Generate a random address that doesn't have to belong to anyone
my_address = Address.random()

# Tag is optional here
my_tag = Tag(b'MY9FIRST9TAG')

# Prepare a transaction object
tx = ProposedTransaction(
    address=my_address,
    value=0,
    tag=my_tag,
    message=my_data
)

# Send the transaction to the network
response = api.send_transfer([tx])

pprint('Check your transaction on the Tangle!')
pprint('https://utils.iota.org/transaction/%s/devnet' % response['bundle'][0].hash)

Discussion

1
2
3
4
5
from iota import Iota, TryteString, Address, Tag, ProposedTransaction
from pprint import pprint

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443', devnet=True)

We have seen this part before. Note, that now we import more objects which we will use to construct our transaction.

Notice devnet=True in the argument list of the API instantiation. We tell the API directly that we will use IOTA’s testnet, known as the devnet. By default, the API is configured for the mainnet.

7
8
# Prepare custom data
my_data = TryteString.from_unicode('Hello from the Tangle!')

If you read Basic Concepts and PyOTA Types, it shouldn’t be a surprise to you that most things in IOTA are represented as trytes, that are TryteString in PyOTA.

Here, we encode our message with TryteString.from_unicode() into trytes.

10
11
# Generate a random address that doesn't have to belong to anyone
my_address = Address.random()

To put anything (transactions) on the Tangle, it needs to be associated with an address. Since we will be posting a zero-value transaction, nobody has to own this address; therefore we can use the TryteString.random() (an Address is just a TryteString with some additional attributes and fixed length) method to generate one.

13
14
# Tag is optional here
my_tag = Tag(b'MY9FIRST9TAG')

To tag our transaction, we might define a custom Tag object. Notice, that the b means we are passing a bytestring value instead of a unicode string. This is so that PyOTA interprets our input as literal trytes, rather than a unicode string that needs to be encoded into trytes.

When passing a bytestring to a PyOTA class, each byte is interpreted as a tryte; therefore we are restricted to the tryte alphabet.

16
17
18
19
20
21
22
# Prepare a transaction object
tx = ProposedTransaction(
    address=my_address,
    value=0,
    tag=my_tag,
    message=my_data
)

It’s time to construct the transaction. According to Transaction Types, PyOTA uses ProposedTransaction to build transactions that are not yet broadcast to the network. Oberve, that the value=0 means this is a zero-value transaction.

24
25
# Send the transaction to the network
response = api.send_transfer([tx])

Next, we send the transfer to the node for tip selection, proof-of-work calculation, broadcasting and storing. The API takes care of all these tasks, and returns the resulting Bundle object.

Note

send_transfer() takes a list of ProposedTransaction objects as its transfers argument. An IOTA transfer (bundle) usually consists of multiple transactions linked together, however, in this simple example, there is only one transaction in the bundle. Regardless, you need to pass this sole transaction as a list of one transaction.

27
28
pprint('Check your transaction on the Tangle!')
pprint('https://utils.iota.org/transaction/%s/devnet' % response['bundle'][0].hash)

Finally, we print out the transaction’s link on the Tangle Explorer. Observe how we extract the transaction hash from the response dict. We take the first element of the bundle, as it is just a sequence of transactions, and access its hash attribute.

3. Fetch Data

In this example, you will learn how to:

  • Fetch transaction objects from the Tangle based on a criteria.

  • Decode messages from transactions.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from iota import Iota, Address
from iota.codecs import TrytesDecodeError

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443', devnet=True)

# Address to fetch data from
# Replace with your random generated address from Tutorial 2. to fetch the data
# that you uploaded.
addy = Address(b'WWO9DRAUDDSDSTTUPKJRNPSYLWAVQBBXISLKLTNDPVKOPMUERDUELLUPHNT9L9YWBDKOLYVWRAFRKIBLP')

print('Fetching data from the Tangle...')
# Fetch the transaction objects of the address from the Tangle
response = api.find_transaction_objects(addresses=[addy])

if not response['transactions']:
    print('Couldn\'t find data for the given address.')
else:
    print('Found:')
    # Iterate over the fetched transaction objects
    for tx in response['transactions']:
        # data is in the signature_message_fragment attribute as trytes, we need
        # to decode it into a unicode string
        data = tx.signature_message_fragment.decode(errors='ignore')
        print(data)

Discussion

1
2
3
4
5
from iota import Iota, Address
from iota.codecs import TrytesDecodeError

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443', devnet=True)

The usual part again, but we also import TrytesDecodeError from iota.codec. We will use this to detect if the fetched trytes contain encoded text.

 7
 8
 9
10
# Address to fetch data from
# Replace with your random generated address from Tutorial 2. to fetch the data
# that you uploaded.
addy = Address(b'WWO9DRAUDDSDSTTUPKJRNPSYLWAVQBBXISLKLTNDPVKOPMUERDUELLUPHNT9L9YWBDKOLYVWRAFRKIBLP')

We declare an IOTA address on the Tangle to fetch data from. You can replace this address with your own from the previous example 2. Send Data, or just run it as it is.

12
13
14
print('Fetching data from the Tangle...')
# Fetch the transaction objects of the address from the Tangle
response = api.find_transaction_objects(addresses=[addy])

We use find_transaction_objects() extended API method to gather the transactions that belong to our address. This method is also capable of returning Transaction objects for bundle hashes, tags or approving transactions. Note that you can supply multiple of these, the method always returns a dict with a list of transactions.

Note

Remember, that the parameters need to be supplied as lists, even if there is only one value.

16
17
18
19
20
21
22
23
24
25
if not response['transactions']:
    print('Couldn\'t find data for the given address.')
else:
    print('Found:')
    # Iterate over the fetched transaction objects
    for tx in response['transactions']:
        # data is in the signature_message_fragment attribute as trytes, we need
        # to decode it into a unicode string
        data = tx.signature_message_fragment.decode(errors='ignore')
        print(data)

Finally, we extract the data we are looking for from the transaction objects. A Transaction has several attributes, one of which is the signature_message_fragment. This contains the payload message for zero-value transactions, and the digital signature that authorizes spending for value transactions.

Since we are interested in data now, we decode its content (raw trytes) into text. Notice, that we pass the errors='ignore' argument to the decode() method to drop values we can’t decode using utf-8, or if the raw trytes can’t be decoded into legit bytes. A possible reason for the latter can be if the attribute contains a signature rather than a message.

4.a Generate Address

In this example, you will learn how to:

  • Generate a random seed.

  • Generate an IOTA address that belongs to your seed.

  • Acquire free devnet IOTA tokens that you can use to play around with.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from iota import Iota, Seed

# Generate a random seed, or use one you already have (for the devnet)
print('Generating a random seed...')
my_seed = Seed.random()
# my_seed = Seed(b'MYCUSTOMSEED')
print('Your seed is: ' + str(my_seed))

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=my_seed,
    devnet=True,
)

print('Generating the first unused address...')
# Generate the first unused address from the seed
response = api.get_new_addresses()

addy = response['addresses'][0]

print('Your new address is: ' + str(addy))
print('Go to https://faucet.devnet.iota.org/ and enter you address to receive free devnet tokens.')

Discussion

1
2
3
4
5
6
7
from iota import Iota, Seed

# Generate a random seed, or use one you already have (for the devnet)
print('Generating a random seed...')
my_seed = Seed.random()
# my_seed = Seed(b'MYCUSTOMSEED')
print('Your seed is: ' + str(my_seed))

We start off by generating a random seed with the help of the library. You are also free to use your own seed, just uncomment line 6 and put it there.

If you choose to generate one, your seed is written to the console so that you can save it for later. Be prepared to do so, because you will have to use it in the following tutorials.

 9
10
11
12
13
14
# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=my_seed,
    devnet=True,
)

Notice, how we pass the seed argument to the API class’s init method. Whenever the API needs to work with addresses or private keys, it will derive them from this seed.

Important

Your seed never leaves the library and your computer. Treat your (mainnet) seed like any other password for a financial service: safe. If your seed is compromised, attackers can steal your funds.

16
17
18
19
20
print('Generating the first unused address...')
# Generate the first unused address from the seed
response = api.get_new_addresses()

addy = response['addresses'][0]

To generate a new address, we call get_new_addresses() extended API method. Without arguments, this will return a dict with the first unused address starting from index 0. An unused address is address that has no transactions referencing it on the Tangle and was never spent from.

If we were to generate more addresses starting from a desired index, we could specify the start and count parameters. Read more about how to generate addresses in PyOTA at Generating Addresses.

On line 20 we access the first element of the list of addresses in the response dictionary.

22
23
print('Your new address is: ' + str(addy))
print('Go to https://faucet.devnet.iota.org/ and enter you address to receive free devnet tokens.')

Lastly, the address is printed to the console, so that you can copy it. Visit https://faucet.devnet.iota.org/ and enter the address to receive free devnet tokens of 1000i.

You might need to wait 1-2 minutes until the sum arrives to you address. To check your balance, go to 4.b Check Balance or 4.c Get Account Data.

4.b Check Balance

In this example, you will learn how to:

  • Check the balance of a specific IOTA address.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from iota import Iota, Address
import time

# Put your address from Tutorial 4.a here
my_address = Address(b'YOURADDRESSFROMTHEPREVIOUSTUTORIAL')

# Declare an API object
api = Iota(adapter='https://nodes.devnet.iota.org:443', devnet=True)

# Script actually runs until you load up your address
success = False

while not success:
    print('Checking balance on the Tangle for a specific address...')
    # API method to check balance
    response = api.get_balances(addresses=[my_address])

    # response['balances'] is a list!
    if response['balances'][0]:
        print('Found the following information for address ' + str(my_address) + ':')
        print('Balance: ' + str(response['balances'][0]) + 'i')
        success = True
    else:
        print('Zero balance found, retrying in 30 seconds...')
        time.sleep(30)

Discussion

1
2
3
4
5
6
7
8
from iota import Iota, Address
import time

# Put your address from Tutorial 4.a here
my_address = Address(b'YOURADDRESSFROMTHEPREVIOUSTUTORIAL')

# Declare an API object
api = Iota(adapter='https://nodes.devnet.iota.org:443', devnet=True)

The first step to check the balance of an address is to actually have an address. Exchange the sample address on line 5 with your generated address from 4.a Generate Address.

Since we don’t need to generate an address, there is no need for a seed to be employed in the API object. Note the time import, we need it for later.

10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# Script actually runs until you load up your address
success = False

while not success:
    print('Checking balance on the Tangle for a specific address...')
    # API method to check balance
    response = api.get_balances(addresses=[my_address])

    # response['balances'] is a list!
    if response['balances'][0]:
        print('Found the following information for address ' + str(my_address) + ':')
        print('Balance: ' + str(response['balances'][0]) + 'i')
        success = True
    else:
        print('Zero balance found, retrying in 30 seconds...')
        time.sleep(30)

Our script will poll the network for the address balance as long as the returned balance is zero. Therefore, the address you declared as my_address should have some balance. If you see the Zero balance found... message a couple of times, head over to https://faucet.devnet.iota.org/ and load up your address.

get_balances() returns the confirmed balance of the address. You could supply multiple addresses at the same time and get their respective balances in a single call. Don’t forget, that the method returns a dict. More details about it can be found at get_balances().

4.c Get Account Data

In this example, you will learn how to:

  • Gather addresses, balance and bundles associated with your seed on the Tangle.

Warning

Account in the context of this example is not to be confused with the Account Module, that is a feature yet to be implemented in PyOTA.

Account here simply means the addresses and funds that belong to your seed.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from iota import Iota, Seed
from pprint import pprint
import time

# Put your seed from Tutorial 4.a here
my_seed = Seed(b'YOURSEEDFROMTHEPREVIOUSTUTORIAL99999999999999999999999999999999999999999999999999')

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=my_seed,
    devnet=True
)

# Script actually runs until it finds balance
success = False

while not success:
    print('Checking account information on the Tangle...')
    # Gather addresses, balance and bundles
    response = api.get_account_data()

    # response['balance'] is an integer!
    if response['balance']:
        print('Found the following information based on your seed:')
        pprint(response)
        success = True
    else:
        print('Zero balance found, retrying in 30 seconds...')
        time.sleep(30)

Discussion

1
2
3
from iota import Iota, Seed
from pprint import pprint
import time

We will need pprint for a prettified output of the response dict and time for polling until we find non-zero balance.

 5
 6
 7
 8
 9
10
11
12
13
# Put your seed from Tutorial 4.a here
my_seed = Seed(b'YOURSEEDFROMTHEPREVIOUSTUTORIAL99999999999999999999999999999999999999999999999999')

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=my_seed,
    devnet=True
)

Copy your seed from 4.a Generate Address onto line 6. The API will use your seed to generate addresses and look for corresponding transactions on the Tangle.

15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# Script actually runs until it finds balance
success = False

while not success:
    print('Checking account information on the Tangle...')
    # Gather addresses, balance and bundles
    response = api.get_account_data()

    # response['balance'] is an integer!
    if response['balance']:
        print('Found the following information based on your seed:')
        pprint(response)
        success = True
    else:
        print('Zero balance found, retrying in 30 seconds...')
        time.sleep(30)

Just like in the previous example, we will poll for information until we find a non-zero balance. get_account_data() without arguments generates addresses from index 0 until it finds the first unused. Then, it queries the node about bundles of those addresses and sums up their balance.

Note

If you read get_account_data() documentation carefully, you notice that you can gain control over which addresses are checked during the call by specifying the start and stop index parameters.

This can be useful when your addresses with funds do not follow each other in the address namespace, or a snapshot removed transactions from the Tangle. It is recommended that you keep a local database of your already used address indices.

Once implemented in PyOTA, Account Module will address the aforementioned problems.

The response dict contains the addresses, bundles and total balance of your seed.

5. Send Tokens

In this example, you will learn how to:

  • Construct a value transfer with PyOTA.

  • Send a value transfer to an arbitrary IOTA address.

  • Analyze a bundle of transactions on the Tangle.

Note

As a prerequisite to this tutorial, you need to have completed 4.a Generate Address, and have a seed that owns devnet tokens.

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from iota import Iota, Seed, Address, TryteString, ProposedTransaction, Tag

# Put your seed here from Tutorial 4.a, or a seed that owns tokens (devnet)
my_seed = Seed(b'YOURSEEDFROMTHEPREVIOUSTUTORIAL')

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=my_seed,
    devnet=True,
)

# Addres to receive 1i
# Feel free to replace it. For example, run the code from Tutorial 4.a
# and use that newly generated address with a 'fresh' seed.
receiver = Address(b'WWUTQBO99YDCBVBPAPVCANW9ATSNUPPLCPGDQXGQEVLUBSFHCEWOA9DIYYOXJONDIRHYPXQXOYXDPHREZ')

print('Constructing transfer of 1i...')
# Create the transfer object
tx = ProposedTransaction(
    address=receiver,
    value=1,
    message=TryteString.from_unicode('I just sent you 1i, use it wisely!'),
    tag=Tag('VALUETX'),
)

print('Preparing bundle and sending it to the network...')
# Prepare the transfer and send it to the network
response = api.send_transfer(transfers=[tx])

print('Check your transaction on the Tangle!')
print('https://utils.iota.org/bundle/%s/devnet' % response['bundle'].hash)

Discussion

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from iota import Iota, Seed, Address, TryteString, ProposedTransaction, Tag

# Put your seed here from Tutorial 4.a, or a seed that owns tokens (devnet)
my_seed = Seed(b'YOURSEEDFROMTHEPREVIOUSTUTORIAL')

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=my_seed,
    devnet=True,
)

We are going to send a value transaction, that requires us to prove that we own the address containg the funds to spend. Therefore, we need our seed from which the address was generated.

Put your seed from 4.a Generate Address onto line 4. We pass this seed to the API object, that will utilize it for signing the transfer.

13
14
15
16
# Addres to receive 1i
# Feel free to replace it. For example, run the code from Tutorial 4.a
# and use that newly generated address with a 'fresh' seed.
receiver = Address(b'WWUTQBO99YDCBVBPAPVCANW9ATSNUPPLCPGDQXGQEVLUBSFHCEWOA9DIYYOXJONDIRHYPXQXOYXDPHREZ')

In IOTA, funds move accross addresses, therefore we need to define a receiver address. For testing value transfers, you should send the funds only to addresses that you control; if you use a randomly-generated receiver address, you won’t be able to recover the funds afterward! Re-run 4.a Generate Address for a new seed and a new address, or just paste a valid IOTA address that you own onto line 16.

18
19
20
21
22
23
24
25
print('Constructing transfer of 1i...')
# Create the transfer object
tx = ProposedTransaction(
    address=receiver,
    value=1,
    message=TryteString.from_unicode('I just sent you 1i, use it wisely!'),
    tag=Tag('VALUETX'),
)

We declare a ProposedTransaction object like we did before, but this time, with value=1 parameter. The smallest value you can send is 1 iota (“1i”), there is no way to break it into smaller chunks. It is a really small value anyway. You can also attach a message to the transaction, for example a little note to the beneficiary of the payment.

27
28
29
print('Preparing bundle and sending it to the network...')
# Prepare the transfer and send it to the network
response = api.send_transfer(transfers=[tx])

To actually send the transfer, all you need to do is call send_transfer() extended API method. This method will take care of:

  • Gathering inputs (addresses you own and have funds) to fund the 1i transfer.

  • Generating a new change_address, and automatically sending the remaining funds (balance of chosen inputs - 1i) from inputs to change_address.

    Warning

    This step is extremely important, as it prevents you from spending twice from the same address.

    When an address is used as an input, all tokens will be withdrawn. Part of the tokens will be used to fund your transaction, the rest will be transferred to change_address.

  • Constructing the transfer bundle with necessary input and output transactions.

  • Finalizing the bundle and signing the spending transactions.

  • Doing proof-of-work for each transaction in the bundle and sending it to the network.

31
32
print('Check your transaction on the Tangle!')
print('https://utils.iota.org/bundle/%s/devnet' % response['bundle'].hash)

Open the link and observe the bundle you have just sent to the Tangle. Probably it will take a couple of seconds for the network to confirm it.

What you see is a bundle with 4 transactions in total, 1 input and 3 outputs. But why are there so many transactions?

  • There is one transaction that withdraws iotas, this has negative value. To authorize this spending, a valid signature is included in the transaction’s signature_message_fragment field. The signature however is too long to fit into one transaction, therefore the library appends a new, zero-value transaction to the bundle that holds the second part of the signature. This you see on the output side of the bundle.

  • A 1i transaction to the receiver address spends part of the withdrawn amount.

  • The rest is transfered to change_address in a new output transaction.

Once the bundle is confirmed, try rerunning the script from 4.c Get Account Data with the same seed as in this tutorial. Your balance should be decremented by 1i, and you should see a new address, which was actually the change_address.

6. Store Encrypted Data

In this example, you will learn how to:

  • Convert Python data structures to JSON format.

  • Encrypt data and include it in a zero-value transaction.

  • Store the zero-value transaction with encrypted data on the Tangle.

Warning

We will use the simple-crypt external library for encryption/decryption. Before proceeding to the tutorial, make sure you install it by running:

pip install simple-crypt

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
"""
Encrypt data and store it on the Tangle.

simplecrypt library is needed for this example (`pip install simple-crypt`)!
"""
from iota import Iota, TryteString, Tag, ProposedTransaction
from simplecrypt import encrypt
from base64 import b64encode
from getpass import getpass

import json

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=b'YOURSEEDFROMTHEPREVIOUSTUTORIAL',
    devnet=True,
)

# Some confidential information
data = {
    'name' : 'James Bond',
    'age' : '32',
    'job' : 'agent',
    'address' : 'London',
}

# Convert to JSON format
json_data = json.dumps(data)

# Ask user for a password to use for encryption
password = getpass('Please supply a password for encryption:')

print('Encrypting data...')
# Encrypt data
# Note, that in Python 3, encrypt returns 'bytes'
cipher = encrypt(password, json_data)

# Encode to base64, output contains only ASCII chars
b64_cipher = b64encode(cipher)

print('Constructing transaction locally...')
# Convert to trytes
trytes_encrypted_data = TryteString.from_bytes(b64_cipher)

# Generate an address from your seed to post the transfer to
my_address = api.get_new_addresses(index=42)['addresses'][0]

# Tag is optional here
my_tag = Tag(b'CONFIDENTIALINFORMATION')

# Prepare a transaction object
tx = ProposedTransaction(
    address=my_address,
    value=0,
    tag=my_tag,
    message=trytes_encrypted_data,
)

print('Sending transfer...')
# Send the transaction to the network
response = api.send_transfer([tx])

print('Check your transaction on the Tangle!')
print('https://utils.iota.org/transaction/%s/devnet' % response['bundle'][0].hash)
print('Tail transaction hash of the bundle is: %s' % response['bundle'].tail_transaction.hash)

Discussion

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
"""
Encrypt data and store it on the Tangle.

simplecrypt library is needed for this example (`pip install simple-crypt`)!
"""
from iota import Iota, TryteString, Tag, ProposedTransaction
from simplecrypt import encrypt
from base64 import b64encode
from getpass import getpass

import json

# Declare an API object
api = Iota(
    adapter='https://nodes.devnet.iota.org:443',
    seed=b'YOURSEEDFROMTHEPREVIOUSTUTORIAL',
    devnet=True,
)

We will use the encrypt method to encipher the data, and b64encode for representing it as ASCII characters. getpass will prompt the user for a password, and the json library is used for JSON formatting.

We will need an address to upload the data, therefore we need to supply the seed to the Iota API instance. The address will be generated from this seed.

20
21
22
23
24
25
26
# Some confidential information
data = {
    'name' : 'James Bond',
    'age' : '32',
    'job' : 'agent',
    'address' : 'London',
}

The data to be stored is considered confidential information, therefore we can’t just put it on the Tangle as plaintext so everyone can read it. Think of what would happen if the world’s most famous secret agent’s identity was leaked on the Tangle…

28
29
# Convert to JSON format
json_data = json.dumps(data)

Notice, that data is a Python dict object. As a common way of exchanging data on the web, we would like to convert it to JSON format. The json.dumps() method does exactly that, and the result is a JSON formatted plaintext.

31
32
33
34
35
36
37
38
39
40
# Ask user for a password to use for encryption
password = getpass('Please supply a password for encryption:')

print('Encrypting data...')
# Encrypt data
# Note, that in Python 3, encrypt returns 'bytes'
cipher = encrypt(password, json_data)

# Encode to base64, output contains only ASCII chars
b64_cipher = b64encode(cipher)

Next, we will encrypt this data with a secret password we obtain from the user.

Note

When you run this example, please remember the password at least until the next tutorial!

The output of the encrypt method is a bytes object in Python3 and contains many special characters. This is a problem, since we can only convert ASCII characters from bytes directly into TryteString.

Therefore, we first encode our binary data into ASCII characters with Base64 encoding.

42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
print('Constructing transaction locally...')
# Convert to trytes
trytes_encrypted_data = TryteString.from_bytes(b64_cipher)

# Generate an address from your seed to post the transfer to
my_address = api.get_new_addresses(index=42)['addresses'][0]

# Tag is optional here
my_tag = Tag(b'CONFIDENTIALINFORMATION')

# Prepare a transaction object
tx = ProposedTransaction(
    address=my_address,
    value=0,
    tag=my_tag,
    message=trytes_encrypted_data,
)

Now, we are ready to construct the transfer. We convert the encrypted Base64 encoded data to trytes and assign it to the ProposedTransaction object’s message argument.

An address is also needed, so we generate one with the help of get_new_addresses() extended API method. Feel free to choose the index of the generated address, and don’t forget, that the method returns a dict with a list of addresses, even if it contains only one. For more detailed explanation on how addresses are generated in PyOTA, refer to the adresses:Generating Addresses page.

We also attach a custom Tag to our ProposedTransaction. Note, that if our trytes_encrypted_data was longer than the maximum payload of a transaction, the library would split it accross more transactions that together form the transfer bundle.

60
61
62
63
64
65
66
print('Sending transfer...')
# Send the transaction to the network
response = api.send_transfer([tx])

print('Check your transaction on the Tangle!')
print('https://utils.iota.org/transaction/%s/devnet' % response['bundle'][0].hash)
print('Tail transaction hash of the bundle is: %s' % response['bundle'].tail_transaction.hash)

Finally, we use Iota.send_transfer() to prepare the transfer and send it to the network.

Click on the link to check your transaction on the Tangle Explorer.

The tail transaction (a tail transaction is the one with index 0 in the bundle) hash is printed on the console, because you will need it in the next tutorial, and anyway, it is a good practice to keep a reference to your transfers.

In the next example, we will try to decode the confidential information from the Tangle.

7. Fetch Encrypted Data

In this example, you will learn how to:

  • Fetch bundles from the Tangle based on their tail transaction hashes.

  • Extract messages from a bundle.

  • Decrypt encrypted messages from a bundle.

Warning

We will use the simple-crypt external library for encryption/decryption. Before proceeding to the tutorial, make sure you install it by running:

pip install simple-crypt

Code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
"""
Decrypt data fetched from the Tangle.

simplecrypt library is needed for this example (`pip install simple-crypt`)!
"""
from iota import Iota
from simplecrypt import decrypt
from base64 import b64decode
from getpass import getpass

import json

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443', devnet=True)

# Prompt user for tail tx hash of the bundle
tail_hash = input('Tail transaction hash of the bundle: ')

print('Looking for bundle on the Tangle...')
# Fetch bundle
bundle = api.get_bundles([tail_hash])['bundles'][0]

print('Extracting data from bundle...')
# Get all messages from the bundle and concatenate them
b64_encrypted_data = "".join(bundle.get_messages())

# Decode from base64
encrypted_data = b64decode(b64_encrypted_data)

# Prompt for passwword
password = getpass('Password to be used for decryption:')

print('Decrypting data...')
# Decrypt data
# decrypt returns 'bytes' in Python 3, decode it into string
json_data = decrypt(password, encrypted_data).decode('utf-8')

# Convert JSON string to python dict object
data = json.loads(json_data)

print('Succesfully decrypted the following data:')
print(data)

Discussion

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
"""
Decrypt data fetched from the Tangle.

simplecrypt library is needed for this example (`pip install simple-crypt`)!
"""
from iota import Iota
from simplecrypt import decrypt
from base64 import b64decode
from getpass import getpass

import json

# Declare an API object
api = Iota('https://nodes.devnet.iota.org:443', devnet=True)

In contrast to 6. Store Encrypted Data where we intended to encrypt data, in this tutorial we will do the reverse, and decrypt data from the Tangle. Therefore, we need the decrypt method from simplecrypt library and the b64decode method from base64 library.

Furthermore, getpass is needed to prompt the user for a decryption password, and json for deserializing JSON formatted string into Python object.

16
17
# Prompt user for tail tx hash of the bundle
tail_hash = input('Tail transaction hash of the bundle: ')

To fetch transactions or bundles from the Tangle, a reference is required to retreive them from the network. Transactions are identified by their transaction hash, while a group of transaction (a bundle) by bundle hash. Hashes ensure the integrity of the Tangle, since they contain verifiable information about the content of the transfer objects.

input() asks the user to give the tail transaction hash of the bundle that holds the encrypted messages. The tail transaction is the first in the bundle with index 0. Copy and paste the tail transaction hash from the console output of 6. Store Encrypted Data when prompted.

19
20
21
print('Looking for bundle on the Tangle...')
# Fetch bundle
bundle = api.get_bundles([tail_hash])['bundles'][0]

Next, we fetch the bundle from the Tangle with the help of the get_bundles() extended API command. It takes a list of tail transaction hashes and returns the bundles for each of them. The response dict contains a bundles key with the value being a list of bundles in the same order as the input argument hashes. Also note, that the bundles in the response are actual PyOTA Bundle objects.

To simplify the code, several operations are happening on line 21:

  • Calling get_bundles() that returns a dict,

  • accessing the 'bundles' key in the dict,

  • and taking the first element of the the list of bundles in the value associated with the key.

23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
print('Extracting data from bundle...')
# Get all messages from the bundle and concatenate them
b64_encrypted_data = "".join(bundle.get_messages())

# Decode from base64
encrypted_data = b64decode(b64_encrypted_data)

# Prompt for passwword
password = getpass('Password to be used for decryption:')

print('Decrypting data...')
# Decrypt data
# decrypt returns 'bytes' in Python 3, decode it into string
json_data = decrypt(password, encrypted_data).decode('utf-8')

# Convert JSON string to python dict object
data = json.loads(json_data)

The next step is to extract the content of the message fields of the transactions in the bundle. We call Bundle.get_messages() to carry out this operation. The method returns a list of unicode strings, essentially the signature_message_fragment fields of the transactions, decoded from trytes into unicode characters.

We then combine these message chunks into one stream of characters by using string.join().

We know that at this stage that we can’t make sense of our message, because it is encrypted and encoded into Base64. Let’s peel that onion layer by layer:

  • On line 28, we decode the message into bytes with b64decode.

  • On line 31, we ask the user for thr decryption password (from the previous tutorial).

  • On line 36, we decrypt the bytes cipher with the password and decode the result into a unicode string.

  • Since we used JSON formatting in the previous tutorial, there is one additional step to arrive at our original data. On line 39, we deserialize the JSON string into a Python object, namely a dict.

41
42
print('Succesfully decrypted the following data:')
print(data)

If everything went according to plan and the user supplied the right password, we should see our original data printed out to the console.

Now you know how to use the Tangle for data storage while keeping privacy. When you need more granular access control on how and when one could read data from the Tangle, consider using Masked Authenticated Messaging (MAM).

8. Send and Monitor Concurrently

In this example, you will learn how to:

  • Use the asynchronous PyOTA API.

  • Send transactions concurrently.

  • Monitor confirmation of transactions concurrently.

  • Execute arbitrary code concurrently while doing the former two.

Warning

If you are new to coroutines and asynchronous programming in Python, it is strongly recommended that you check out this article and the official asyncio documentation before proceeding.

Code

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from iota import AsyncIota, ProposedTransaction, Address, TryteString
from typing import List
import asyncio

# Asynchronous API instance.
api = AsyncIota(
        adapter='https://nodes.devnet.iota.org:443',
        devnet=True,
)

# An arbitrary address to send zero-value transactions to.
addy = Address('PZITJTHCIIANKQWEBWXUPHWPWVNBKW9GMNALMGGSIAUOYCKNWDLUUIGAVMJYCHZXHUBRIVPLFZHUVDLME')

# Timeout after which confirmation monitoring stops (seconds).
timeout = 120
# How often should we poll for confirmation? (seconds)
polling_interval = 5


async def send_and_monitor(
    transactions: List[ProposedTransaction]
) -> bool:
    """
    Send a list of transactions as a bundle and monitor their confirmation
    by the network.
    """
    print('Sending bundle...')
    st_response = await api.send_transfer(transactions)

    sent_tx_hashes = [tx.hash for tx in st_response['bundle']]

    print('Sent bundle with transactions: ')
    print(*sent_tx_hashes, sep='\n')

    # Measure elapsed time
    elapsed = 0

    print('Checking confirmation...')
    while len(sent_tx_hashes) > 0:
        # Determine if transactions are confirmed
        gis_response = await api.get_inclusion_states(sent_tx_hashes, None)

        for i, (tx, is_confirmed) in enumerate(zip(sent_tx_hashes, gis_response['states'])):
            if is_confirmed:
                print('Transaction %s is confirmed.' % tx)
                # No need to check for this any more
                del sent_tx_hashes[i]
                del gis_response['states'][i]

        if len(sent_tx_hashes) > 0:
            if timeout <= elapsed:
                # timeout reached, terminate checking
                return False
            # Show some progress on the screen
            print('.')
            # Put on hold for polling_interval. Non-blocking, so you can
            # do other stuff in the meantime.
            await asyncio.sleep(polling_interval)
            elapsed = elapsed + polling_interval

    # All transactions in the bundle are confirmed
    return True


async def do_something() -> None:
    """
    While waiting for confirmation, you can execute arbitrary code here.
    """
    for _ in range(5):
        print('Doing something in the meantime...')
        await asyncio.sleep(2)


async def main() -> None:
    """
    A simple application that sends zero-value transactions to the Tangle and
    monitors the confirmation by the network. While waiting for the
    confirmation, we schedule a task (`do_something()`) to be executed concurrently.
    """
    # Transactions to be sent.
    transactions = [
        ProposedTransaction(
            address=addy,
            value=0,
            message=TryteString.from_unicode('First'),
        ),
        ProposedTransaction(
            address=addy,
            value=0,
            message=TryteString.from_unicode('Second'),
        ),
        ProposedTransaction(
            address=addy,
            value=0,
            message=TryteString.from_unicode('Third'),
        ),
    ]

    # Schedule coroutines as tasks, wait for them to finish and gather their
    # results.
    result = await asyncio.gather(
            send_and_monitor(transactions),
            # Send the same content. Bundle will be different!
            send_and_monitor(transactions),
            do_something(),
        )

    if not (result[0] and result[1]):
        print('Transactions did not confirm after %s seconds!' % timeout)
    else:
        print('All transactions are confirmed!')

if __name__ == '__main__':
    # Execute main() inside an event loop if the file is ran
    asyncio.run(main())

Discussion

This example is divided into 4 logical parts:

  1. Imports and constant declarations

  2. Coroutine to send and monitor a list of transactions as a bundle.

  3. Coroutine to execute arbitrary code concurrently.

  4. A main coroutine to schedule the execution of our application.

Let’s start with the most simple one: Imports and Constants.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from iota import AsyncIota, ProposedTransaction, Address, TryteString
from typing import List
import asyncio

# Asynchronous API instance.
api = AsyncIota(
        adapter='https://nodes.devnet.iota.org:443',
        devnet=True,
)

# An arbitrary address to send zero-value transactions to.
addy = Address('PZITJTHCIIANKQWEBWXUPHWPWVNBKW9GMNALMGGSIAUOYCKNWDLUUIGAVMJYCHZXHUBRIVPLFZHUVDLME')

# Timeout after which confirmation monitoring stops (seconds).
timeout = 120
# How often should we poll for confirmation? (seconds)
polling_interval = 5

Notice, that we import the AsyncIota api class, because we would like to use the asynchronous and concurrent features of PyOTA. List from the typing library is needed for correct type annotations, and we also import the asyncio library. This will come in handy when we want to schedule and run the coroutines.

On line 6, we instantiate an asynchronous IOTA api. Functionally, it does the same operations as Iota, but the api calls are defined as coroutines. For this tutorial, we connect to a devnet node, and explicitly tell this as well to the api on line 8.

On line 12, we declare an IOTA address. We will send our zero value transactions to this address. Feel free to change it to your own address.

Once we have sent the transactions, we start monitoring their confirmation by the network. Confirmation time depends on current network activity, the referenced tips, etc., therefore we set a timeout of 120 seconds on line 15. You might have to modify this value later to see the confirmation of your transactions.

You can also fine-tune the example code by tinkering with polling_interval. This is the interval between two subsequent confirmation checks.

Let’s move on to the next block, namely the send and monitor coroutine.

20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
async def send_and_monitor(
    transactions: List[ProposedTransaction]
) -> bool:
    """
    Send a list of transactions as a bundle and monitor their confirmation
    by the network.
    """
    print('Sending bundle...')
    st_response = await api.send_transfer(transactions)

    sent_tx_hashes = [tx.hash for tx in st_response['bundle']]

    print('Sent bundle with transactions: ')
    print(*sent_tx_hashes, sep='\n')

    # Measure elapsed time
    elapsed = 0

    print('Checking confirmation...')
    while len(sent_tx_hashes) > 0:
        # Determine if transactions are confirmed
        gis_response = await api.get_inclusion_states(sent_tx_hashes, None)

        for i, (tx, is_confirmed) in enumerate(zip(sent_tx_hashes, gis_response['states'])):
            if is_confirmed:
                print('Transaction %s is confirmed.' % tx)
                # No need to check for this any more
                del sent_tx_hashes[i]
                del gis_response['states'][i]

        if len(sent_tx_hashes) > 0:
            if timeout <= elapsed:
                # timeout reached, terminate checking
                return False
            # Show some progress on the screen
            print('.')
            # Put on hold for polling_interval. Non-blocking, so you can
            # do other stuff in the meantime.
            await asyncio.sleep(polling_interval)
            elapsed = elapsed + polling_interval

    # All transactions in the bundle are confirmed
    return True

Notice, that coroutines are defined in python by the async def keywords. This makes them awaitable.

From the type annotations, we see that send_and_monitor() accepts a list of ProposedTransaction objects and return a bool.

On line 28, we send the transfers with the help of AsyncIota.send_transfer(). Since this is not a regular method, but a coroutine, we have to await its result. AsyncIota.send_transfer() takes care of building the bundle, doing proof-of-work and sending the transactions within the bundle to the network.

Once we sent the transfer, we collect individual transaction hashes from the bundle, which we will use for confirmation checking.

On line 39, the so-called confirmation checking starts. With the help of AsyncIota.get_inclusion_states(), we determine if our transactions have been confirmed by the network.

Note

You might wonder how your transactions get accepted by the network, that is, how they become confirmed.

The None value for the tips parameter in the argument list basically means that we check against the latest milestone.

On line 43, we iterate over our original sent_tx_hashes list of sent transaction hashes and gis_response['states'], which is a list of bool values, at the same time using the built-in zip method. We also employ enumerate, because we need the index of the elements in each iteration.

If a transaction is confirmed, we delete the corresponding elements from the lists. When all transactions are confirmed, sent_tx_hashes becomes empty, and the loop condition becomes False.

If however, not all transactions have been confirmed, we should continue checking for confirmation. Observe line 58, where we suspend the coroutine with asyncio.sleep() for polling_interval seconds. Awaiting the result of asyncio.sleep() will cause our coroutine to continue execution in polling_interval time. While our coroutine is sleeping, other coroutines can run concurrently, hence it is a non-blocking call.

To do something in the meantime, we can execute another coroutine concurrently:

65
66
67
68
69
70
71
async def do_something() -> None:
    """
    While waiting for confirmation, you can execute arbitrary code here.
    """
    for _ in range(5):
        print('Doing something in the meantime...')
        await asyncio.sleep(2)

This is really just a dummy coroutine that prints something to the terminal and then goes to sleep periodically, but in a real application, you could do meaningful tasks here.

Now let’s look at how to schedule the execution of our application with the main coroutine:

 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def main() -> None:
    """
    A simple application that sends zero-value transactions to the Tangle and
    monitors the confirmation by the network. While waiting for the
    confirmation, we schedule a task (`do_something()`) to be executed concurrently.
    """
    # Transactions to be sent.
    transactions = [
        ProposedTransaction(
            address=addy,
            value=0,
            message=TryteString.from_unicode('First'),
        ),
        ProposedTransaction(
            address=addy,
            value=0,
            message=TryteString.from_unicode('Second'),
        ),
        ProposedTransaction(
            address=addy,
            value=0,
            message=TryteString.from_unicode('Third'),
        ),
    ]

    # Schedule coroutines as tasks, wait for them to finish and gather their
    # results.
    result = await asyncio.gather(
            send_and_monitor(transactions),
            # Send the same content. Bundle will be different!
            send_and_monitor(transactions),
            do_something(),
        )

    if not (result[0] and result[1]):
        print('Transactions did not confirm after %s seconds!' % timeout)
    else:
        print('All transactions are confirmed!')

if __name__ == '__main__':
    # Execute main() inside an event loop if the file is ran
    asyncio.run(main())

First, we declare a list of ProposedTransaction() objects, that will be the input for our send_and_monitor() coroutine.

The important stuff begins on line 101. We use asyncio.gather() to submit our coroutines for execution, wait for their results and then return them in a list. gather takes our coroutines, transforms them into runnable tasks, and runs them concurrently.

Notice, that we listed send_and_monitor() twice in asyncio.gather() with the same list of ProposedTransaction() objects. This is to showcase how you can send and monitor multiple transfers concurrently. In this example, two different bundles will be created from the same ProposedTransaction() objects. The two bundles post zero value transactions to the same address, contain the same messages respectively, but are not dependent on each other in any way. That is why we can send them concurrently.

As discussed previously, result will be a list of results of the coroutines submitted to asyncio.gather(), preserving their order. result[0] is the result from the first send_and_monitor(), and result[1] is the result from the second send_and_monitor() from the argument list. If any of these are False, confirmation did not happen before timeout.

When you see the message from line 109 in your terminal, try increasing timeout, or check the status of the network, maybe there is a temporary downtime on the devnet due to maintenance.

Lastly, observe lines 113-115. If the current file (python module) is run from the terminal, we use ayncio.run() to execute the main coroutine inside an event loop.

To run this example, navigate to examples/tutorial inside the cloned PyOTA repository, or download the source file of Tutorial 8 from GitHub and run the following in a terminal:

$ python 08_async_send_monitor.py