Forem: Elnaril

The 🐍 SDK for the Uniswap Universal Router 🦄 is part of Hacktoberfest 2025

Elnaril — Tue, 30 Sep 2025 14:56:52 +0000

As you all probably know, Hacktoberfest is an annual event which aims to promote open-source and having fun while growing skills, and takes place during all October.

This year, the Python Codec for the Uniswap Universal Router will participate. For that, I created and tagged several improvements/updates, for intermediate and advanced but also for beginner developers. Look for the hacktoberfest label.

To participate check the Hacktoberfest website.

Please, share it! :)

Python Tutorial: How to rate limit async requests to credit-based APIs

Elnaril — Sun, 20 Apr 2025 15:33:48 +0000

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")

What Uniswap Universal Router commands are actually used?

Elnaril — Wed, 20 Dec 2023 11:08:13 +0000

Context

There are more than 30 functions implemented in the Universal Router (UR) at the moment, in the repository main branch.

As the author of the Python Uniswap Universal Router SDK, I always wonder what should be the next one(s) this library should support.

Until now, I've added the most obvious functions:

the ones that allow to swap ERC-20 tokens:
- V3_SWAP_EXACT_IN
- V3_SWAP_EXACT_OUT
- V2_SWAP_EXACT_IN
- V2_SWAP_EXACT_OUT
those often needed before or after:
- WRAP_ETH
- UNWRAP_WETH
- PERMIT2_PERMIT
and a couple ones that allow more possibilities:
- SWEEP
- PAY_PORTION

But for the next supported functions, I decided to list those that are actually used, and order them from the most to the least called.

Methodology

To reach this goal, I wrote a simple Python script that looks into all blocks included in Mainnet during November 2023, gets all transactions sent to the Universal Router, discards those with a too small input field, and decodes them. And counts how many times each command (function) is used.

This script can be found here, so the results can be reproduced. Just change the RPC providers.

No fancy optimizations were implemented in the script apart from good ol' asyncio and it does not check whether the transactions were rejected by the EVM or not, to avoid requesting the receipts. This is a bit annoying in terms of clean data, but allowed me to get the results in less than half a day.

Another limitation with this method: it does not take into account transactions using the UR through custom contracts.

Results

The scripts scanned:

214 307 blocks (from 18 473 543 to 18 687 850)
2 453 214 transactions sent to the UR
5 435 322 commands

And the raw results is given in the below dict, with the decimal commands as keys and the number of calls as values:

Results: {11: 1362528, 8: 1886773, 9: 97389, 1: 52226, 12: 856319, 10: 593542, 0: 494207, 5: 8346, 4: 36076, 6: 47881, 16: 25, 25: 2, 14: 8}

There were 32 transactions that could not be decoded, either because they are wrongly formatted and rejected by the EVM, or their format does not respect the specifications (as I understand them) but still were accepted by the EVM (more on them in another post). Anyway they are not statistically relevant, so let's just disregard them for now.

The following table lists the results with the function names, ordered by counts (desc) and whether or not it is already supported by the codec:

Dec - Hex Code	Name	Counts	Supported
8 - 0x08	V2_SWAP_EXACT_IN	1886773	✅
11 - 0x0b	WRAP_ETH	1362528	✅
12 - 0x0c	UNWRAP_WETH	856319	✅
10 - 0x0a	PERMIT2_PERMIT	593542	✅
0 - 0x00	V3_SWAP_EXACT_IN	494207	✅
9 - 0x09	V2_SWAP_EXACT_OUT	97389	✅
1 - 0x01	V3_SWAP_EXACT_OUT	52226	✅
6 - 0x06	PAY_PORTION	47881	✅
4 - 0x04	SWEEP	36076	✅
5 - 0x05	TRANSFER	8346	❌
16 - 0x10	SEAPORT_V1_5	25	❌
14 - 0x0e	placeholder	8	N/A
25 - 0x19	SUDOSWAP	2	❌

As a Pie Chart:

So, basically, most of the used commands (99.8% of the calls) are already supported by the Python UR SDK. The only one really missing is TRANSFER with 8346 counts (< 0.2%), the others are anecdotal:

SEAPORT_V1_5 with 25 counts
a placeholder with no function yet (8 counts)
SUDOSWAP with 2 counts

Other interesting results

1/ The 4 swap functions (V2_SWAP_EXACT_IN, V3_SWAP_EXACT_IN, V2_SWAP_EXACT_OUT and V3_SWAP_EXACT_OUT) were called 2 530 595 times:

1 984 162 (78.4%) for a V2 pool, most of it (74.6%) through the V2_SWAP_EXACT_IN function
546 433 (21.6%) for a V3 pool

2/ The second most used function is WRAP_ETH with 1 362 528 calls, which is 53.8% of all swap commands. Knowing that several swap commands can be chained after a single WRAP_ETH, this suggests that ETH is largely used to initiate a swap.

3/ There are 2.2 commands/functions per transaction on average.

4/ Still on average, each block contains 11.4 transactions sent to the Universal Router!

Conclusion

As expected, the functions related to ERC-20 DeFi are extensively used. But, basically, nothing else is used apart the TRANSFER function: no batch ones, no permit2 transfers, no NFTs, ...
So, I guess I'll implement TRANSFER sooner or later, but what about the other ones?
Do you think there is an interest in them? If, yes which ones and why are they not already used?

Python SDK for the Uniswap Universal Router - Latest Version just Released

Elnaril — Tue, 04 Jul 2023 19:34:45 +0000

Decode and encode easily swap transactions to the Uniswap Universal Router.

The latest version (v0.9.0) of this open source library has just been released and provide support for more features.

Source on GitHub

Packaged on PyPI

What functions of the UR would you like to see in the next release?

✨✨ Tutorial: How to Buy a Token on the 🦄 Uniswap Universal Router 🦄 with 🐍 Python 🐍 | HackerNoon ✨✨

Elnaril — Tue, 06 Jun 2023 11:07:24 +0000

Creating a swap transaction for the latest Uniswap router is not as straightforward as it used to be with the previous routers: you cannot just use the router ABI and invoke a swap function.

This guide will teach you how to use Python to build the transaction input data you need to send to the Uniswap Universal Router (UR) in order to buy a token with ETH.

✨✨ Tutorial link ✨✨

I have developed some time ago an open source Python codec for the UR.Since then I have been asked to clarify some points about its usage. So I decided to write this tutorial.

I would love to hear any constructive feedback about both the Uniswap Universal Router codec and this tutorial, so feel free to comment about them!

Thanks in advance! :)