Part 2: Streaming EOS Blocks With Python

in utopian-io •  4 months ago


This tutorial is part of a series where different aspects of programming with EOS are explained. This part will look at streaming blocks from the EOS Blockchain by using Python. The full curriculum can be found below.


What will I learn

  • Create an iterable block stream
  • Increase performance by using a session
  • Dealing with status codes


  • Python3.6


  • basic



Download the file from Github. There is 1 file Running the file without any arguments will initiate a performance test, it is also possible to run with 2 arguments start_block and block_count.

Run scripts as following:
> python
> python 1 100

Create an iterable block stream

The previous tutorial looked at how to retrieve EOS blocks by using API POST requests. Expanding on this code a basic version for streaming these block would be to build a loop. In order to have iterable output yield can be used.

def stream_blocks(start_block, block_count):
    for block_num in range(start_block, start_block + block_count):
            yield get_block(block_num)

This makes the function callable as a list which can be iterated, which in turn simplifies processing the blocks afterwards.

for block in stream_blocks(start_block, block_count):
    # perform action

Increase performance by using a session

The problem with looping standard POST requests is that the performance will suffer. For every request a new connection has to be established. Instead a Session can be created which is then used to handle all the requests.

import requests

s = requests.Session()
request =, data=parameters)

The session needs to be created outside of the loop.

s = requests.Session()

for block_num in range(start_block, start_block + block_count):
    yield get_block(block_num, s)

Dealing with status codes

The increased performance creates a new set of problems. Depending on which api_endpoint is being used there might be restrictions on how many requests can be made in a specific time frame. Which can be solved by adding a delay time.sleep() before each request.

EOS has a block_time of 500 ms or 2 blocks per second. This means that when the head_block is reached requests can be made for blocks that do not exist yet. Each request comes with a status_code, 200 means that everything is ok. There are several status_codes that can indicate different kind of problems. EOS nodes use the status_code 500 when a block that does not exist is requested.

    "code": 500,
    "message": "Internal Service Error",
    "error": {
        "code": 3100002,
        "name": "unknown_block_exception",
        "what": "unknown block",
        "details": [{
            "message": "Could not find block: 6000000",
            "file": "chain_plugin.cpp",
            "line_number": 832,
            "method": "get_block"

By creating a loop that only returns when the correct status_code the stream will halt until the issue has been dealt with.

while True:
    request =, data=parameters)
    if verify_request(request):
        return request.text

This allows for unique handling of each status_code.

def verify_request(request):
    if request.status_code == 200:
        return True
    elif request.status_code == 500:
        # Deal with the problem

This can be taken a step further by looking at the error codeinside the text part of the request.

    elif request.status_code == 500:
        error_code = json.loads(request.text)['error']['code']
        if error_code == 3100002:
            print('Waiting for blockchain to catch up\n')

Running the code

Running the code without any arguments will initiate the performance test. 20 blocks will be retrieved with and without using a session. The total time for requesting all blocks will be displayed.


Starting performance test without session

Block 1
Block 2
Block 20

Took 12.307921886444092 seconds for completion

Starting performance test with session

Block 1
Block 2
Block 20

Took 3.4480578899383545 seconds for completion

In addition running the code with the arguments start_block and block_count will stream the block_count amount of blocks from start_block.

python 1 10
Block 1
Block 2


Part 1: Using Python To Make API Requests To EOS Public API Endpoints

The code for this tutorial can be found on GitHub!

This tutorial was written by @juliank.

Authors get paid when people like you upvote their post.
If you enjoyed what you read here, create your account today and start earning FREE STEEM!
Sort Order:  

Thank you for your contribution @steempytutorials.

I liked your tutorial, but if the tutorial was more detailed I liked it more!

Your contribution has been evaluated according to Utopian policies and guidelines, as well as a predefined set of questions pertaining to the category.

To view those questions and the relevant answers related to your post, click here.

Need help? Write a ticket on
Chat with us on Discord.

Hey @steempytutorials
Thanks for contributing on Utopian.
We’re already looking forward to your next contribution!

Want to chat? Join us on Discord

Vote for Utopian Witness!