Tareas asíncronas

#Un ciclo de vida para todos los modelos asíncronos

1. 1

   Envía una sola vez

   Usa una Idempotency-Key única y guarda la respuesta de inmediato.

   POST /v1/{capability}
2. 2

   Guarda task_id

   Guárdalo antes de que el worker confirme el trabajo.

   task_id
3. 3

   Consulta o recibe el webhook

   Compara status sin distinguir mayúsculas y minúsculas, y detente en un estado final.

   GET /{task_id}
4. 4

   Obtén y guarda el resultado

   Llama a /content tras el éxito y guarda la URL de la plataforma devuelta.

   GET /{task_id}/content

#Familias de endpoints

#Valores de estado

Un ciclo de vida, dos vocabularios de API. Las tareas de imagen, música y 3D usan los términos de tareas de OpenAI; los videos y todos los payloads de webhook usan los términos de video de OpenAI. Una comprobación de estado escrita para una familia nunca coincide en la otra, sin avisar — esta tabla es la correspondencia.

#Bucle de consulta seguro

Detente en los términos finales de la familia que estás consultando, mantén un plazo de tiempo real para todo lo demás y aplica backoff para que una tarea atascada no se convierta en un bucle intensivo.

import time, requests

# The two vocabularies. Pick the one that matches the endpoint you poll.
TASK_DONE, TASK_FAILED = {"succeeded"}, {"failed"}      # images / music / 3D
VIDEO_DONE, VIDEO_FAILED = {"completed"}, {"failed"}    # videos + webhooks
DEADLINE_SECONDS = 15 * 60

def wait_for_image_task(task_id, api_key):
    started = time.monotonic()
    delay = 2
    while True:
        response = requests.get(
            f"https://www.aisplice.store/v1/images/tasks/{task_id}",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30,
        )
        response.raise_for_status()
        task = response.json()
        status = task.get("status", "")

        if status in TASK_DONE:
            return task
        if status in TASK_FAILED:
            raise RuntimeError((task.get("error") or {}).get("message", "task failed"))

        # queued / processing / unknown all mean "not finished". Wait — but
        # never forever: hand an overdue task to reconciliation.
        if time.monotonic() - started > DEADLINE_SECONDS:
            raise TimeoutError(f"task {task_id} still {status} after the deadline")

        time.sleep(delay)
        delay = min(delay * 1.5, 15)

#Retención de resultados

Llama a /content solo después del éxito. Los archivos de resultado se guardan 3 días después de que termina la tarea; después /content responde 410 con el código media_expired. Descarga una copia en esos 3 días si necesitas conservarla.