Respeite os limites de taxa da API com um mecanismo de recuo

Este artigo foi atualizado em abril de 2025

Ao trabalhar com as APIs de comunicação da Vonage—ou qualquer API, na verdade—você deve estar ciente dos limites de taxa. Os limites de taxa são uma das formas pelas quais os provedores de serviço podem reduzir a carga em seus servidores, impedir atividades maliciosas ou garantir que um único usuário não monopolize os recursos disponíveis.

Neste artigo, veremos como você pode gerenciar da melhor forma suas chamadas de API para garantir que seja um “bom usuário de API”. Veremos como você pode respeitar os limites de taxa da API de Comunicação da Vonage limites de taxa e, ao mesmo tempo, ser eficiente e concluir suas chamadas de API o mais rápido possível, dentro do permitido.

O que significa ser um bom usuário de API?

Ao trabalhar com APIs externas, devemos sempre procurar manter nossa taxa de processamento em um nível aceitável. Mas erros acontecem; pode ocorrer um aumento repentino no uso, e acabamos excedendo o limite de taxa, fazendo com que nossa chamada à API falhe.

Quando isso acontece, pode ser tentador tentar novamente imediatamente, mas fazer isso é contraproducente. Se suas chamadas de API estão falhando porque você atingiu o limite de taxa, isso significa que o serviço está pedindo para você diminuir o ritmo. Tentar a mesma solicitação novamente de imediato não significa diminuir o ritmo e pode fazer com que você seja banido de alguns serviços. Em vez disso, você deve “dar um tempo” e fazer uma pausa antes de tentar novamente.

Atrasando chamadas de API com backoff

Um backoff é quando você espera antes de realizar uma ação. O tempo de espera pode ser calculado por meio de diversas estratégias, mas algumas das mais comuns são:

• Constante: aguardar um intervalo de tempo constante entre cada tentativa. Por exemplo, se tivermos um atraso constante de 1 segundo, nossas tentativas ocorrerão aos 1s, 2s, 3s, 4s, 5s, 6s, 7s, etc.

• Fibonacciano: aqui usamos o número de Fibonacci correspondente à tentativa atual, definindo nossos intervalos como 1s, 1s, 2s, 3s, 5s, 8s, 13s, etc.

• Exponencial: o atraso é calculado como 2 elevado ao número de tentativas malsucedidas realizadas. Por exemplo:

• 2^1 = 2 = 2

• 2² = 2 × 2 = 4

• 2³ = 2 2 2 = 8

• 2^4 = 2 2 2 * 2 = 16

• 2^5 = 2 2 2 2 2 = 32

• 2^6 = 2 2 2 2 2 * 2 = 64

• 2^7 = 2 2 2 2 2 2 2 = 128

Existem outras estratégias — fixa, linear, polinomial —, mas, para os fins deste artigo, vamos nos ater à estratégia de backoff exponencial fornecida pelo pacote backoff do Python.

Testando o Backoff

Não quero atingir o limite de chamadas da API da Vonage só para demonstrar o pacote Backoff. Em vez disso, vamos criar um código simulado com asyncio.

import asyncio
from datetime import datetime
 
import backoff
import uvloop
 
start_time = datetime.now().timestamp()
 
 
@backoff.on_predicate(backoff.constant, max_time=300, jitter=lambda x: x)
async def slow_operation():
   with open("./attempts.log", "a") as f:
       f.write(f"{datetime.now().timestamp() - start_time}\n")
   return False
 
 
async def main(loop):
   for x in range(0, 500):
       asyncio.ensure_future(slow_operation(), loop=loop)
 
 
if __name__ == "__main__":
   loop = uvloop.new_event_loop()
   loop.create_task(main(loop))
   loop.run_forever()

Neste exemplo, a slow_operation() função registra os milissegundos desde a época e, em seguida, retorna False, garantindo que o decorador backoff seja executado sempre que chamarmos a função. O backoff continuará sendo executado slow_operation() até que o atraso atinja o max_time de 300 segundos; nesse momento, ele desistirá.

