Intégrer l'API de génération vidéo par IA en Python : polling, Webhook et maîtrise des coûts

Choisir le mode de récupération des résultats

Contrairement à la plupart des appels d'API, la génération vidéo est lente. Une requête qui génère quelques secondes de vidéo peut prendre de plusieurs dizaines de secondes à plusieurs minutes, ce qui signifie que la réponse HTTP de création de tâche n'est presque jamais la réponse contenant la vidéo. Avant d'écrire une seule ligne de code Python, décidez comment vous obtiendrez le résultat. Il existe trois modes pratiques.

Mode 1 : soumettre et interroger

Vous envoyez une requête de génération, recevez immédiatement un handle de tâche, puis appelez l'endpoint de statut à intervalles fixes jusqu'à ce que la tâche atteigne un état terminal. C'est le mode le plus simple à implémenter et à déboguer : c'est du code requête/réponse classique, qui fonctionne derrière n'importe quel pare-feu et sans exigence réseau entrant. Son inconvénient est un plancher de latence — la fraîcheur du résultat dépend uniquement de votre intervalle d'interrogation — et il gaspille des appels lorsque l'interrogation est plus rapide que l'avancement de la tâche.

Le polling convient aux charges de travail interactives : un utilisateur attend dans l'UI, la tâche devrait se terminer dans un temps borné, et vous voulez un chemin de code unique sur lequel raisonner.

Mode 2 : callback Webhook

Vous soumettez la tâche avec une URL de callback, rendez le contrôle à l'appelant et laissez la plateforme POSTer le résultat vers votre endpoint à la fin de la tâche. Cela élimine entièrement le polling et vous donne une notification quasi instantanée. Cela continue également de fonctionner après le redémarrage du processus qui a soumis la tâche, puisque la livraison est pilotée de l'extérieur.

Le coût est opérationnel : il vous faut un endpoint HTTPS accessible publiquement, vérifier que le callback provient bien du fournisseur, et gérer le cas où le callback n'arrive jamais — car les réseaux tombent en panne, les endpoints sont mal configurés et les payloads se perdent.

Le Webhook convient aux charges de travail longue durée ou par lots : grandes files de tâches, pipelines asynchrones, et tout scénario où maintenir une connexion ou boucler à vide pendant plusieurs minutes est du gaspillage.

Mode 3 : mode hybride avec repli

Le mode que finissent par adopter la plupart des systèmes de production : le Webhook comme signal principal, mais avec une réconciliation périodique lente qui interroge les tâches encore marquées en attente au-delà d'un seuil de péremption. Si un callback est manqué, le poller le rattrape. Si le poller est en retard, le callback arrive généralement en premier et marque la tâche comme terminée.

Choisissez selon la charge de travail, pas selon la base de code. Un endpoint interactif peut interroger ; une file d'arrière-plan peut utiliser des callbacks ; le stockage partagé des tâches et la logique d'état terminal doivent être identiques pour les deux, afin que la stratégie de récupération puisse changer sans toucher au reste du système.

Configurer le client et soumettre une tâche

Le squelette de soumission doit rester minimal. Tout ce qui suit suppose un seul endroit pour votre clé d'API et votre URL de base, ainsi qu'une fonction qui transforme un prompt plus des paramètres en handle de tâche.

Configuration et gestion des clés

Ne codez jamais les clés en dur. Lisez-les depuis des variables d'environnement et échouez bruyamment au démarrage si elles manquent, pour ne pas découvrir le problème à la première requête client.

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # le KeyError au démarrage est volontaire

Gardez les clés hors des logs. Si vous journalisez les requêtes pour déboguer, journalisez le prompt et les paramètres, mais masquez l'en-tête d'autorisation.

L'appel de soumission unique

L'appel de soumission ne doit faire qu'une chose : valider les paramètres que vous contrôlez, envoyer la requête et renvoyer l'identifiant de tâche. N'y lancez pas de polling et n'y téléchargez rien.

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

La clé d'idempotence est importante. Si votre appel de soumission expire, vous ne savez pas réellement si la tâche a été créée. Réessayer avec la même clé est sûr pour renvoyer la requête, au lieu de créer — et de payer pour — des tâches en double.

Ce qu'il faut persister immédiatement

Au moment où la soumission revient, avant toute autre action, écrivez une ligne dans votre propre base de données :

• l'ID de tâche et la clé d'idempotence
• votre propre ID de requête interne ainsi que l'utilisateur ou le lot auquel elle appartient
• les paramètres exacts de la requête, pour pouvoir la reproduire ou l'auditer
• le statut pending et un horodatage

