Integrare l'API di generazione video AI in Python: polling, webhook e controllo dei costi

Scegliere la modalità di ottenimento dei risultati

Rispetto alla maggior parte delle chiamate API, la generazione video è lenta. Una richiesta che genera un video di pochi secondi può richiedere da decine di secondi a diversi minuti, il che significa che la risposta HTTP alla creazione del task quasi mai è la risposta che contiene il video. Prima di scrivere una sola riga di codice Python, decidi come otterrai i risultati. Esistono tre modalità pratiche.

Modalità 1: invia e interroga

Invii una richiesta di generazione, ricevi immediatamente un handle del task, poi chiami l'endpoint di stato a intervalli regolari finché il task non raggiunge uno stato terminale. È la modalità più semplice da implementare e debuggare: è normale codice richiesta/risposta, funziona dietro qualsiasi firewall e non richiede requisiti di rete in ingresso. Il suo svantaggio è un limite inferiore di latenza — la freschezza del risultato dipende solo dal tuo intervallo di polling — e spreca chiamate quando il polling è più veloce dell'avanzamento del task.

Il polling è adatto a carichi di lavoro interattivi: l'utente attende nell'interfaccia, si prevede che il task si completi in un tempo limitato e desideri un unico percorso di codice su cui ragionare.

Modalità 2: callback webhook

Invii il task allegando un URL di callback, restituisci il controllo al chiamante e lasci che la piattaforma effettui un POST del risultato al tuo endpoint al completamento del task. Questo elimina completamente il polling e ti offre una notifica quasi istantanea. Funziona anche dopo il riavvio del processo che ha inviato il task, perché la consegna è guidata dall'esterno.

Il costo è operativo: hai bisogno di un endpoint HTTPS accessibile pubblicamente, devi verificare che il callback provenga effettivamente dal fornitore e devi gestire il caso in cui il callback non arrivi mai — perché le reti si guastano, gli endpoint vengono configurati male e i payload si perdono.

I webhook sono adatti a carichi di lavoro di lunga durata o batch: grandi code di task, pipeline asincrone e qualsiasi scenario in cui mantenere una connessione o un ciclo a vuoto per diversi minuti è uno spreco.

Modalità 3: modalità ibrida con fallback

La modalità che la maggior parte dei sistemi di produzione finisce per adottare: il webhook come segnale principale, ma mantenendo una riconciliazione periodica lenta che interroga i task ancora contrassegnati come in attesa oltre una soglia di obsolescenza. Se il callback viene perso, il poller lo cattura. Se il poller è in ritardo, il callback di solito arriva prima e contrassegna il task come completato.

Scegli in base al carico di lavoro, non in base alla codebase. Gli endpoint interattivi possono usare il polling; le code in background possono usare i callback; l'archivio condiviso dei task e la logica di stato terminale dovrebbero essere identici per entrambi, così la strategia di ottenimento può cambiare senza toccare il resto del sistema.

Configurare il client e inviare il task

Lo scheletro di invio dovrebbe restare snello. Tutto quanto segue presuppone un unico luogo in cui conservare la tua chiave API e l'URL di base, e una funzione che converte prompt più parametri in un handle del task.

Configurazione e gestione delle chiavi

Non codificare mai le chiavi in modo fisso. Leggile dalle variabili d'ambiente e fallisci rumorosamente all'avvio se mancano, così non scoprirai il problema alla prima richiesta di un cliente.

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # il KeyError all'avvio è intenzionale

Tieni le chiavi lontane dai log. Se registri le richieste per il debug, registra i prompt e i parametri, ma oscura l'header di autorizzazione.

Chiamata di invio singola

La chiamata di invio dovrebbe fare una sola cosa: validare i parametri che controlli, inviare la richiesta e restituire l'identificatore del task. Non avviare il polling al suo interno e non scaricare nulla.

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),  # connessione, lettura
    )
    resp.raise_for_status()
    return resp.json()

La chiave di idempotenza è importante. Se la tua chiamata di invio va in timeout, in realtà non sai se il task è stato creato. Riprovare con la stessa chiave è sicuro: invii di nuovo invece di generare — e pagare per — task duplicati.

Cosa persistere immediatamente

Nel momento in cui l'invio ritorna, prima di fare qualsiasi altra cosa, scrivi una riga nel tuo database:

• ID del task e chiave di idempotenza
• Il tuo ID di richiesta interno e l'utente o il batch a cui appartiene
• I parametri esatti della richiesta, così puoi riprodurla o verificarla
• Lo stato pending e un timestamp

