Integración de la API de generación de vídeo con IA en Python: sondeo, webhooks y control de costes

Elegir el modo de obtención de resultados

La generación de vídeo es lenta en comparación con la mayoría de las llamadas a API. Una solicitud que genera unos segundos de vídeo puede tardar desde decenas de segundos hasta varios minutos, lo que significa que la respuesta HTTP de creación de la tarea casi nunca es la respuesta que contiene el vídeo. Antes de escribir una sola línea de código Python, decide cómo vas a obtener el resultado. Hay tres patrones prácticos.

Patrón 1: enviar y sondear

Envías una solicitud de generación, recibes inmediatamente un identificador de tarea y luego llamas al endpoint de estado a intervalos fijos hasta que la tarea alcanza un estado terminal. Este es el patrón más sencillo de implementar y depurar: es código de solicitud/respuesta ordinario, funciona detrás de cualquier firewall y no tiene requisitos de red entrante. Su desventaja es un límite inferior de latencia —la frescura del resultado depende únicamente de tu intervalo de sondeo— y desperdicia llamadas cuando se sondea más rápido de lo que avanza la tarea.

El sondeo encaja bien en cargas de trabajo interactivas: el usuario espera en la UI, se espera que la tarea termine en un tiempo acotado y quieres una única ruta de código sobre la que razonar.

Patrón 2: callback por webhook

Envías la tarea con una URL de callback, devuelves el control al llamador y dejas que la plataforma publique el resultado en tu endpoint cuando la tarea termine. Esto elimina por completo el sondeo y te da una notificación casi instantánea. También sigue funcionando después de que el proceso que envió la tarea se reinicie, porque la entrega la impulsa un agente externo.

El precio es operativo: necesitas un endpoint HTTPS accesible públicamente, necesitas verificar que el callback procede realmente del proveedor y necesitas manejar el caso en que el callback nunca llega —porque las redes fallan, los endpoints se configuran mal y las cargas útiles se pierden.

Los webhooks encajan en cargas de trabajo de larga duración o por lotes: colas grandes de tareas, canalizaciones asíncronas y cualquier escenario en el que mantener una conexión abierta o un bucle de espera activa durante minutos sea un desperdicio.

Patrón 3: híbrido con red de seguridad

El patrón que la mayoría de los sistemas en producción acaba adoptando: el webhook como señal principal, pero manteniendo una reconciliación periódica lenta que sondea las tareas que siguen marcadas como pendientes una vez superado un umbral de antigüedad. Si se pierde un callback, el sondeador lo captura. Si el sondeador se retrasa, normalmente el callback llega antes y marca la tarea como completada.

Elige según la carga de trabajo, no según el código base. Los endpoints interactivos pueden sondear; las colas en segundo plano pueden usar callbacks; el almacén de tareas compartido y la lógica de estado terminal deben ser idénticos para ambos, de modo que la estrategia de obtención pueda cambiar sin tocar el resto del sistema.

Configurar el cliente y enviar tareas

El esqueleto de envío debe mantenerse mínimo. Todo lo siguiente asume un único lugar donde guardar tu clave de API y la URL base, y una función que convierte un prompt más parámetros en un identificador de tarea.

Configuración y gestión de claves

Nunca codifiques las claves de forma rígida. Léelas de variables de entorno y falla de forma ruidosa al arrancar si faltan, para no descubrir el problema en la primera solicitud del cliente.

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # el KeyError al arrancar es intencionado

Mantén las claves fuera de los logs. Si registras solicitudes para depurar, registra el prompt y los parámetros, pero redacta la cabecera de autorización.

Una sola llamada de envío

La llamada de envío debe hacer solo una cosa: validar los parámetros que controlas, enviar la solicitud y devolver el identificador de la tarea. No inicies el sondeo dentro de ella ni descargues nada.

import requests