Para gerar muitos pontos de dados para o gráfico, colocamos a slow_operation() função 500 vezes dentro do nosso loop asyncio.

Se representarmos graficamente o número de chamadas de função tentadas por segundo, eis como fica quando usamos uma estratégia constante:

A constant traffic pattern visualised

Há um pico intenso na faixa de 40 a 60 chamadas de função, portanto, um backoff constante não é adequado para nossas necessidades. A cada segundo, estamos sobrecarregando a API com solicitações, mantendo nossa taxa de processamento alta demais, e é provável que continuemos sofrendo limitação de taxa.

Mas, se executarmos o mesmo código com a estratégia exponencial, obtemos um gráfico bem diferente.

A visualisation of a traffic pattern using exponential backoff

Este gráfico está bem melhor. Podemos ver onde a estratégia de backoff aumentou o atraso, reduzindo a taxa de transferência e, esperamos, nos dando tempo suficiente para encerrar a limitação de taxa. Mas agora temos outro problema.

No gráfico, podemos ver que as chamadas agora estão se agrupando próximo ao fim dos atrasos. Poderíamos chegar a uma situação em que esses agrupamentos continuassem acionando o limitador de taxa novamente. Para impedir que esses agrupamentos se formem, usamos o jitter.

Criando uma carga de trabalho distribuída de forma mais equitativa por meio da aleatoriedade

O jitter adiciona um fator aleatório ao cálculo da duração do atraso em nosso backoff.

sleep = random.uniform(0, delay)

O pacote backoff do Python inclui esse jitter por padrão. Nos exemplos de código acima, estou removendo-o com uma função lambda; portanto, vamos gerar o gráfico exponencial novamente, mas desta vez com jitter.

An exponential backoff traffic pattern visualised

Como ainda estamos usando uma estratégia exponencial, podemos observar que o número de chamadas diminui muito rapidamente, mas, graças à aleatoriedade adicional proporcionada pelo jitter, não vemos nenhum aglomerado. Em vez disso, as chamadas de função por segundo são baixas e estão distribuídas de maneira mais uniforme.

Processamento de filas de SMS com backoff

Os limites de taxa variam dependendo da API da Vonage Communications que você estiver utilizando. Por exemplo, as APIs Redact e Applications têm um limite de taxa de 170 solicitações por segundo. No entanto, devido a restrições das operadoras, o limite de taxa para SMS de saída pode chegar a apenas uma solicitação por segundo. Isso torna o SMS o candidato perfeito para a aplicação das técnicas de backoff que analisamos acima.

Filas de tarefas e corretores

O Python oferece uma grande variedade de filas de tarefas à escolha —Celery, [huey²], RQ, Kuyruk, Taskmaster, Dramatiq, WorQ—e quase tantos corretores—MongoDB, Redis, RabbitMQ, SQS. Algumas dessas filas de tarefas já vêm com suporte integrado ao backoff, mas também aumentam bastante a complexidade, o que as coloca fora do escopo deste artigo.

No entanto, assim que você se sentir à vontade com as técnicas subjacentes e o raciocínio por trás do uso de uma fila de tarefas, backoff, jitter e assim por diante, recomendo que você revisite os links sobre filas de tarefas mencionados acima. Os exemplos de código que veremos no restante deste artigo são intencionalmente sucintos para que possamos nos concentrar apenas no gerenciamento de throughput; já os pacotes mencionados acima são muito mais robustos e prontos para produção.

Envio assíncrono de SMS com as APIs de comunicação da Vonage

Para garantir que a latência da rede em qualquer solicitação não bloqueie todo o nosso aplicativo, vamos enviar nossas mensagens SMS de forma assíncrona. No entanto, isso significa não podemos usar o SDK do Python da Vonage para enviar nossas mensagens SMS. O SDK para Python não é assíncrono, pois utiliza o Requests, que é bloqueante.

Podemos dar uma olhada na exemplo de solicitação da Messages API da documentação para ter uma ideia do que o SDK Python da Vonage está fazendo por nós:

