Mastering Python Rate Limiting: From ratelimiter to SlowAPI

Understanding Access Rate Limiting

In web crawling and API design, limiting the number of requests per time unit is a common technique to prevent IP bans, captcha challenges, and server overload. This article focuses on the "access rate limit" method.

Using ratelimiter in Python

import time
from ratelimiter import RateLimiter

def limited(until):
    duration = int(round(until - time.time()))
    print('Rate limited, sleeping for {:d} seconds'.format(duration))

# 3 seconds can only be accessed 2 times
rate_limiter = RateLimiter(max_calls=2, period=3, callback=limited)

for i in range(3):
    with rate_limiter:
        print('Iteration', i)

The program prints two iterations, then triggers the callback, sleeps for three seconds, and prints the third iteration, confirming the rate limit works as expected.

Asyncio version

import asyncio
import time
from ratelimiter import RateLimiter

async def limited(until):
    duration = int(round(until - time.time()))
    print('Rate limited, sleeping for {:d} seconds'.format(duration))

async def coro():
    rate_limiter = RateLimiter(max_calls=2, period=3, callback=limited)
    for i in range(3):
        async with rate_limiter:
            print('Iteration', i)

loop = asyncio.get_event_loop()
loop.run_until_complete(coro())

The asynchronous example produces the same output, demonstrating that ratelimiter works seamlessly with async with blocks.

Introducing SlowAPI for FastAPI/Flask

For HTTP request rate limiting, the SlowAPI library offers a flexible and easy‑to‑use solution. It integrates smoothly with FastAPI and Flask, and similar extensions exist for Django.

# -*- coding: utf-8 -*-
# @Time : 2020/11/11 11:09
from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

# Initialize SlowAPI and register it with FastAPI
limiter = Limiter(key_func=get_remote_address)
FastAPI().state.limiter = limiter
FastAPI().add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

Usage example:

# -*- coding: utf-8 -*-
# @Time : 2020/11/11 11:09

# Allow only 4 calls per hour for this endpoint
@limiter.limit("4/hour")
async def startSpider(*, request: Request, d_obj: dict):
    if request.method == "POST":
        statusDict = get_spider_status(shopId=d_obj.get("shopId"))
        if not statusDict.get("data"):
            return resp_422(message='程序正在抓取中，请勿重复调度')
        try:
            result = all_run(d_obj.get("shopId"))
            return result
        except:
            return resp_401()
    return "这是一个GET请求"

This method limits an API to a configurable number of calls per hour, minute, or second, helping backend developers reduce server load and providing a simple protection mechanism for crawler engineers.