Integrando a API de geração de vídeo com IA em Python: polling, webhook e controle de custos

Escolhendo o modo de obtenção de resultados

Comparado à maioria das chamadas de API, a geração de vídeo é lenta. Uma requisição que gera alguns segundos de vídeo pode levar de dezenas de segundos a vários minutos, o que significa que a resposta HTTP de criação da tarefa quase nunca é a resposta que contém o vídeo. Antes de escrever uma linha de código Python, decida como você obterá o resultado. Existem três padrões práticos.

Padrão 1: enviar e fazer polling

Você envia uma requisição de geração, recebe imediatamente um identificador de tarefa e então chama o endpoint de status em intervalos fixos até que a tarefa atinja um estado terminal. Este é o padrão mais simples de implementar e depurar: é código comum de requisição/resposta, funciona atrás de qualquer firewall e não exige requisitos de rede de entrada. A desvantagem é um piso de latência — o frescor do resultado depende apenas do seu intervalo de polling — e desperdiça chamadas quando o polling é mais rápido que o avanço da tarefa.

O polling é adequado para cargas de trabalho interativas: o usuário espera em uma UI, espera-se que a tarefa termine em um tempo limitado e você quer um único caminho de código que possa raciocinar.

Padrão 2: callback via webhook

Você envia a tarefa com uma URL de callback anexada, devolve o controle ao chamador e deixa a plataforma fazer POST do resultado no seu endpoint quando a tarefa terminar. Isso elimina completamente o polling e fornece notificação quase instantânea. Também continua funcionando se o processo que enviou a tarefa reiniciar, porque a entrega é conduzida externamente.

O custo é operacional: você precisa de um endpoint HTTPS acessível publicamente, precisa verificar que o callback realmente veio do provedor e precisa lidar com o caso em que o callback nunca chega — porque redes falham, endpoints são configurados incorretamente e cargas úteis se perdem.

Webhooks são adequados para cargas de trabalho de longa duração ou em lote: grandes filas de tarefas, pipelines assíncronos e qualquer cenário em que manter uma conexão ou um loop ocioso por vários minutos seja desperdício.

Padrão 3: híbrido com fallback

O padrão que a maioria dos sistemas de produção acaba adotando: use o webhook como sinal principal, mas mantenha uma reconciliação periódica lenta que faça polling de tarefas que ainda estejam marcadas como pendentes após um limiar de obsolescência. Se um callback for perdido, o poller o captura. Se o poller atrasar, o callback geralmente chega primeiro e marca a tarefa como concluída.

Escolha pelo tipo de carga de trabalho, não pelo tipo de código. Endpoints interativos podem fazer polling; filas em segundo plano podem usar callbacks; o armazenamento de tarefas e a lógica de estado terminal devem ser idênticos para ambos, para que a estratégia de obtenção possa mudar sem tocar no resto do sistema.

Configurando o cliente e enviando a tarefa

O esqueleto de envio deve permanecer enxuto. Tudo a seguir pressupõe haver um único lugar para armazenar sua chave de API e URL base, e uma função que transforma prompt mais parâmetros em um identificador de tarefa.

Configuração e tratamento de chaves

Nunca codifique chaves diretamente. Leia de variáveis de ambiente e falhe ruidosamente na inicialização se estiverem ausentes, para que você não descubra o problema na primeira requisição de um cliente.

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # KeyError na inicialização é intencional

Mantenha as chaves longe dos logs. Se você registrar requisições para depuração, registre prompts e parâmetros, mas mascare o cabeçalho de autorização.

A chamada de envio única

A chamada de envio deve fazer apenas uma coisa: validar os parâmetros que você controla, enviar a requisição e retornar o identificador da tarefa. Não inicie o polling dentro dela e não baixe 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),  # conexão, leitura
    )
    resp.raise_for_status()
    return resp.json()

A chave de idempotência importa. Se sua chamada de envio expirar, você na verdade não sabe se a tarefa foi criada. Repetir com a mesma chave reenvia com segurança, em vez de gerar — e pagar por — uma tarefa duplicada.

O que persistir imediatamente

No momento em que o envio retorna, antes de fazer qualquer outra coisa, grave uma linha no seu próprio banco de dados:

• O ID da tarefa e a chave de idempotência
• Seu próprio ID de requisição interno e o usuário ou lote ao qual pertence
• Os parâmetros exatos da requisição, para que você possa reproduzi-la ou auditá-la
• O estado pending e um timestamp