Se il tuo processo va in crash dopo l'invio ma prima della persistenza, il task esiste a monte ma non ha un record locale. Persistere prima e poi continuare rende il tuo database la fonte di verità per ciò che devi riconciliare.

Interrogare i risultati nel modo giusto

I cicli di polling sono facili da scrivere male. La versione ingenua — un while True senza sleep — consuma quota, martella l'API e rende i tuoi log difficili da leggere. Un ciclo corretto ha quattro proprietà: intervallo, backoff, limite massimo e controllo dello stato terminale.

Il ciclo

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"Il task {job_id} è terminato con stato {status}: {body.get('error')}")

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

    raise TimeoutError(f"Il task {job_id} è ancora in attesa dopo {max_attempts} tentativi")

Perché ogni parte è importante

Backoff. I task di generazione avanzano per fasi: in coda, in esecuzione, rendering. Interrogare ogni secondo durante la fase iniziale di coda restituisce la stessa risposta venti volte. Partire da pochi secondi e raddoppiare fino al limite massimo rende il volume di chiamate proporzionale alla durata del task, non alla tua pazienza.

Jitter. Aggiungere frazioni di secondo casuali impedisce che ogni client che esegui interroghi in modo sincronizzato dopo un deploy o una tempesta di retry.

Numero massimo di tentativi e budget di tempo reale. Un backoff esponenziale senza limite di tentativi può andare avanti per ore. Scegli un limite in base alla durata peggiore prevista del task e tratta il suo superamento come un fallimento da esporre, non come un ciclo che continua silenziosamente. Tieni anche traccia del tempo trascorso: una volta che il limite dell'intervallo entra in vigore, il solo numero di tentativi diventa fuorviante.

Rilevamento dello stato terminale. Distingui chiaramente successo, fallimento e annullamento. Un task fallito non dovrebbe essere ritentato dal poller; dovrebbe portare con sé il rapporto di errore del fornitore, così puoi decidere se reinviarlo con parametri diversi.

Un'altra regola: non eseguire il polling all'interno di un handler di richieste web. Sposta il ciclo in un worker in background, così i task lenti non occupano il pool di thread delle richieste.

Ricevere i risultati tramite callback webhook

L'endpoint webhook è una rotta HTTP pubblica e non autenticata per impostazione predefinita, su cui chiunque può effettuare un POST. Trattala di conseguenza.

Verificare il mittente

Se la piattaforma firma i callback, verifica la firma sul corpo grezzo della richiesta prima di analizzare il JSON — il controllo della firma opera sui byte e la riserializzazione del payload potrebbe modificarli. Confronta usando una funzione a tempo costante e rifiuta le discrepanze con uno stato non 2xx. Se la firma non è disponibile, usa almeno un segmento di percorso segreto più una allowlist o un token di header condiviso, e non fidarti mai del contenuto del payload prima di averlo associato a un task che hai creato.

Rispondere subito, lavorare dopo

L'unico compito del tuo endpoint è confermare la ricezione. Analizza, verifica, accoda, rispondi. Qualsiasi cosa lenta — scaricare il video, accedere al database sotto lock, chiamare servizi a valle — avviene nel worker, non nell'handler. Il fornitore ritenterà i callback che non hanno ricevuto una risposta di successo rapida, e un handler lento trasforma un evento in più eventi.

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

Riconciliare con i tuoi record

Prima di agire in base a un callback, cerca l'ID del task nella tua tabella. Quella ricerca produce tre casi:

• ID task sconosciuto: potrebbe essere un evento obsoleto di un task eliminato o una richiesta falsificata. Registralo e scartalo. Non creare record dal payload in ingresso.
• Task già contrassegnato come completato: si tratta di una consegna duplicata. Confermala e non fare altro. Scaricare di nuovo e addebitare di nuovo lo stesso task è il classico difetto dei webhook.
• Task in attesa: contrassegnalo come completato (o fallito), memorizza il riferimento all'asset e passalo al passo di download.

Rendi le transizioni di stato condizionate allo stato corrente — ovvero aggiornamenti che riescono solo se il task è ancora in attesa — così due consegne concorrenti non possono entrambe vincere. Combinalo con il poller ibrido descritto in precedenza e nessun task andrà perso anche se il callback non arriva mai.

Timeout, retry e gestione dei fallimenti

I retry sono il punto in cui l'integrazione si guasta silenziosamente. Le regole seguenti si applicano a invio, polling e callback, ma i confini di sicurezza sono diversi.