def submit_video_job(prompt: str, *, idempotency_key: str, **params) -> dict:
    payload = {"prompt": prompt, **params}
    resp = requests.post(
        f"{BASE_URL}/v1/video/generations",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
            "Idempotency-Key": idempotency_key,
        },
        json=payload,
        timeout=(5, 30),  # conexión, lectura
    )
    resp.raise_for_status()
    return resp.json()

La clave de idempotencia importa. Si tu llamada de envío agota el tiempo de espera, en realidad no sabes si la tarea se creó. Reintentar con la misma clave reenvía de forma segura, en lugar de generar —y pagar por— una tarea duplicada.

Qué persistir de inmediato

En el momento en que el envío devuelve, antes de hacer cualquier otra cosa, escribe una fila en tu propia base de datos:

• El ID de la tarea y la clave de idempotencia
• Tu propio ID de solicitud interno y el usuario o lote al que pertenece
• Los parámetros exactos de la solicitud, para que puedas reproducirla o auditarla
• El estado pending y una marca de tiempo

Si tu proceso se cae después del envío y antes de persistir, la tarea existe aguas arriba pero no hay registro local. Persistir primero y continuar después convierte tu base de datos en la fuente de verdad de lo que tienes que reconciliar.

Sondear resultados de la forma correcta

Los bucles de sondeo son fáciles de escribir mal. La versión ingenua —un while True sin pausas— consume cuota, machaca la API y hace que tus logs sean ilegibles. Un bucle correcto tiene cuatro propiedades: intervalo, retroceso, límite máximo y comprobación de estado terminal.

El bucle

import time, random

def poll_job(job_id: str, *, max_attempts: int = 60, base: float = 2.0, cap: float = 20.0):
    for attempt in range(1, max_attempts + 1):
        resp = requests.get(
            f"{BASE_URL}/v1/video/generations/{job_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=(5, 15),
        )
        resp.raise_for_status()
        body = resp.json()
        status = body["status"]

        if status == "succeeded":
            return body
        if status in {"failed", "cancelled"}:
            raise RuntimeError(f"La tarea {job_id} terminó con {status}: {body.get('error')}")

        delay = min(cap, base * (2 ** (attempt - 1)))
        time.sleep(delay + random.uniform(0, 1.0))

    raise TimeoutError(f"La tarea {job_id} sigue pendiente tras {max_attempts} intentos")

Por qué importa cada parte

Retroceso.Las tareas de generación avanzan por fases: en cola, en ejecución, renderizando. Sondear cada segundo durante la fase temprana de cola devuelve la misma respuesta veinte veces. Empezar con unos pocos segundos y duplicar hasta un límite hace que el volumen de llamadas sea proporcional a la duración de la tarea, no a tu paciencia.

Jitter.Añadir unas décimas de segundo aleatorias evita que todos los clientes que ejecutas sondeen al mismo tiempo tras un despliegue o una tormenta de reintentos.

Máximo de intentos y presupuesto de tiempo real.El retroceso exponencial sin límite de intentos puede ejecutarse durante horas. Elige un límite basado en la duración peor esperada de la tarea y trata superarlo como un fallo que debes exponer, no como un bucle que sigue ejecutándose en silencio. Lleva también la cuenta del tiempo transcurrido: una vez que se alcanza el límite del intervalo, el número de intentos por sí solo engaña.

Detección de estado terminal.Distingue explícitamente entre éxito, fallo y cancelación. El sondeador no debe reintentar una tarea fallida; debe devolverla con el informe de error del proveedor para que decidas si reenviarla con parámetros diferentes.

Una regla más: no sondees dentro del manejador de una solicitud web. Mueve el bucle a un trabajador en segundo plano para que las tareas lentas no ocupen tu pool de hilos de solicitudes.

Recibir resultados mediante callbacks de webhook

El endpoint de webhook es una ruta HTTP pública y sin autenticar por defecto a la que cualquiera puede hacer POST. Trátala como tal.

Verificar al remitente