Si votre processus plante après la soumission mais avant la persistance, la tâche existe en amont mais sans enregistrement local. Persister d'abord, puis continuer, fait de votre base de données la source de vérité à réconcilier.

Interroger les résultats correctement

Une boucle d'interrogation est facile à écrire mal. La version naïve — une boucle while True sans sommeil — consomme du quota, martèle l'API et rend vos logs illisibles. La bonne boucle possède quatre propriétés : intervalle, backoff, plafond strict et vérification d'état terminal.

La boucle

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"Tâche {job_id} terminée avec le statut {status} : {body.get('error')}")

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

    raise TimeoutError(f"Tâche {job_id} toujours en attente après {max_attempts} tentatives")

Pourquoi chaque élément compte

Backoff. Une tâche de génération avance par phases : en file d'attente, en cours d'exécution, puis rendu. Interroger chaque seconde pendant la longue phase de mise en file renvoie vingt fois la même réponse. Démarrer à quelques secondes et doubler jusqu'au plafond proportionne le volume d'appels à la durée de la tâche, et non à votre patience.

Jitter. Ajouter quelques dixièmes de seconde aléatoires empêche chaque client de votre parc d'interroger en synchronie après un déploiement ou une tempête de retries.

Nombre maximal de tentatives et budget de temps horloge. Un backoff exponentiel sans plafond de tentatives peut tourner des heures. Choisissez un plafond basé sur la durée de tâche attendue dans le pire des cas, et traitez son dépassement comme un échec que vous exposez, et non comme une boucle qui continue silencieusement. Suivez aussi le temps écoulé — une fois le plafond d'intervalle atteint, le seul nombre de tentatives est trompeur.

Détection d'état terminal. Distinguez explicitement succès, échec et annulation. Une tâche échouée ne doit pas être réessayée par le poller ; elle doit être renvoyée avec le rapport d'erreur du fournisseur, pour que vous décidiez s'il faut resoumettre avec des paramètres différents.

Une règle de plus : n'interrogez pas à l'intérieur d'un handle de requête Web. Déplacez la boucle dans un worker d'arrière-plan, pour qu'une tâche lente ne monopolise pas votre pool de threads de requête.

Recevoir les résultats via des callbacks Webhook

L'endpoint Webhook est une route HTTP publique et non authentifiée par défaut, vers laquelle n'importe qui peut POSTer. Traitez-la en conséquence.

Vérifier l'expéditeur

Si la plateforme signe les callbacks, vérifiez la signature sur le corps brut de la requête avant de parser le JSON — la vérification de signature s'effectue sur les octets, et la resérialisation du payload peut les modifier. Comparez avec une fonction à temps constant et rejetez les discordances avec un statut non 2xx. Si une signature n'est pas disponible, utilisez au minimum un segment de chemin secret plus une liste d'autorisation ou un token d'en-tête partagé, et ne faites jamais confiance au contenu du payload avant de l'avoir fait correspondre à une tâche que vous avez créée.

Répondre vite, travailler plus tard

Le seul travail de votre endpoint est d'accuser réception. Parsez, validez, mettez en file, répondez. Tout ce qui est lent — télécharger la vidéo, appeler la base de données sous verrou, appeler un service en aval — se produit dans un worker, pas dans le handle. Le fournisseur réessaie les callbacks qui n'ont pas reçu de réponse rapide et réussie, et un handle lent transforme un événement en plusieurs.

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)  # retour immédiat
    return {"ok": True}, 200

Réconcilier avec vos enregistrements

Avant d'agir sur un callback, recherchez l'ID de tâche dans votre propre table. Cette recherche donne trois cas :

• ID de tâche inconnu : il peut s'agir d'un événement obsolète d'une tâche supprimée ou d'une requête forgée. Journalisez-le et ignorez-le. Ne créez pas d'enregistrement à partir d'un payload entrant.
• Tâche déjà marquée terminée : c'est une livraison en double. Accusez réception et ne faites rien d'autre. Retélécharger et refacturer la même tâche est le défaut classique des Webhooks.
• Tâche en attente : marquez-la comme terminée (ou échouée), stockez la référence de l'asset et passez à l'étape de téléchargement.

Rendez les transitions d'état conditionnelles à l'état courant — c'est-à-dire des mises à jour qui ne réussissent que si la tâche est encore en attente — pour que deux livraisons concurrentes ne puissent pas gagner toutes les deux. Combinez cela avec le poller hybride décrit plus haut, et aucune tâche ne sera perdue même si un callback n'arrive jamais.

Timeouts, retries et gestion des échecs

Les retries sont l'endroit où l'intégration se dégrade discrètement. Les règles ci-dessous s'appliquent à la soumission, au polling et aux callbacks, mais leur périmètre de sécurité diffère.

