Tutorial: How to rate limit Python async requests to Etherscan - and other APIs

Introduction

As you know, API providers enforce rate limits to prevent their servers to be overwhelmed or to ensure a fair usage between users in respect of their plan. An application that relies heavily on an API is likely to reach its rate limit, thus a strategy to manage it must be implemented.

There are mainly 2 ways to handle rate limits on the client side:

• catch and retry the rejected requests

• manage the rate limit in the application

The first approach is not always a good one: the provider may impose penalties, such as incremental temporary bans or additional charges.

The second method is the most reliable but not always straightforward to implement, especially in an asynchronous code. And since we’re talking of applications heavy on the i/o side of the force, chances are the code is indeed asynchronous.

This article is a small tutorial on how to easily do this, with an example script that calls the Etherscan’s API but that could be any API that enforces a rate limit based on the number of requests per time unit.

What tools are we going to use ?

We’re going to use the 2 following open-source Python libraries:

• aiohttp: a Python async HTTP Client/Server. But that could be httpx of course.

• credit-rate-limit: the Python async Rate Limiter that’s going to make our dev life suddenly easier!

Disclaimer: I’m the author of the credit-rate-limit library. ;)

To install them in your virtual environment:

pip install aiohttp credit-rate-limit

Ok, but why choose credit-rate-limit ? Benefits Explained!

This library just works out of the box as you will see with the example. Other libs I tried, if just configured with the official API rate limit, fail miserably and the API returns an error for many of our requests. To make them work, you need to figure out a rate limit that would work.

With credit-rate-limit, you don’t waste your time to figure out a rate limit: you just configure the official one from the API you call! And only if you wish and think it’s worth it, you can optimize the rate limiter (see how at the end of tutorial).

But why is it called credit-rate-limit ? Because it can also easily be used for APIs that enforce rate limits based on credits per time unit (or CUPS, or request units, …) and not only the number of requests.

And about the API? What do we need to know?

• Etherscan API rate limit is 5 calls/s for the Free Tier plan.

• we’re going to use the “Normal Transactions By Address” endpoint.

• And you need a free API key

Nice! Now, let’s code!

First, let’s write the API request itself. For the purpose of this tutorial, we’re going to request the 10 000 first transactions between 2 given blocks for the address 0x3fC91A3afd70395Cd496C647d5a6CC9D4B2b7FAD, and return the transaction count to be sure we get all of them.

Why this particular address, you ask?

Simple: it’s the current Uniswap Universal Router address on Ethereum, and I also happen to be the author of the Python Uniswap Universal Router SDK! :)

import os

api_key = os.environ["ETHERSCAN_API_KEY"]
transaction_count = 10_000

async def get_tx_list(session, start_block, end_block):
    params = {
        "module": "account",
        "action": "txlist",
        "address": "0x3fC91A3afd70395Cd496C647d5a6CC9D4B2b7FAD",
        "startblock": start_block,
        "endblock": end_block,
        "page": 1,
        "offset": transaction_count,
        "sort": "asc",
        "apikey": api_key,
    }
    async with session.get(
            url="https://api.etherscan.io/api",
            params=params,
    ) as resp:
        resp_body = await resp.json()
        if resp_body["message"] == "NOTOK":
            raise ValueError(resp_body["result"])
        return len(resp_body["result"])  # trx_count

Ok, so, with that out of the way, straight to the point: how do we rate limit this request?

It’s as easy as:

from credit_rate_limit import CountRateLimiter, throughput

rate_limiter = CountRateLimiter(max_count=5, interval=1)

@throughput(rate_limiter=rate_limiter)
async def get_tx_list(session, start_block, end_block):
    ...

Let’s breakdown the important parts:

With rate_limiter = CountRateLimiter(max_count=5, interval=1), we define a rate limiter that will limit the number of requests to max_count per interval seconds. So, here 5 requests per second.

With the decorator @throughput(rate_limiter=rate_limiter), we apply the previously defined rate limiter to the decorated function.

And now, it’s time to finish our little script by calling our rate limited function 100 times.

import asyncio
from aiohttp import ClientSession

first_block = 20600000  # arbitrary 'recent' block
request_number = 100

async def run_throttled_api_request():
    async with ClientSession() as session:
        coros = []
        for i in range(request_number):
            block = first_block + 1000 * i
            coros.append(get_tx_list(session, block, block + 10000))  # request limited to 10 000 blocks
        results = await asyncio.gather(*coros)
        if all(map(lambda result: result == transaction_count, results)):
            print("SUCCESS !!")
        else:
            print("ERROR !!")


if __name__ == "__main__":
    asyncio.run(run_throttled_api_request())

100 requests will be scheduled on the asyncio loop and launched immediately.

Result:

SUCCESS !!

As you can see, rate limiting with credit-rate-limit is super easy!

You’ll find the full script at the end of the tutorial.

Bonus Tips: Optimizing Your Rate Limiter for Better Performance!

CountRateLimiter comes with a handy parameter, adjustment. It’s a float that can take any value between 0 (default) and interval. There’s no right value, you just need to try and see if the requests are rejected by the API or not. Depending on your use case, this parameter can greatly improve the performances.

For this particular request, I could set up this parameter to its max possible value without any problem and for a massive performance gain: adjustment=1

Conclusion

Implementing a robust rate limiting strategy is crucial for applications that rely heavily on APIs like Etherscan here. Using the library credit-rate-limit allows you, Pythonistas, to ensure your asynchronous Python applications run smoothly right of the box, without headache!

• Library code source and doc

• PyPI Package

That’s it, happy coding!! :)

Full Script: Putting It All Together for Seamless Execution

import asyncio
import os

from aiohttp import ClientSession

from credit_rate_limit import CountRateLimiter, throughput


api_key = os.environ["ETHERSCAN_API_KEY"]
first_block = 20600000
request_number = 10
transaction_count = 10_000

rate_limiter = CountRateLimiter(max_count=5, interval=1, adjustment=0)


@throughput(rate_limiter=rate_limiter)
async def get_tx_list(session, start_block, end_block):
    params = {
        "module": "account",
        "action": "txlist",
        "address": "0x3fC91A3afd70395Cd496C647d5a6CC9D4B2b7FAD",
        "startblock": start_block,
        "endblock": end_block,
        "page": 1,
        "offset": transaction_count,
        "sort": "asc",
        "apikey": api_key,
    }
    async with session.get(
            url="https://api.etherscan.io/api",
            params=params,
    ) as resp:
        resp_body = await resp.json()
        if resp_body["message"] == "NOTOK":
            raise ValueError(resp_body["result"])
        return len(resp_body["result"])  # trx_count


async def run_throttled_api_request():
    async with ClientSession() as session:
        coros = []
        for i in range(request_number):
            block = first_block + 1000 * i
            coros.append(get_tx_list(session, block, block + 10000))  # request limited to 10 000 blocks
        results = await asyncio.gather(*coros)
        # print(results)
        if all(map(lambda result: result == transaction_count, results)):
            print("SUCCESS !!")
        else:
            print("ERROR !!")


if __name__ == "__main__":
    asyncio.run(run_throttled_api_request())