Si la plataforma firma los callbacks, verifica la firma sobre el cuerpo sin procesar de la solicitud antes de analizar el JSON: la comprobación de la firma se ejecuta sobre bytes, y volver a serializar la carga útil puede cambiarlos. Compara con una función de tiempo constante y rechaza las discrepancias con un estado que no sea 2xx. Si no hay firma disponible, usa al menos un segmento de ruta secreto más una lista de permitidos o un token de cabecera compartido, y nunca confíes en el contenido de la carga útil antes de hacerla coincidir con una tarea que creaste.

Responder rápido, trabajar después

El único trabajo de tu endpoint es confirmar la recepción. Analiza, verifica, encola, responde. Todo lo lento —descargar el vídeo, tocar la base de datos bajo un lock, llamar a servicios posteriores— ocurre en el trabajador, no en el manejador. El proveedor reintenta los callbacks que no reciben una respuesta de éxito rápida, y un manejador lento convierte un evento en varios.

from flask import Flask, request, abort

app = Flask(__name__)

@app.post("/callbacks/video")
def video_callback():
    raw = request.get_data()
    if not verify_signature(raw, request.headers):
        abort(401)
    event = request.get_json(force=True)
    job_id = event.get("job_id")
    if not job_id:
        abort(400)
    queue.enqueue("process_video_event", event)  # devolver inmediatamente
    return {"ok": True}, 200

Conciliar con tus propios registros

Antes de actuar según un callback, busca el ID de la tarea en tu propia tabla. Esa búsqueda da tres casos:

• ID de tarea desconocido: puede ser un evento obsoleto de una tarea eliminada o una solicitud falsificada. Regístralo y descártalo. No crees registros a partir de la carga útil entrante.
• Tarea ya marcada como completada: es una entrega duplicada. Confírmala sin hacer nada más. Volver a descargar y volver a facturar la misma tarea es el defecto clásico de los webhooks.
• Tarea pendiente: márcala como completada (o fallida), guarda la referencia del recurso y pásala al paso de descarga.

Haz que las transiciones de estado dependan del estado actual —es decir, actualizaciones que solo tienen éxito si la tarea sigue pendiente— para que dos entregas concurrentes no puedan ganar ambas. Combínalo con el sondeador híbrido descrito antes y ninguna tarea se perderá aunque el callback nunca llegue.

Tiempos de espera, reintentos y gestión de fallos

Los reintentos son donde las integraciones se estropean en silencio. Las siguientes reglas se aplican al envío, al sondeo y a los callbacks, pero los límites de seguridad difieren.

Clasificar los errores primero

Transitorios: reinicios de conexión, respuestas 5xx, respuestas de limitación de tasa, tiempos de espera de lectura. Estos pueden tener éxito en el segundo intento.

Terminales: errores de validación 4xx, parámetros no admitidos, fallos de autenticación y fallos a nivel de tarea (como un prompt rechazado). Reintentar estos solo repite el mismo fallo y, en el caso de tareas ya enviadas, también el mismo coste.

En caso de duda, inspecciona el código de estado y el cuerpo del error en lugar de reintentar todo por defecto.

Retroceso con jitter y parada forzosa

Usa retroceso exponencial con jitter aleatorio para errores transitorios, y limita tanto los intentos como el tiempo total transcurrido. Una política de reintentos sin límite es un generador de fallos: cuando una dependencia se degrada, tu volumen de reintentos multiplica la carga justo cuando menos puede soportarla.

def with_retries(fn, *, attempts: int = 4, base: float = 0.5, cap: float = 8.0):
    for i in range(attempts):
        try:
            return fn()
        except TransientError:
            if i == attempts - 1:
                raise
            delay = min(cap, base * (2 ** i))
            time.sleep(delay + random.uniform(0, delay / 2))

Idempotencia en las fronteras

• Envío: envía siempre la misma clave de idempotencia al reintentar un envío. Es la única forma segura de recuperarse de un tiempo de espera en una solicitud que puede haber creado una tarea. Genera la clave una vez, guárdala junto con la solicitud prevista y reutilízala.
• Sondeo: idempotente por naturaleza: solo lectura. Puedes reintentar libremente dentro de tu presupuesto de tiempo.
• Callbacks: la entrega es esencialmente al menos una vez. Deduplica por ID de tarea más tipo de evento y haz que las transiciones de estado del manejador sean condicionales, para que las entregas repetidas no hagan nada.