Classer les erreurs d'abord

Transitoires : réinitialisations de connexion, réponses 5xx, réponses de limitation de débit, timeouts de lecture. Elles peuvent réussir à la deuxième tentative.

Définitives : erreurs de validation 4xx, paramètres non pris en charge, échecs d'authentification et échecs au niveau de la tâche (comme un prompt rejeté). Les réessayer ne fait que reproduire le même échec et, pour une tâche déjà soumise, reproduce le coût.

En cas de doute, examinez le code de statut et le corps d'erreur, plutôt que de réessayer tout par défaut.

Backoff avec jitter et arrêt strict

Utilisez un backoff exponentiel avec jitter aléatoire pour les erreurs transitoires, avec un plafond à la fois sur le nombre de tentatives et sur le temps total. Une politique de retry sans plafond est un générateur de pannes : quand une dépendance se dégrade, votre volume de retries multiplie la charge au pire moment.

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

Idempotence aux frontières

• Soumission : envoyez toujours la même clé d'idempotence lorsque vous réessayez une soumission. C'est le seul moyen sûr de récupérer d'un timeout sur une requête qui a peut-être créé une tâche. Générez la clé une fois, stockez-la avec la requête prévue et réutilisez-la.
• Polling : naturellement idempotent — lecture seule. Vous pouvez réessayer librement dans votre budget de temps.
• Callbacks : la livraison est essentiellement au moins une fois. Dédupliquez par ID de tâche plus type d'événement, et rendez les transitions d'état du handle conditionnelles, pour qu'une livraison en double soit une opération neutre.

Exposer un état terminal

Chaque chemin de tâche doit finir par stocker exactement un résultat : un succès avec référence d'asset, ou un échec avec la cause et l'historique des tentatives. Ne laissez jamais une tâche rester indéfiniment en pending — votre poller hybride existe précisément pour forcer une décision après un seuil de péremption. Exposez cet état à la partie qui consomme la tâche, pour que le code amont puisse décider de resoumettre, d'alerter ou de restituer l'attention, plutôt que d'attendre éternellement.

Télécharger et stocker les vidéos terminées

L'URL du payload de fin est temporaire. Elle peut être signée, peut expirer, et son hôte n'est pas une chose sur laquelle votre application doit s'appuyer à la lecture. La dernière étape de l'intégration consiste à déplacer l'artefact vers un stockage que vous contrôlez.

Récupérer l'asset

Téléchargez avec une requête en streaming, pour que les gros fichiers ne tombent pas en mémoire sous forme d'un seul objet bytes, et fixez un timeout qui reflète la taille du fichier plutôt qu'un appel d'API classique.

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 votre backend de stockage prend en charge l'écriture en streaming, passez-lui directement l'itérateur de réponse plutôt que de passer par un fichier local temporaire.

Vérifier avant de faire confiance

Vérifiez au minimum que le statut de la réponse indique un succès, que la longueur du contenu correspond à toute taille déclarée, et que la taille du fichier n'est pas triviale. Si le payload de fin contient des métadonnées de durée ou de format, vérifiez par échantillonnage que l'artefact téléchargé est vraisemblablement une vidéo, et non une page d'erreur ou un fichier partiel tronqué. Un asset corrompu qui parvient à l'utilisateur est pire qu'une tâche que vous avez marquée comme échouée et resoumise.

Déplacer vers votre propre stockage

Écrivez l'objet dans votre bucket ou votre origine CDN, avec une clé dérivée de votre ID de tâche interne, et enregistrez cette clé sur la ligne de la tâche. Servez les utilisateurs depuis vos propres URL. Les bénéfices s'accumulent :

• Les URL temporaires du fournisseur ne fuient jamais vers votre frontend, vos e-mails ou vos logs.
• L'expiration n'a plus d'importance ; vous pouvez régénérer des dérivés ou réencoder selon votre propre calendrier.
• La rétention, le contrôle d'accès et la suppression deviennent votre politique, pas celle du fournisseur.

Gardez l'étape de téléchargement idempotente : si la clé de destination existe déjà et est valide, sautez la récupération. Lors de callbacks en double, cette vérification est précisément ce qui empêche un second téléchargement et une seconde écriture dans le stockage.

Maîtriser les coûts de génération

La génération est facturée au travail demandé — généralement sur la durée de sortie ou des unités équivalentes en tokens, le tarif exact étant indiqué dans le panneau de tarification de la plateforme. Comme chaque appel coûte bien plus cher qu'une requête texte classique, le contrôle des coûts doit se faire au point d'appel, et non lors d'une revue mensuelle.

