As APIs são a espinha dorsal das aplicações modernas, permitindo a comunicação eficiente entre clientes e servidores. No entanto, uma API sem proteção contra requisições excessivas pode ser vítimas de ataques, levando à degradação do desempenho, aumento dos custos de infraestrutura ou até mesmo à indisponibilidade do serviço.

Por que o Rate Limiting é importante?

O Rate Limiting é um mecanismo essencial para garantir a estabilidade, segurança e uso justo de uma API. Ele ajuda a:

• Prevenir abusos – Usuários mal-intencionados ou bots podem sobrecarregar uma API com requisições, tentando explorar vulnerabilidades ou derrubar o serviço.
• Reduzir custos de infraestrutura – Um tráfego descontrolado pode aumentar os custos com servidores e impactar a escalabilidade da aplicação.
• Garantir uso justo – As limitações evitam que um único cliente monopolize os recursos da API, distribuindo melhor a carga.
• Mitigar ataques DDoS – Ao restringir o número de requisições por usuário ou IP, a API fica mais resistente a ataques de negação de serviço.

Onde o Rate Limiting é utilizado?

O Rate Limiting é amplamente aplicado em diversos setores, como:

• Redes Sociais – Para evitar spam e abuso de bots (exemplo: o Twitter limita a quantidade de tweets postados por hora).
• E-commerce – Para impedir que robôs sobrecarreguem páginas de produtos ou ferramentas de rastreamento de preços.
• Serviços Financeiros – Para proteger APIs bancárias contra transações automatizadas excessivas.
• Aplicações SaaS – Para diferenciar planos gratuitos e pagos, limitando o número de requisições por usuário.

Hands on!

Implementando Rate Limiting no FastAPI com Redis

Uma abordagem comum para implementar rate limiting é utilizar o Redis como um armazenamento em memória para rastrear requisições por usuário, token ou IP. O FastAPI permite a fácil integração de middlewares para aplicar essas restrições de forma eficiente. A ideia é interceptar a request antes que ela passe para sua funcionalidade principal, o Middleware vai verificar se há espaço para uma nova requisição, se tiver ela permite seguir, se não retorna imediatamente um Status Code 429 “Too Many Requests”.

Supondo que já possua o FastApi up & running:

pip install redis

Vamos criar a classe RateLimiter:

Nesse caso optei por colocar um limite de 10 requests a cada 1 minuto. No final mostrarei os resultados…

class RateLimiter:
    """
    Rate Limiter para APIs.
    Limita o número de requisições por usuário em um determinado intervalo de tempo
    redis_client: Cliente Redis para armazenar os dados de rate limiting
    max_requests: Número máximo de requisições permitidas
    window: Intervalo de tempo em segundos
    """
    def __init__(self, redis_client, max_requests=10, window=60):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window = window
    def allow_request(self, user_id) -> bool:
        """Verifica se a requisição pode ser processada dentro do limite."""
        now = time.time()
        key = f"rate_limit:{user_id}"
        pipe = self.redis.pipeline()
        pipe.zadd(key, {now: now})
        pipe.zremrangebyscore(key, 0, now - self.window)
        pipe.zcard(key)
        pipe.expire(key, self.window * 2)
        _, _, request_count, _ = pipe.execute()
        result = request_count <= self.max_requests
        return result
class RateLimiter:
    """
    Rate Limiter para APIs.
    Limita o número de requisições por usuário em um determinado intervalo de tempo
    redis_client: Cliente Redis para armazenar os dados de rate limiting
    max_requests: Número máximo de requisições permitidas
    window: Intervalo de tempo em segundos
    """

    def __init__(self, redis_client, max_requests=10, window=60):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window = window

    def allow_request(self, user_id) -> bool:
        """Verifica se a requisição pode ser processada dentro do limite."""
        now = time.time()
        key = f"rate_limit:{user_id}"

        pipe = self.redis.pipeline()
        pipe.zadd(key, {now: now})
        pipe.zremrangebyscore(key, 0, now - self.window)
        pipe.zcard(key)
        pipe.expire(key, self.window * 2)

        _, _, request_count, _ = pipe.execute()

        result = request_count <= self.max_requests

        return result
class RateLimiter:
    """
    Rate Limiter para APIs.
    Limita o número de requisições por usuário em um determinado intervalo de tempo
    redis_client: Cliente Redis para armazenar os dados de rate limiting
    max_requests: Número máximo de requisições permitidas
    window: Intervalo de tempo em segundos
    """

    def __init__(self, redis_client, max_requests=10, window=60):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window = window

    def allow_request(self, user_id) -> bool:
        """Verifica se a requisição pode ser processada dentro do limite."""
        now = time.time()
        key = f"rate_limit:{user_id}"

        pipe = self.redis.pipeline()
        pipe.zadd(key, {now: now})
        pipe.zremrangebyscore(key, 0, now - self.window)
        pipe.zcard(key)
        pipe.expire(key, self.window * 2)

        _, _, request_count, _ = pipe.execute()

        result = request_count <= self.max_requests

        return result

Enter fullscreen mode Exit fullscreen mode

Configurando a aplicação (normalmente no seu main.py ou app.py):

redis_client = redis.Redis(host='redis', port=6379, db=0)
redis_client = redis.Redis(host='redis', port=6379, db=0)
redis_client = redis.Redis(host='redis', port=6379, db=0)

Enter fullscreen mode Exit fullscreen mode

Criando o Middleware

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Middleware de rate limiting para todas as rotas"""
    user_id = request.headers.get('X-User-ID', 'anonymous')
    if not limiter.allow_request(user_id):
        return JSONResponse(
            status_code=429,
            content={"message": "Limite de requisições excedido"}
        )
    response = await call_next(request)
    return response
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Middleware de rate limiting para todas as rotas"""
    user_id = request.headers.get('X-User-ID', 'anonymous')

    if not limiter.allow_request(user_id):
        return JSONResponse(
            status_code=429,
            content={"message": "Limite de requisições excedido"}
        )

    response = await call_next(request)
    return response
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Middleware de rate limiting para todas as rotas"""
    user_id = request.headers.get('X-User-ID', 'anonymous')

    if not limiter.allow_request(user_id):
        return JSONResponse(
            status_code=429,
            content={"message": "Limite de requisições excedido"}
        )

    response = await call_next(request)
    return response

Enter fullscreen mode Exit fullscreen mode

Para testar, basta configurar o Insomnia, Postman ou qualquer ferramenta que possa simular cargas!

No nosso caso aqui, utilizei o Insomnia! Segue as evidências:

O código fonte está no seguinte repo. Podem utilizar para estudos e afins! Há um uso do Locust como teste de carga dentro da pasta load_test/

Conclusão

O Rate Limiting é uma estratégia fundamental para proteger APIs e garantir um uso equilibrado e manter a performance. Integrando Redis com FastAPI, os desenvolvedores podem aplicar essas restrições de forma eficiente, tornando suas APIs mais robustas e escaláveis.

Links úteis e inspirações:

• https://redis.io/glossary/rate-limiting/
• https://medium.com/@m-elbably/rate-limiting-a-dynamic-distributed-rate-limiting-with-redis-339f9504200f
• https://www.youtube.com/watch?v=YV4ePyW3DO8

原文链接：Como Limitar e Proteger suas APIs com Rate Limit?