Exponer el estado terminal

Cada ruta de tarea debe acabar almacenando exactamente un resultado: éxito con una referencia al recurso, o fallo con una causa y el historial de intentos. Nunca dejes una tarea en pending indefinidamente: tu sondeador híbrido existe precisamente para forzar una decisión tras el umbral de antigüedad. Expón ese estado a quien consume la tarea, para que el código aguas arriba pueda decidir reenviar, alertar o devolver la atención, en lugar de esperar para siempre.

Descargar y almacenar los vídeos terminados

La URL de la carga útil de finalización es temporal. Puede estar firmada, puede caducar y su host no es algo de lo que la lectura de tu aplicación deba depender. El último paso de la integración es mover el artefacto a un almacenamiento que controles.

Obtener el recurso

Descarga con una solicitud en streaming para que los archivos grandes no acaben en memoria como un único objeto de bytes, y establece tiempos de espera que reflejen el tamaño del archivo en lugar de una llamada típica a la API.

def download_asset(url: str, dest_path: str) -> None:
    with requests.get(url, stream=True, timeout=(5, 300)) as r:
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)

Si tu backend de almacenamiento admite escritura en streaming, pásale directamente el iterador de la respuesta en lugar de pasar por un archivo local temporal.

Verificar antes de confiar

Como mínimo, comprueba que el estado de la respuesta es de éxito, que la longitud del contenido coincide con cualquier tamaño declarado y que el tamaño del archivo no es trivial. Si la carga útil de finalización incluye metadatos de duración o formato, comprueba por muestreo que el artefacto descargado es probablemente un vídeo y no una página de error o un archivo parcial truncado. Un recurso corrupto que llega al usuario es peor que una tarea que marcas como fallida y reenvías.

Mover a tu propio almacenamiento

Escribe el objeto en tu propio bucket o origen de CDN, con una clave derivada de tu ID de tarea interno, y registra esa clave en la fila de la tarea. Sirve a los usuarios desde tus propias URLs. Los beneficios se acumulan:

• Las URLs temporales del proveedor nunca se filtran a tu frontend, tus correos o tus logs.
• La caducidad deja de importar; puedes regenerar derivados o recodificar según tu propio calendario.
• La retención, el control de acceso y la eliminación pasan a ser tu política, no la del proveedor.

Mantén el paso de descarga idempotente: si la clave de destino ya existe y es válida, omite la obtención. Con callbacks duplicados, esta comprobación es justo lo que evita una segunda descarga y una segunda escritura en el almacenamiento.

Controlar el coste de generación

La generación tiene un precio por el trabajo solicitado: normalmente se factura por la duración de salida o por unidades equivalentes a tokens, y la tarifa exacta aparece en el panel de precios de la plataforma. Como cada llamada es mucho más cara que una solicitud de texto típica, el control de costes debe estar en el punto de llamada, no en una revisión mensual.

Limitar la concurrencia

La protección individual más eficaz es limitar el número de tareas de generación concurrentes. Sin ella, una tormenta de reintentos, un bucle descontrolado o una ráfaga de solicitudes de usuarios pueden multiplicar el gasto más rápido de lo que cualquier panel puede mostrar. Coloca un semáforo o un pool de trabajadores acotado delante del envío y dimensiona su tamaño de forma deliberada.

from threading import Semaphore

_submit_gate = Semaphore(5)  # ajústalo a tu presupuesto y a los límites del proveedor

def guarded_submit(prompt, *, idempotency_key, **params):
    with _submit_gate:
        return submit_video_job(prompt, idempotency_key=idempotency_key, **params)

Validar antes de enviar