Se seu processo falhar após o envio e antes da persistência, a tarefa existe upstream mas não há registro local. Persistir primeiro, depois continuar, faz do seu banco de dados a fonte da verdade para o que precisa ser reconciliado.

Fazendo polling de resultados do jeito certo

Loops de polling são fáceis de escrever mal. A versão ingênua — um while True sem sleep — consome cota, martela a API e torna seus logs difíceis de ler. Um loop correto tem quatro propriedades: intervalo, backoff, limite rígido e verificação de estado terminal.

O loop

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"A tarefa {job_id} terminou como {status}: {body.get('error')}")

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

    raise TimeoutError(f"A tarefa {job_id} ainda estava pendente após {max_attempts} tentativas")

Por que cada parte importa

Backoff. Tarefas de geração avançam por fases: enfileirada, executando, renderizando. Fazer polling a cada segundo nas fases iniciais de enfileiramento retorna a mesma resposta vinte vezes. Começar com alguns segundos e dobrar até um teto faz com que o volume de chamadas seja proporcional à duração da tarefa, e não à sua paciência.

Jitter. Adicionar alguns décimos de segundo aleatórios evita que todos os clientes que você executa façam polling sincronizado após um deploy ou uma tempestade de retentativas.

Número máximo de tentativas e orçamento de tempo de parede. Backoff exponencial sem um limite de tentativas pode rodar por horas. Escolha um teto com base na duração esperada do pior caso da tarefa e trate a ultrapassagem como uma falha a ser exposta, não como um loop que continua silenciosamente. Também acompanhe o tempo decorrido — depois que o teto do intervalo entra em vigor, o número de tentativas sozinho engana.

Detecção de estado terminal. Distinga explicitamente sucesso, falha e cancelamento. Uma tarefa com falha não deve ser repetida pelo poller; ela deve ser reportada com o erro do provedor para que você decida se reenvia com parâmetros diferentes.

Mais uma regra: não faça polling dentro de um handler de requisição web. Mova o loop para um worker em segundo plano, para que tarefas lentas não ocupem o pool de threads das suas requisições.

Recebendo resultados via callback de webhook

O endpoint de webhook é uma rota HTTP pública e, por padrão, não autenticada, para a qual qualquer pessoa pode fazer POST. Trate-o como tal.

Verificando o remetente

Se a plataforma assina callbacks, verifique a assinatura sobre o corpo bruto da requisição antes de analisar o JSON — verificações de assinatura são executadas sobre bytes, e re-serializar a carga útil pode alterá-los. Compare usando uma função de tempo constante e rejeite incompatibilidades com um status não-2xx. Se a assinatura não estiver disponível, no mínimo use um segmento de caminho secreto mais uma lista de permissões ou um token de cabeçalho compartilhado, e nunca confie no conteúdo da carga útil antes de correlacioná-la a uma tarefa que você criou.

Responda rápido, trabalhe depois

O único trabalho do seu endpoint é confirmar a recepção. Analise, verifique, enfileire, responda. Qualquer coisa lenta — baixar o vídeo, chamar o banco de dados sob um lock, chamar serviços downstream — acontece em um worker, não no handler. Provedores repetem callbacks que não receberam uma resposta de sucesso rápida, e um handler lento transforma um evento em vários.

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)  # retorna imediatamente
    return {"ok": True}, 200

Reconcilie com seus registros

Antes de agir com base em um callback, procure o ID da tarefa na sua própria tabela. Essa busca leva a três casos:

• ID de tarefa desconhecido: pode ser um evento obsoleto de uma tarefa excluída ou uma requisição forjada. Registre e descarte. Não crie registros a partir de cargas úteis de entrada.
• Tarefa já marcada como concluída: é uma entrega duplicada. Confirme e não faça mais nada. Baixar novamente e cobrar novamente pela mesma tarefa é um defeito clássico de webhook.
• Tarefa pendente: marque-a como concluída (ou com falha), armazene a referência do ativo e passe para a etapa de download.

Faça a transição de estado condicional ao estado atual — ou seja, uma atualização que só tem sucesso se a tarefa ainda estiver pendente — para que duas entregas concorrentes não possam vencer ao mesmo tempo. Combine isso com o poller híbrido descrito anteriormente, e nenhuma tarefa será perdida mesmo que o callback nunca chegue.