Classificare prima gli errori

Transitori: reset di connessione, risposte 5xx, risposte di rate limiting, timeout di lettura. Questi potrebbero riuscire al secondo tentativo.

Terminali: errori di validazione 4xx, parametri non supportati, errori di autenticazione e fallimenti a livello di task (come un prompt rifiutato). Riprovarli ripete solo lo stesso fallimento e, per i task già inviati, anche il costo.

In caso di dubbio, controlla il codice di stato e il corpo dell'errore invece di riprovare tutto per impostazione predefinita.

Backoff con jitter e arresto netto

Usa un backoff esponenziale con jitter casuale per gli errori transitori, con limiti sia sul numero di tentativi che sul tempo totale trascorso. Una strategia di retry senza limiti è un generatore di guasti: quando una dipendenza si degrada, il volume dei tuoi retry moltiplica il carico nel momento in cui è meno in grado di sopportarlo.

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

Idempotenza attraverso i confini

• Invio: invia sempre la stessa chiave di idempotenza quando ritenti un invio. È l'unico modo sicuro per riprendersi da un timeout di una richiesta che potrebbe aver creato il task. Genera la chiave una volta, memorizzala insieme alla richiesta prevista e riutilizzala.
• Polling: naturalmente idempotente — sola lettura. Puoi ritentare liberamente entro il tuo budget di tempo.
• Callback: la consegna è per natura at-least-once. Deduplica per ID task più tipo di evento e rendi condizionate le transizioni di stato dell'handler, così una consegna duplicata è un'operazione nulla.

Esporre lo stato terminale

Ogni percorso di un task dovrebbe alla fine memorizzare esattamente un risultato: successo con un riferimento all'asset, oppure fallimento con una causa e la cronologia dei tentativi. Non lasciare mai un task in pending a tempo indefinito — il tuo poller ibrido esiste proprio per imporre una decisione dopo la soglia di obsolescenza. Esponi quello stato a chi consuma il task, così il codice a monte può decidere se reinviare, avvisare o dare priorità altrove, invece di attendere per sempre.

Scaricare e archiviare i video completati

L'URL nel payload di completamento è temporaneo. Potrebbe essere firmato, potrebbe scadere e il suo host non è qualcosa su cui la tua applicazione dovrebbe fare affidamento in lettura. Il passo finale dell'integrazione è spostare l'artefatto in uno storage che controlli tu.

Recuperare l'asset

Scarica con una richiesta in streaming, così i file grandi non finiscono in memoria come un unico oggetto bytes, e imposta timeout che riflettano la dimensione del file invece di quelli di una tipica chiamata 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)

Se il tuo backend di storage supporta la scrittura in streaming, passa direttamente l'iteratore della risposta invece di appoggiarti a un file locale.

Verificare prima di fidarsi

Controlla almeno che lo stato della risposta sia di successo, che la lunghezza del contenuto corrisponda a qualsiasi dimensione dichiarata e che la dimensione del file non sia banale. Se il payload di completamento contiene metadati di durata o formato, verifica a campione che l'artefatto scaricato possa essere un video, piuttosto che una pagina di errore o un file parziale troncato. Un asset danneggiato che arriva all'utente è peggio di un task che hai contrassegnato come fallito e reinviato.

Spostare nel tuo storage

Scrivi l'oggetto nel tuo bucket o nell'origin della tua CDN, con una chiave derivata dal tuo ID task interno, e registra quella chiave sulla riga del task. Servi gli utenti dal tuo URL. I vantaggi si accumulano:

• Gli URL temporanei del fornitore non trapelano mai nel tuo frontend, nelle email o nei log.
• La scadenza non conta più; puoi rigenerare derivati o ricodificare secondo i tuoi tempi.
• Conservazione, controllo degli accessi e cancellazione diventano la tua policy, non quella del fornitore.

Mantieni idempotente il passo di download: se la chiave di destinazione esiste già ed è valida, salta il recupero. In caso di callback duplicati, questo controllo è proprio ciò che impedisce un secondo download e una seconda scrittura nello storage.

Controllare i costi di generazione

La generazione è fatturata in base al lavoro per richiesta — di solito in base alla durata dell'output o a unità equivalenti a token, con le tariffe esatte mostrate nel pannello dei prezzi della piattaforma. Poiché ogni chiamata è molto più costosa di una tipica richiesta di testo, il controllo dei costi va collocato nel punto di chiamata, non nella revisione mensile.

Limitare la concorrenza