Neste trecho de código, podemos ver que estamos enviando uma solicitação POST para o endpoint https://api.nexmo.com/v0.1/message. A solicitação inclui algumas informações sobre o tipo de conteúdo que estamos enviando e esperando como resposta. Mas os elementos essenciais a serem observados são o cabeçalho Authorization e a opção de dados (-d).

A solicitação é autorizada por meio de JSON Web Tokens (JWT). O JWT é um padrão aberto do setor, e há vários pacotes em Python disponíveis para auxiliar na sua geração. Mas, convenientemente, o SDK do Python da Vonage já possui uma função que podemos chamar para criar um JWT válido para a solicitação. Como a geração do JWT é rápida, não requer nenhuma E/S de rede e é realizada apenas uma vez no início do script, não importa que ela não seja assíncrona.

Criando seu aplicativo

Instale a CLI do Vonage globalmente com este comando:

npm install @vonage/cli -g

Em seguida, configure a CLI com sua chave e seu segredo da API da Vonage. Você pode encontrar essas informações no Painel do Desenvolvedor.

vonage config:set --apiKey=VONAGE_API_KEY --apiSecret=VONAGE_API_SECRET

Crie um novo diretório para o seu projeto e acesse-o com o comando `cd`:

mkdir my_project
CD my_project

Agora, use a CLI para criar um aplicativo da Vonage.

vonage apps:create
✔ Application Name … my_project
✔ Select App Capabilities › Messages
✔ Create messages webhooks? … no
✔ Allow use of data for AI training? no

Este comando armazenará sua chave privada no arquivo my_project.key. Precisaremos dela ao gerar nosso JWT, juntamente com o ID do aplicativo. O ID do aplicativo é exibido no terminal quando você executa o app:create comando, ou você pode encontrá-lo no seu painel da Vonage.

Agora você precisa de um número para poder receber chamadas. Você pode alugar um usando o comando a seguir (substituindo o código do país pelo seu). Por exemplo, se você estiver nos EUA, substitua GB por US:

Agora, vincule o número ao seu aplicativo:

vonage apps:link --number=VONAGE_NUMBER APP_ID

Envio da mensagem SMS

import vonage


vonage_client = vonage.Client(
   application_id=os.environ["VONAGE_APPLICATION_ID"],
   private_key=os.environ["VONAGE_PRIVATE_KEY"],
)
jwt = vonage_client.generate_application_jwt()
 
@backoff.on_predicate(backoff.expo, max_time=300)
async def send_sms(recipient, message):
   async with httpx.AsyncClient() as httpx_client:
       response = await httpx_client.post(
           "https://api.nexmo.com/v0.1/messages",
           headers={
               "Authorization": b"Bearer " + jwt,
               "Content-Type": "application/json",
               "Accept": "application/json",
           },
           json={
               "from": {"type": "sms", "number": os.environ["VONAGE_NUMBER"]},
               "to": {"type": "sms", "number": recipient},
               "message": {"content": {"type": "text", "text": message,}},
           },
       )
 
        return response.status_code == 202

No início do nosso script, fora da função assíncrona, instanciamos nosso cliente Vonage com o ID do aplicativo e a chave privada. Armazenei esses dados em variáveis de ambiente, para que não fiquem codificados diretamente no meu script.

A send_sms função faz a solicitação à API; por isso, essa função possui nosso decorador de backoff. Estamos usando on_predicate, portanto, se a função retornar False, ela tentará novamente. Mantive o max_time em 300 segundos, mas também poderíamos definir um max_attempt limite ou ambos!

Para fazer a solicitação POST assíncrona, usamos o httpx. O httpx é um cliente HTTP para Python 3 com uma interface muito semelhante à do Requests, mas que oferece suporte à assíncronia. Estruturei a solicitação do httpx da forma mais próxima possível do exemplo do cURL que vimos acima. Temos os cabeçalhos com as informações do tipo de conteúdo, bem como o cabeçalho Authorization, que inclui o JWT gerado para nós pelo SDK do Python da Vonage.