Timeouts, retentativas e tratamento de falhas

Retentativas são onde as integrações silenciosamente dão errado. As regras a seguir se aplicam a envio, polling e callbacks, mas os limites de segurança diferem.

Classifique os erros primeiro

Transitórios: conexão redefinida, respostas 5xx, respostas de rate limit, timeout de leitura. Esses podem ter sucesso na segunda tentativa.

Terminais: erros de validação 4xx, parâmetros não suportados, falhas de autenticação e falhas no nível da tarefa (por exemplo, prompt rejeitado). Repetir esses apenas repete a mesma falha e, para tarefas já enviadas, repete o custo.

Em caso de dúvida, inspecione o código de status e o corpo do erro em vez de assumir por padrão que tudo deve ser repetido.

Backoff com jitter e parada rígida

Use backoff exponencial com jitter aleatório para erros transitórios, com limites tanto no número de tentativas quanto no tempo total decorrido. Uma política de retentativas sem limite é um gerador de falhas: quando uma dependência degrada, seu volume de retentativas multiplica a carga exatamente quando ela menos consegue suportar.

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

Idempotência entre fronteiras

• Envio: sempre envie a mesma chave de idempotência ao repetir um envio. É a única forma segura de se recuperar de um timeout de uma requisição que talvez tenha criado uma tarefa. Gere a chave uma vez, armazene-a junto com a requisição pretendida e reutilize-a.
• Polling: naturalmente idempotente — somente leitura. Repita livremente dentro do seu orçamento de tempo.
• Callbacks: a entrega é essencialmente pelo menos uma vez. Deduplique por ID de tarefa mais tipo de evento e faça as transições de estado do handler condicionais, para que entregas duplicadas sejam no-op.

Exponha estados terminais

Cada caminho de tarefa deve terminar armazenando exatamente um resultado: sucesso com uma referência de ativo, ou falha com um motivo e histórico de tentativas. Nunca deixe uma tarefa permanecer indefinidamente em pending — seu poller híbrido existe exatamente para forçar uma decisão após um limiar de obsolescência. Exponha esse estado para quem consome a tarefa, para que o código upstream possa decidir reenviar, alertar ou devolver a atenção, em vez de esperar para sempre.

Baixando e armazenando os vídeos concluídos

A URL na carga útil de conclusão é temporária. Ela pode ser assinada, pode expirar e seu host não é algo com que seu aplicativo deva contar no momento da leitura. A etapa final da integração é mover o artefato para um armazenamento que você controle.

Obtendo o ativo

Baixe com uma requisição em streaming, para que arquivos grandes não caiam na memória como um único objeto bytes, e defina timeouts que reflitam o tamanho do arquivo, não uma chamada de API típica.

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)

Se seu backend de armazenamento suportar escrita em streaming, entregue o iterador de resposta diretamente a ele em vez de gravar em um arquivo local temporário.

Verifique antes de confiar

No mínimo, verifique se o status da resposta foi bem-sucedido, se o comprimento do conteúdo corresponde a qualquer tamanho declarado e se o tamanho do arquivo é não trivial. Se a carga útil de conclusão incluir metadados de duração ou formato, verifique por amostragem se o artefato baixado parece um vídeo, e não uma página de erro ou um arquivo parcial truncado. Um ativo corrompido que chega ao usuário é pior do que uma tarefa que você marca como falha e reenvia.

Mova para o seu próprio armazenamento

Grave o objeto no seu bucket ou origem de CDN, com uma chave derivada do seu ID de tarefa interno, e registre essa chave na linha da tarefa. Sirva os usuários a partir da sua própria URL. Os benefícios se acumulam:

• URLs temporárias do provedor nunca vazam para seu frontend, e-mails ou logs.
• A expiração deixa de importar; você pode regenerar derivados ou recodificar no seu próprio cronograma.
• Retenção, controle de acesso e exclusão passam a ser sua política, não a do provedor.

Mantenha a etapa de download idempotente: se a chave de destino já existe e é válida, pule a obtenção. Em callbacks duplicados, essa verificação é exatamente o que impede um segundo download e uma segunda gravação no armazenamento.

Controlando os custos de geração

A geração é cobrada pelo trabalho por requisição — normalmente por duração de saída ou unidade equivalente a token, com a taxa exata exibida no painel de preços da plataforma. Como cada chamada é muito mais cara que uma requisição de texto típica, o controle de custos deve ficar no ponto de chamada, não em uma revisão mensal.