L'unica barriera più efficace è limitare il numero di task di generazione concorrenti. Senza di essa, tempeste di retry, cicli impazziti o picchi di richieste degli utenti possono moltiplicare la spesa a una velocità che nessuna dashboard fa in tempo a mostrare. Posiziona un semaforo o un pool di worker limitato prima dell'invio, e impostane la dimensione consapevolmente.

from threading import Semaphore

_submit_gate = Semaphore(5)  # regola in base al tuo budget e ai limiti del fornitore

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

Validare prima dell'invio

La maggior parte delle spese impreviste deriva da richieste che non avrebbero dovuto essere inviate: parametri di durata molto più alti di quanto la funzionalità richieda, cicli di retry che reinviano task ancora in esecuzione, prompt vuoti che attivano difetti del client. Valida i parametri — durata, risoluzione, lunghezza del prompt, riferimenti ai media di input — prima che la richiesta lasci il tuo processo, e rifiutala presto con un errore chiaro.

Registrare il volume dei task, non solo gli errori

Registra ogni invio con i suoi parametri, la durata della richiesta, l'utente e la chiave di idempotenza. Conta i task per funzionalità e per utente per periodo. Il volume aggregato dei task è la metrica che traccia davvero la spesa, e ti permette di intercettare schemi anomali — un chiamante che genera dieci volte il ritmo normale — mentre sono ancora piccoli.

Non pagare due volte lo stesso task

Le modalità di fallimento della generazione duplicata hanno tre origini comuni, ciascuna con la sua difesa:

• Retry di invio in timeout effettuati senza chiave di idempotenza. Riusa la chiave.
• Una nuova consegna webhook che attiva un nuovo download o un nuovo task. Deduplica per ID task e rendi condizionate le transizioni.
• Un poller che si arrende e reinvia mentre il task originale è ancora in esecuzione. Verifica lo stato del task esistente prima di crearne uno sostitutivo, e traccia la relazione di sostituzione nella tabella dei task.

Tratta ogni reinvio come una decisione che richiede una ragione registrata, non come un percorso di recupero predefinito.

Checklist prima del go-live

Prima che l'integrazione vada in produzione, verifica end-to-end queste decisioni e modalità di fallimento.

Ottenimento e ciclo di vita

• La modalità di ottenimento è scelta in base al carico di lavoro, con un fallback ibrido, così i callback persi non fanno perdere task.
• Dopo l'invio, la riga del task viene scritta prima di qualsiasi altro effetto collaterale, registrando parametri, chiave di idempotenza e proprietario.
• Ogni task raggiunge esattamente uno stato terminale; la soglia di obsolescenza impone una decisione per qualsiasi cosa ancora in attesa.
• Le transizioni di stato sono condizionate allo stato corrente, così consegne duplicate e polling concorrenti sono innocui.

Comportamento di rete

• Ogni richiesta imposta timeout di connessione e lettura, inclusi i download.
• Gli errori sono classificati in transitori e terminali, e si ritentano solo i transitori.
• I retry usano un backoff esponenziale con jitter, con limiti rigidi su tentativi e tempo trascorso.
• Il backoff del polling ha sia un limite di intervallo sia un budget di tempo reale, e il ciclo di polling gira in un worker invece che nell'handler delle richieste.

Callback e storage

• L'endpoint di callback verifica il mittente sul corpo grezzo, risponde rapidamente e accoda il lavoro.
• Ogni callback viene riconciliato con i tuoi record dei task prima di intraprendere qualsiasi azione.
• Gli asset completati vengono inviati in streaming nel tuo storage, verificati e serviti solo dai tuoi URL; il passo di download è idempotente.

Costi e operazioni

• L'invio è controllato da limiti di concorrenza e da una validazione dei parametri su durata, risoluzione e input.
• Il volume dei task è registrato per funzionalità e per utente, così la spesa è visibile come tendenza, non come sorpresa.
• I percorsi di generazione duplicata — retry di invio, riconsegna webhook, reinvio del poller — sono tutti gestiti esplicitamente.
• Esistono avvisi per i task bloccati in attesa oltre la soglia e per le situazioni che toccano ripetutamente i limiti di concorrenza.

Esegui un task reale attraverso ogni percorso — successo, fallimento, callback perso, callback duplicato — e conferma che stati terminali, scritture nello storage e contatori appaiano corretti. Se ognuno dei punti precedenti regge in questa esercitazione, l'integrazione è pronta per il go-live con l'API di generazione video di HiAPI.ai.