Respetar los límites de velocidad de la API con un backoff

Este artículo se actualizó en abril de 2025

Al trabajar con las API de comunicación de Vonage-o cualquier API en realidad-debes tener en cuenta sus límites de tarifas. Los límites de velocidad son una de las formas en que los proveedores de servicios pueden reducir la carga en sus servidores, evitar actividades maliciosas o garantizar que un solo usuario no monopolice los recursos disponibles.

En este artículo, veremos cómo puedes administrar mejor tus llamadas a la API para asegurarte de ser un "buen ciudadano de la API". Veremos cómo puedes respetar los límites de tarifa de Vonage Communication API de Vonage Communication y al mismo tiempo ser eficiente y completar tus llamadas a la API lo más rápido posible.

Vonage API Account

To complete this tutorial, you will need a Vonage API account. If you don’t have one already, you can sign up today and start building with free credit. Once you have an account, you can find your API Key and API Secret at the top of the Vonage API Dashboard.

This tutorial also uses a virtual phone number. To purchase one, go to Numbers > Buy Numbers and search for one that meets your needs.

Ready to start building?

Experience seamless connectivity, real-time messaging, and crystal-clear voice and video calls—all at your fingertips.

¿Qué significa ser un buen ciudadano de la API?

Cuando trabajamos con APIs externas, siempre debemos intentar mantener nuestro rendimiento a un nivel aceptable. Pero los errores ocurren, puede haber un aumento repentino en el uso, y terminamos excediendo el límite de velocidad causando que nuestra llamada a la API falle.

Cuando esto ocurre, puede ser tentador volver a intentarlo inmediatamente, pero hacerlo es contraproducente. Si tus llamadas a la API fallan porque has alcanzado el límite de velocidad, el servicio te está diciendo que reduzcas la velocidad. Volver a intentar inmediatamente la misma solicitud no es ir más despacio y puede hacer que le prohíban el acceso a algunos servicios. En su lugar, debe "retroceder" y hacer una pausa antes de volver a intentarlo.

Retraso de llamadas a la API con Backoff

El tiempo de espera es el que se espera antes de realizar una acción. La cantidad de tiempo de espera puede calcularse utilizando muchas estrategias diferentes, pero algunas de las más comunes son:

• Constante: espera un tiempo constante entre cada intento. Por ejemplo, si tenemos un retardo constante de 1 segundo, nuestros intentos ocurrirán en 1s, 2s, 3s, 4s, 5s, 6s, 7s, etc.

• Fibonaccial: aquí utilizamos el número Fibonacci correspondiente al intento actual, haciendo que nuestros retrasos sean 1s, 1s, 2s, 3s, 5s, 8s, 13s, etc.

• Exponencial: el retraso se calcula como 2 a la potencia del número de intentos fallidos que se hayan realizado. Por ejemplo:

• 2^1 = 2 = 2

• 2^2 = 2 * 2 = 4

• 2^3 = 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

Existen otras estrategias -fija, lineal, polinómica-, pero para este artículo vamos a ceñirnos a la estrategia de backoff exponencial que proporciona el paquete paquete backoff de Python.

Probando el Backoff

No quiero activar el límite de tarifa de la API de Vonage sólo para demostrar el paquete Backoff. En su lugar, vamos a crear un código simulado con 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()

En este ejemplo, la función slow_operation() registra los milisegundos transcurridos desde la época y luego devuelve Falseasegurando que el decorador backoff se ejecuta cada vez que llamamos a la función. Backoff seguirá ejecutándose slow_operation() hasta que el retardo alcance max_time de 300 segundos, momento en el que se rendirá.

Para generar muchos puntos de datos para el gráfico, ponemos en cola la función slow_operation() 500 veces dentro de nuestro bucle asyncio.

Si graficamos el número de llamadas a funciones intentadas por segundo, esto es lo que parece cuando utilizamos una estrategia constante:

A constant traffic pattern visualised

Hay una banda gruesa en el rango de 40 a 60 llamadas de función, por lo que un backoff constante no es apropiado para nuestras necesidades. Cada segundo, estamos inundando la API con peticiones, manteniendo nuestro rendimiento demasiado alto, y es probable que sigamos teniendo una tasa limitada.

Pero si ejecutamos el mismo código con la estrategia exponencial, obtenemos un gráfico muy diferente.

A visualisation of a traffic pattern using exponential backoff

Este gráfico es mucho mejor. Podemos ver donde la estrategia backoff ha aumentado el retraso, la reducción de la producción y es de esperar que nos da tiempo suficiente para poner fin a la tasa de limitación. Pero ahora tenemos otro problema.