Limite a concorrência

A proteção única mais eficaz é limitar o número de tarefas de geração concorrentes. Sem isso, tempestades de retentativas, loops descontrolados ou picos de requisições de usuários podem multiplicar gastos mais rápido do que qualquer painel consegue mostrar. Coloque um semáforo ou um pool de workers limitado antes do envio, e dimensione-o deliberadamente.

from threading import Semaphore

_submit_gate = Semaphore(5)  # ajuste conforme seu orçamento e os limites do provedor

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

Valide antes de enviar

A maior parte dos estouros acidentais de orçamento vem de requisições que nunca deveriam ter sido enviadas: parâmetros de duração muito acima do que o recurso precisa, loops de retentativa que reenviam tarefas ainda em execução, prompts vazios disparados por um defeito do cliente. Valide os parâmetros — duração, resolução, comprimento do prompt, referências de mídia de entrada — antes que a requisição saia do seu processo, e rejeite cedo com um erro claro.

Registre o volume de tarefas, não apenas erros

Registre cada envio com seus parâmetros, a duração da requisição, o usuário e a chave de idempotência. Conte tarefas por recurso e por usuário por período. O volume agregado de tarefas é a métrica que realmente acompanha os gastos e permite capturar padrões anômalos — um chamador gerando dez vezes a taxa normal — enquanto ainda são pequenos.

Não pague duas vezes pela mesma tarefa

Há três fontes comuns de falha que causam geração duplicada, cada uma com sua defesa:

• Retentativa de envio após timeout sem chave de idempotência. Reutilize a chave.
• Reentrega de webhook disparando um novo download ou uma nova tarefa. Deduplique por ID de tarefa e torne as transições condicionais.
• O poller desiste e reenvia enquanto a tarefa original ainda está em execução. Verifique o estado da tarefa existente antes de criar uma substituta e acompanhe a relação de substituição na tabela de tarefas.

Trate cada reenvio como uma decisão que precisa de um motivo registrado, não como um caminho de recuperação padrão.

Lista de verificação antes do lançamento

Antes de colocar a integração em produção, valide essas decisões e modos de falha de ponta a ponta.

Obtenção e ciclo de vida

• Escolha o modo de obtenção por carga de trabalho, com um fallback híbrido para que callbacks perdidos não percam tarefas.
• Após o envio, grave a linha da tarefa antes de qualquer outro efeito colateral, registrando parâmetros, chave de idempotência e proprietário.
• Cada tarefa atinge exatamente um estado terminal; um limiar de obsolescência força uma decisão sobre qualquer coisa ainda pendente.
• As transições de estado são condicionais ao estado atual, de modo que entregas duplicadas e polling concorrente são inofensivos.

Comportamento de rede

• Defina timeouts de conexão e leitura para cada requisição, incluindo downloads.
• Classifique erros como transitórios e terminais, e repita apenas os transitórios.
• As retentativas usam backoff exponencial com jitter e limites rígidos de tentativas e tempo decorrido.
• O backoff de polling tem tanto um teto de intervalo quanto um orçamento de tempo de parede, e o loop de polling roda em um worker, não no handler de requisição.

Callbacks e armazenamento

• O endpoint de callback verifica o remetente em relação aos bytes brutos, retorna rápido e enfileira o trabalho.
• Cada callback é reconciliado com seus próprios registros de tarefas antes de qualquer ação.
• Ativos concluídos são transmitidos em streaming para o seu próprio armazenamento, validados e servidos somente a partir da sua URL; a etapa de download é idempotente.

Custos e operações

• O envio é limitado por concorrência e validado por parâmetros de duração, resolução e entrada.
• O volume de tarefas é registrado por recurso e por usuário, para que os gastos apareçam como tendência, e não como surpresa.
• Os caminhos de geração duplicada — retentativa de envio, reentrega de webhook, reenvio pelo poller — são tratados explicitamente.
• Existem alertas para tarefas presas em pending além de um limiar e para casos que atingem continuamente o limite de concorrência.

Execute uma tarefa real por cada caminho — sucesso, falha, callback perdido, callback duplicado — e confirme que os estados terminais, as gravações no armazenamento e os contadores parecem corretos. Se cada item acima se mantiver nesse ensaio, a integração está pronta para entrar em produção com a API de geração de vídeo do HiAPI.ai.