Limiter la concurrence

Le garde-fou le plus efficace est de limiter le nombre de tâches de génération concurrentes. Sans cela, une tempête de retries, une boucle incontrôlée ou un pic de requêtes utilisateurs peut multiplier les dépenses à une vitesse qu'aucun tableau de bord n'a le temps d'afficher. Placez un sémaphore ou un pool de workers borné avant la soumission, et dimensionnez-le intentionnellement.

from threading import Semaphore

_submit_gate = Semaphore(5)  # ajustez selon votre budget et les limites du fournisseur

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

Valider avant de soumettre

La plupart des dépassements accidentels proviennent de requêtes qui n'auraient pas dû être envoyées : un paramètre de durée bien supérieur au besoin de la fonctionnalité, une boucle de retry qui resoumet une tâche encore en cours, un prompt vide déclenché par un défaut côté client. Validez les paramètres — durée, résolution, longueur du prompt, références des médias d'entrée — avant que la requête ne quitte votre processus, et rejetez tôt avec une erreur claire.

Journaliser le volume de tâches, pas seulement les erreurs

Journalisez chaque soumission avec ses paramètres, la durée de la requête, l'utilisateur et la clé d'idempotence. Comptez les tâches par fonctionnalité et par utilisateur sur des périodes. Le volume agrégé de tâches est la métrique qui suit réellement les dépenses et vous permet de détecter les schémas anormaux — un appelant qui génère dix fois le taux normal — tant qu'ils sont encore petits.

Ne payez pas deux fois pour la même tâche

Trois sources courantes produisent des générations en double, chacune avec sa défense :

• Un retry de soumission après timeout sans clé d'idempotence. Réutilisez la clé.
• Une livraison Webhook en double qui déclenche un nouveau téléchargement ou une nouvelle tâche. Dédupliquez par ID de tâche et rendez les transitions conditionnelles.
• Un poller qui abandonne et resoumet alors que la tâche d'origine tourne encore. Vérifiez le statut de la tâche existante avant de créer une tâche de remplacement, et suivez les relations de remplacement dans la table des tâches.

Traitez chaque resoumission comme une décision qui exige une raison consignée, et non comme un chemin de récupération par défaut.

Checklist avant mise en production

Avant de mettre l'intégration en production, validez ces décisions et ces modes de défaillance de bout en bout.

Récupération et cycle de vie

• Le mode de récupération est choisi selon la charge de travail, avec un repli hybride pour qu'un callback manqué ne perde pas de tâche.
• La ligne de tâche est écrite après la soumission et avant tout autre effet de bord, enregistrant les paramètres, la clé d'idempotence et le propriétaire.
• Chaque tâche atteint exactement un état terminal ; un seuil de péremption force une décision pour tout ce qui est encore en attente.
• Les transitions d'état sont conditionnelles à l'état courant, donc les livraisons en double et le polling concurrent sont inoffensifs.

Comportement réseau

• Chaque requête définit des timeouts de connexion et de lecture, y compris les téléchargements.
• Les erreurs transitoires et définitives sont classées, et seules les transitoires sont réessayées.
• Les retries utilisent un backoff exponentiel avec jitter, avec un plafond strict sur le nombre de tentatives et le temps écoulé.
• Le backoff de polling a à la fois un plafond d'intervalle et un budget de temps horloge, et la boucle de polling s'exécute dans un worker plutôt que dans le handle de requête.

Callbacks et stockage

• L'endpoint de callback vérifie l'expéditeur sur le corps brut, répond rapidement et met le travail en file.
• Chaque callback est réconcilié avec vos propres enregistrements de tâches avant toute action.
• Les assets terminés sont streamés vers votre propre stockage, vérifiés, et servis uniquement depuis vos URL ; l'étape de téléchargement est idempotente.

Coûts et exploitation

• La soumission est conditionnée par une limite de concurrence et par la validation des paramètres de durée, de résolution et d'entrée.
• Le volume de tâches est journalisé par fonctionnalité et par utilisateur, pour que les dépenses soient visibles comme une tendance et non comme une surprise.
• Les chemins de génération en double — retries de soumission, relivraisons Webhook, resoumissions du poller — sont tous traités explicitement.
• Des alertes existent pour les tâches bloquées en attente au-delà du seuil et pour les cas touchant continuellement la limite de concurrence.

Faites passer une tâche réelle dans chaque chemin — succès, échec, callback manqué, callback en double — et confirmez que les états terminaux, les écritures de stockage et les compteurs sont corrects. Si chacun des points ci-dessus tient dans cet exercice, l'intégration peut être mise en production avec l'API de génération vidéo de HiAPI.ai.