En el gráfico, podemos ver que las llamadas se agrupan al final de los retrasos. Podríamos llegar a una situación en la que estos grupos activaran de nuevo la limitación de velocidad. Para evitar que se formen estos grupos, utilizamos el jitter.

Crear una carga de trabajo más equitativamente distribuida con aleatoriedad

El jitter añade un factor aleatorio al cálculo de la duración del retardo en nuestro backoff.

sleep = random.uniform(0, delay)

El paquete El paquete backoff de Python incluye este jitter por defecto. En los ejemplos de código anteriores lo estoy eliminando con una función lambda, así que vamos a generar la gráfica exponencial de nuevo, pero esta vez con jitter.

An exponential backoff traffic pattern visualised

Como seguimos utilizando una estrategia exponencial, podemos ver que el número de llamadas disminuye muy rápidamente, pero gracias a la aleatoriedad añadida del jitter, no vemos ningún amontonamiento. En su lugar, las llamadas a funciones por segundo son bajas y se distribuyen de forma más uniforme.

Procesamiento de colas de SMS con backoff

Los límites de tarifas varían según la API de Vonage Communications que estés usando. Por ejemplo, las Redact API y Application API tienen un límite de velocidad de 170 solicitudes por segundo.. Sin embargo, debido a las restricciones del operador, el límite de velocidad para SMS salientes puede ser de sólo una solicitud por segundo.. Esto convierte a los SMS en el candidato perfecto para aplicar las técnicas de backoff que hemos visto antes.

Colas de tareas e intermediarios

Python dispone de un gran número de colas de tareas entre las que elegir.Apio, [huey²], RQ, Kuyruk, Taskmaster, Dramatiq, WorQ-y casi otros tantos brokers-MongoDB, Redis, RabbitMQ, SQS. Algunas de estas colas de tareas vienen con soporte para backoff incorporado, pero también añaden mucha complejidad, haciéndolas fuera del alcance de este artículo.

Sin embargo, una vez que te sientas cómodo con las técnicas subyacentes y el razonamiento detrás del uso de una cola de tareas, backoff, jitter, etc., entonces te recomiendo que vuelvas a visitar los enlaces a las colas de tareas anteriores. Los ejemplos de código que vamos a ver en el resto de este artículo son intencionalmente sucintos para que podamos centrarnos sólo en la gestión del rendimiento, mientras que los paquetes anteriores son mucho más robustos y listos para la producción.

Envío asincrónico de SMS con las API de comunicación de Vonage

Para garantizar que la latencia de la red no bloquee toda la aplicación, vamos a enviar los SMS de forma asíncrona. Pero esto significa que que no podemos utilizar el Vonage Python SDK para enviar nuestros SMS. El SDK de Python no es asíncrono, ya que utiliza Requests, que es bloqueante.

Podemos consultar la solicitud de ejemplo de Messages API de la documentación para tener una idea de lo que el SDK Python de Vonage hace por nosotros:

En este fragmento de código, podemos ver que estamos emitiendo una solicitud POST al endpoint https://api.nexmo.com/v0.1/message. La petición incluye alguna información sobre el tipo de contenido que estamos enviando y esperando como respuesta. Pero las partes esenciales a tener en cuenta son la cabecera Authorization y la opción data (-d).

La solicitud se autoriza utilizando Tokens Web JSON (JWT). JWT es un estándar abierto de la industria, y hay varios paquetes de Python disponibles para ayudar con su generación. Pero, afortunadamente, el SDK de Python de Vonage ya tiene una función que podemos llamar para crear un JWT válido para la solicitud. Como la generación de JWT es rápida, no requiere ninguna E/S de red, y sólo se realiza una vez al inicio del script, no importa que no sea asíncrona.

Crear su aplicación

Instala la CLI de Vonage globalmente con este comando:

npm install @vonage/cli -g

Luego, configura la CLI con tu clave y secreto de API de Vonage. Puedes encontrar esta información en el Panel del desarrollador.

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

Cree un nuevo directorio para su proyecto y CD en él:

mkdir my_project
CD my_project

Ahora, utiliza la CLI para crear una aplicación de 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 almacenará su clave privada en el archivo mi_proyecto.key. Necesitaremos esto cuando generemos nuestro JWT, junto con el id de la aplicación. El identificador de la aplicación aparece en la terminal cuando ejecutas el comando app:create o puedes encontrarlo en tu panel de Vonage.