Nossa carga útil é uma string JSON que contém o número do remetente, o número do destinatário e nossa mensagem.

Por fim, verificamos o código de status HTTP retornado pela Messages API em resposta à solicitação. Qualquer código que não seja 202 Accepted fará com que a função retorne False, acionando uma nova tentativa.

Envio de SMS em loop

No meu script de exemplo, acabei de definir uma lista de destinatários de forma estática.

async def main(loop):
   recipients = [
       "13055550157",
       "15615550134",
   ]
   message = "✨✨✨Hello! This is an SMS from the Vonage Communication APIs Messages API using exponential backoff and jitter 😄"
 
   for recipient in recipients:
       asyncio.ensure_future(send_sms(recipient, message), loop=loop)
 
 
if __name__ == "__main__":
   loop = uvloop.new_event_loop()
   loop.create_task(main(loop))
   loop.run_forever()

Mas é aí que você poderia usar uma fila de tarefas ou um broker. Além disso, também não estou seguindo as boas práticas de uso de APIs! Sei que a Messages API tem um limite de taxa de 1 mensagem por segundo ao enviar mensagens dentro dos EUA, mas não tenho nenhum atraso no meu loop!

Meu script tentará fazer as chamadas à API sem nenhum intervalo entre elas, o que acionará a limitação de taxa muito rapidamente. Embora o backoff nos ajude a gerenciar a situação quando excedemos a taxa máxima permitida, ele deve ser um último recurso. Idealmente, para sermos mais eficientes, queremos chegar o mais próximo possível do limite de taxa, mas sem ultrapassá-lo. A inclusão de uma breve pausa ao adicionar tarefas ao loop deve ajudar nisso.

for recipient in recipients:
       asyncio.ensure_future(send_sms(recipient, message), loop=loop)
       await asyncio.sleep(1)

Juntando tudo isso

Nesta gravação, removi o intervalo de espera e modifiquei o exemplo para que ele tente fazer várias centenas de solicitações de uma só vez, fazendo com que a limitação da Vonage seja acionada quase que instantaneamente. Mas veja o que acontece depois de alguns segundos.

Traffic to an API being backed off over time until blocked requests end

Quase assim que o script começa, vemos que ele excede o limite de taxa da Messages API, e o endpoint passa a retornar um status HTTP 429 “Too Many Requests”. Assim, o script começa a reduzir a frequência. A princípio, o número de solicitações com falha parece permanecer praticamente o mesmo, mas, à medida que o atraso aumenta exponencialmente, o número de solicitações com falha diminui em poucos segundos, e nosso script pode começar a enviar solicitações novamente.

Experimente você mesmo

Sem carga de produção, pode ser bastante complicado gerar solicitações suficientes para acionar a limitação de taxa. Você pode conferir o script de exemplo deste tutorial, bem como as instruções de uso, no GitHub.

Observe que o envio de mensagens acarretará cobrança em sua Account. Se você enviar mensagens com frequência suficiente a ponto de ser limitado pela taxa de envio, poderá violar os termos de serviço da Vonage; além disso, as operadoras não verão com bons olhos o fato de você enviar a mesma mensagem centenas de vezes! Portanto, recomendo que, se quiser testar isso por conta própria, não faça o teste na Messages API, mas sim faça um teste simulado. Existem vários pacotes para o httpx que facilitam esse processo, incluindo o pytest-httpx e respx.

E agora?

Analisamos apenas algumas das funcionalidades disponíveis no backoff do Python. Consulte a documentação para obter mais informações sobre como definir diferentes estratégias de backoff para diferentes tipos de exceções, ou sobre os vários eventos que o backoff emite. Tente modificar o código de exemplo para que, caso o backoff execute o on_giveup handler, o script utilize a Voice API da Vonage para ligar para o engenheiro de plantão.

Leitura complementar

Python Full Stack - Filas de tarefas Blog de Arquitetura da AWS - Backoff exponencial e jitter