This tutorial demonstrates how to use the Python credit-rate-limit library to manage API rate limits based on credits or compute unit per time unit, or cups, for asynchronous requests, illustrated with a working example.

Introduction

In my previous tutorial, I introduced the Python credit-rate-limit library for handling APIs with rate limits based on a number of requests per time unit, like Etherscan’s 5 requests per second.

However, some API providers use credit-based rate limiting, where each request consumes a variable number of “credits” (e.g., endpoint A: 15 credits, endpoint B: 20 credits) within a fixed allowance per time unit (e.g., 50 credits per second).

As an example, we'll write async requests to a Web3 RPC provider (namely Infura), but that could be any other API.

What will do the example script?

For the sake of this tutorial, we want to call 2 endpoints that cost a different number of credits, with async requests.
Let's choose:

  • eth_getBlockByNumber: this endpoint will provide the block for a given block number. From this block we'll take the timestamp.
  • eth_getUncleCountByBlockNumber: this endpoint will provide the number of uncles for a given block number.

So, the script is going to build a list of timestamped counts of uncle blocks, by scheduling on the event loop 100 requests to each of these endpoints.

Uncles made sense when Ethereum was PoW, so we'll look for blocks that are before The Merge.

What do we need ?

The libraries

  • web3.py: the well known Python library for web3. Feel free to use any other library to build your request (aiohttp, httpx, ...) that suits your use case.
  • 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 web3 credit-rate-limit

The Infura API

Requesting our 2 endpoints will cost the following number of credits:

  • eth_getBlockByNumber costs 80 credits per call
  • eth_getUncleCountByBlockNumber costs 150 credits per call

The Infura credit allowance for the free plan is 500 credits per second.

And you will need a free API key.

Ready to code ?

First, let's write the API requests:

import os
from web3 import AsyncWeb3

aw3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(os.environ["WEB3_HTTP_PROVIDER_URL_ETHEREUM_MAINNET"]))

async def get_timestamp(block_number):
    block = await aw3.eth.get_block(block_identifier=block_number)
    counter()
    return block["timestamp"]

async def get_uncle_count(block_number):
    uncle_count = await aw3.eth.get_uncle_count(block_number)
    counter()
    return uncle_count

I added a counter() method to count the number of calls and report it on the terminal while the script is executing.
You can see its definition in the full script at the end of the tutorial.

We gather them so both endpoints are called for a given block number:

import asyncio

async def get_timestamp_and_uncle_count(block_number):
    return await asyncio.gather(get_timestamp(block_number), get_uncle_count(block_number))

And now the main() function where we schedule all 200 requests on the event loop at once:

start_block = 10000000
scanned_blocks = 100

async def main():
    coros = []
    for block_number in range(start_block, start_block + scanned_blocks):
        coros.append(get_timestamp_and_uncle_count(block_number))

    timestamps_and_uncle_counts = await asyncio.gather(*coros)
    # print(timestamps_and_uncle_counts)
    print(f"Got {len(timestamps_and_uncle_counts)} timestamped uncle counts")

And see the full script at the end of the tutorial for the call to this main() function with the execution duration.

First run

❗Boom❗The script crashed! 'Too Many Requests'

aiohttp.client_exceptions.ClientResponseError: 429, message='Too Many Requests'

And that's normal since we have not managed the rate limit yet!

Apply the rate limit to our code

So, let's do it!

First, we define the credit allowance per time unit, which is 500 credits per second:

from credit_rate_limit import CreditRateLimiter, throughput
rate_limiter = CreditRateLimiter(max_credits=500, interval=1)
  • interval is in seconds
  • max_credits is the number of credits you can use during interval seconds.

Then, thanks to the throughput decorator, we decorate each request with the CreditRateLimiter instance we've just defined and the cost associated to the concerned endpoint:

@throughput(rate_limiter=rate_limiter, request_credits=80)
async def get_timestamp(block_number):
    block = await aw3.eth.get_block(block_identifier=block_number)
    counter()
    return block["timestamp"]


@throughput(rate_limiter=rate_limiter, request_credits=150)
async def get_uncle_count(block_number):
    uncle_count = await aw3.eth.get_uncle_count(block_number)
    counter()
    return uncle_count
  • request_credits: the cost in credits of the decorated request
  • rate_limiter: the CreditRateLimiter we wish to use. We could define several of them if we had many APIs and/or providers.

Ready to re-try?

Second run

✅ Success ✅

=> Requests: 200/200 - 100%
Got 100 timestamped uncle counts
Duration: 69.9773 seconds

Bonus tip: Optimizing Your Rate Limiter for Better Performance!

CreditRateLimiter 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.

It also means you can easily make your requests work first, and then and only if you wish, you can optimize them.

For this script I could use adjustment=0.9 without being rate limited:

rate_limiter = CreditRateLimiter(max_credits=500, interval=1, adjustment=0.9)

For a massive performance improvement: the script duration decreased by more than 50%!

=> Requests: 200/200 - 100%
Got 100 timestamped uncle counts
Duration: 32.1815 seconds

Conclusion

The credit-rate-limit gets you covered as well for credit-based rate limit APIs. If you haven't done it yet, I'd recommend to check the tutorial on rate limits based on a number of requests per time unit.

That’s it, happy coding!! :)

Full Script: Putting It All Together for Seamless Execution 

import asyncio
import os
import time

from credit_rate_limit import CreditRateLimiter, throughput
from web3 import AsyncWeb3

aw3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(os.environ["WEB3_HTTP_PROVIDER_URL_ETHEREUM_MAINNET"]))

start_block = 10000000
scanned_blocks = 100

rate_limiter = CreditRateLimiter(max_credits=500, interval=1, adjustment=0.9)


def counter():
    counter.__dict__["value"] = counter.__dict__.get("value", 0) + 1
    end = "\n" if counter.__dict__["value"] == scanned_blocks * 2 else "\r"
    print(f"=> Requests: {counter.__dict__["value"]}/{scanned_blocks * 2}", f"- {counter.__dict__["value"] * 100 // (scanned_blocks * 2)}%", end=end)


@throughput(rate_limiter=rate_limiter, request_credits=80)
async def get_timestamp(block_number):
    block = await aw3.eth.get_block(block_identifier=block_number)
    counter()
    return block["timestamp"]


@throughput(rate_limiter=rate_limiter, request_credits=150)
async def get_uncle_count(block_number):
    uncle_count = await aw3.eth.get_uncle_count(block_number)
    counter()
    return uncle_count


async def get_timestamp_and_uncle_count(block_number):
    return await asyncio.gather(get_timestamp(block_number), get_uncle_count(block_number))


async def main():
    coros = []
    for block_number in range(start_block, start_block + scanned_blocks):
        coros.append(get_timestamp_and_uncle_count(block_number))

    timestamps_and_uncle_counts = await asyncio.gather(*coros)
    # print(timestamps_and_uncle_counts)
    print(f"Got {len(timestamps_and_uncle_counts)} timestamped uncle counts")


if __name__ == "__main__":
    start_time = time.perf_counter()
    asyncio.run(main())
    end_time = time.perf_counter()
    duration = end_time - start_time
    print(f"Duration: {duration:.4f} seconds")