La mayoría de los sobregastos accidentales proceden de solicitudes que nunca deberían haberse enviado: un parámetro de duración mucho mayor de lo que la función necesita, un bucle de reintentos que reenvía una tarea que sigue en ejecución, un prompt vacío que dispara un defecto del cliente. Valida los parámetros —duración, resolución, longitud del prompt, referencias a medios de entrada— antes de que la solicitud salga de tu proceso, y recházala pronto con un error claro.

Registrar el volumen de tareas, no solo los errores

Registra cada envío junto con sus parámetros, la duración de la solicitud, el usuario y la clave de idempotencia. Cuenta las tareas por función y por usuario por periodo. El volumen agregado de tareas es la métrica que realmente sigue el gasto, y te permite detectar patrones anómalos —un llamador que genera diez veces la tasa normal— mientras aún son pequeños.

No pagar dos veces por la misma tarea

Hay tres fuentes comunes del modo de fallo de la generación duplicada, cada una con su defensa:

• Reintentar un envío que agotó el tiempo de espera sin clave de idempotencia. Reutiliza la clave.
• Una reentrega de webhook que dispara una nueva descarga o una nueva tarea. Deduplica por ID de tarea y haz condicionales las transiciones.
• Un sondeador que se rinde y reenvía mientras la tarea original sigue en ejecución. Verifica el estado de la tarea existente antes de crear una tarea sustituta y registra la relación de sustitución en la tabla de tareas.

Trata cada reenvío como una decisión que necesita una razón registrada, no como una ruta de recuperación por defecto.

Lista de comprobación antes del lanzamiento

Antes de que la integración salga a producción, verifica de extremo a extremo estas decisiones y modos de fallo.

Obtención y ciclo de vida

• El modo de obtención se elige según la carga de trabajo, con una red de seguridad híbrida para que un callback perdido no haga perder la tarea.
• La fila de la tarea se escribe tras el envío y antes de cualquier otro efecto secundario, registrando parámetros, clave de idempotencia y propietario.
• Cada tarea alcanza exactamente un estado terminal; el umbral de antigüedad fuerza una decisión sobre cualquier cosa que siga pendiente.
• Las transiciones de estado dependen del estado actual, de modo que las entregas repetidas y el sondeo concurrente son inofensivos.

Comportamiento de red

• Todas las solicitudes establecen tiempos de espera de conexión y lectura, incluidas las descargas.
• Los errores transitorios y terminales se clasifican, y solo se reintentan los transitorios.
• Los reintentos usan retroceso exponencial con jitter y un límite máximo tanto de intentos como de tiempo transcurrido.
• El retroceso del sondeo tiene tanto un límite de intervalo como un presupuesto de tiempo real, y el bucle de sondeo se ejecuta en un trabajador, no en el manejador de la solicitud.

Callbacks y almacenamiento

• El endpoint de callback verifica al remitente contra la experiencia bruta, responde rápido y pone el trabajo en cola.
• Cada callback se concilia con tus propios registros de tareas antes de emprender cualquier acción.
• Los recursos terminados se transmiten a tu propio almacenamiento, se verifican y se sirven solo desde tus URLs; el paso de descarga es idempotente.

Coste y operaciones

• El envío está condicionado por límites de concurrencia y por la validación de parámetros de duración, resolución y entradas.
• El volumen de tareas se registra por función y por usuario, de modo que el gasto sea visible como tendencia y no como sorpresa.
• Las rutas de generación duplicada —reintentos de envío, reentregas de webhook, reenvíos del sondeador— se gestionan explícitamente.
• Existen alertas para tareas que siguen pendientes más allá del umbral y para situaciones que alcanzan de forma sostenida el límite de concurrencia.

Ejecuta una tarea real por cada ruta —éxito, fallo, callback perdido, callback duplicado— y confirma que el estado terminal, las escrituras en el almacenamiento y los contadores son correctos. Si cada uno de los puntos anteriores se cumple en ese ejercicio, la integración está lista para salir a producción contra la API de generación de vídeo de HiAPI.ai.