Ahora necesitas un número para poder recibir llamadas. Puedes alquilar uno utilizando el siguiente comando (sustituyendo el código del país por tu código). Por ejemplo, si estás en EE.UU., sustituye GB por US:

Ahora vincula el número a tu aplicación:

vonage apps:link --number=VONAGE_NUMBER APP_ID

Envío del 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

En la parte superior de nuestro script, fuera de la función async, instanciamos nuestro cliente de Vonage con el id de aplicación y la clave privada. He almacenado estas en variables de entorno, por lo que no están codificadas dentro de mi script.

La función send_sms realiza la petición a la API, por lo que esta función tiene nuestro decorador backoff. Estamos usando on_predicatepor lo que si la función devuelve Falselo intentará de nuevo. He mantenido el max_time en 300 segundos, pero también podríamos establecer un límite de max_attempt o ambos.

Para realizar la petición POST asíncrona, utilizamos httpx. httpx es un cliente HTTP para Python 3 con una interfaz muy similar a la de Requests, pero soporta asíncrono. Estructuré la petición httpx lo más parecido posible al ejemplo cURL que vimos antes. Tenemos las cabeceras con la información del tipo de contenido así como la cabecera de Autorización, que incluye el JWT generado para nosotros por el SDK Python de Vonage.

Nuestra carga útil es una cadena JSON que contiene el número del remitente, el número del destinatario y nuestro mensaje.

Por último, comprobamos el código de estado HTTP devuelto por la Messages API a la solicitud. Cualquier cosa que no sea un 202 Aceptado hará que la función devuelva False, desencadenando otro intento.

Cola de SMS en bucle

En mi script de ejemplo, acabo de codificar una lista de destinatarios.

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

Pero aquí es donde se podría utilizar una cola de tareas o un broker. Además, ¡no estoy siendo un buen ciudadano de la API! Sé que la API de mensajes tiene un límite de velocidad de 1 mensaje por segundo cuando se envían mensajes dentro de EE.UU., ¡pero no tengo ningún retraso en mi bucle!

Mi script intentará realizar las llamadas a la API sin retardo entre ellas, activando muy rápidamente la limitación de velocidad. Aunque el backoff nos ayuda a gestionar cuando excedemos el rendimiento máximo permitido, debería ser el último recurso. Idealmente, para ser más eficientes, queremos acercarnos al límite de velocidad, pero sin excederlo. La adición de un breve reposo cuando se añaden tareas al bucle debería ayudar con esto.

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

Puesta en común

En esta grabación, he eliminado la suspensión y he modificado el ejemplo para que intente realizar varios cientos de solicitudes a la vez, lo que provoca que Vonage provoque el estrangulamiento casi instantáneamente. Pero observa lo que sucede después de unos segundos.

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

Casi tan pronto como el script comienza, vemos que excede el límite de velocidad de Messages API, y el endpoint comienza a devolver un estado HTTP de 429 "Demasiadas Peticiones". Entonces, el script comienza a retroceder. Al principio, el número de peticiones fallidas parece permanecer más o menos igual, pero a medida que el retraso aumenta exponencialmente, el número de peticiones fallidas cae en unos pocos segundos, y nuestro script puede empezar a enviar de nuevo.

Pruébelo usted mismo

Sin carga de producción, puede ser bastante difícil generar suficientes peticiones para activar la limitación de velocidad. Puedes consultar el script de ejemplo de este tutorial, así como las instrucciones de uso en GitHub.

Ten en cuenta que el envío de mensajes supondrá un cargo en tu Account. Si envías rutinariamente suficientes mensajes como para limitar tu tarifa, podrías violar los términos de servicio de Vonage; además, ¡las compañías de telefonía no verán con buenos ojos que envíes el mismo mensaje cientos de veces! Por lo tanto, te recomiendo que si quieres probar esto por ti mismo, no lo pruebes con el Messages API en vivo sino que simularlo. Hay varios paquetes para httpx para hacer este proceso más fácil, incluyendo pytest-httpx y respx.

¿Y ahora qué?

Sólo hemos visto algunas de las funcionalidades disponibles con Python backoff. Consulta la documentación para más información sobre el suministro de diferentes estrategias de backoff para diferentes tipos de excepcioneso los eventos que emite el backoff. Intenta modificar el código de ejemplo para que si backoff ejecuta el manejador on_giveup el script utilizará la Voice API de Vonage para llamar al ingeniero de guardia.

Lecturas complementarias

Full Stack Python - Colas de tareas Blog de arquitectura de AWS - Exponential Backoff y Jitter