Build · Django

Renderiza imágenes de try-on desde un worker de Django

En una frase

Una integración con Django almacena una fila por trabajo, lo envía desde una tarea de Celery, hace polling cada 3 segundos hasta que el estado sea completed y luego descarga el archivo en tu propio almacenamiento.

  • Credits: 5 credits a 2K · 6 credits a 4K
  • Tiempo típico: 1.5–4 min
  • Procesamiento: asíncrono, polling

Cada cifra de esta página proviene de la documentación de la API de Photta · verificado

Un modelo que refleja el job, una tarea de Celery que envía y hace polling, y almacenamiento que conserva el archivo una vez que la URL de entrega caduca.

RESPONSE=$(curl -s -X POST https://api.photta.app/api/v1/tryon/apparel \
  -H "Authorization: Bearer $PHOTTA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
  "product_type": "top",
  "product_images": [
    "https://cdn.yourstore.com/products/shirt-front.jpg"
  ],
  "product_back_image": "https://cdn.yourstore.com/products/shirt-back.jpg",
  "product_category": "t_shirt",
  "mannequin_id": "mnk_f3a9b1",
  "pose_id": "pose_c2d4e5",
  "background": "https://cdn.yourstore.com/backgrounds/studio-loft.jpg",
  "lock_background": true,
  "resolution": "2K",
  "aspect_ratio": "3:4",
  "body_size": "M",
  "expression": "neutral",
  "user_instruction": "Keep the collar pattern crisp and centered."
}')
ID=$(echo "$RESPONSE" | jq -r '.data.id')

Cómo funciona

Una fila, una tarea, un archivo

Modela el trabajo, envíalo desde una tarea, haz polling con backoff, persiste el archivo y expón el estado.

  1. Paso 1

    Modela el trabajo

    Una fila por generación: el producto, el id de generación, el estado y la URL de salida. La fila es lo que leen tus vistas; la API es lo que lee el worker.

  2. Paso 2

    Obtén una clave de API de Photta

    Léela desde el entorno en settings, nunca desde la base de datos, y usa una clave distinta por entorno para que una filtración en staging no consuma credits de producción.

  3. Paso 3

    Envía dentro de una tarea

    Las vistas deben responder de inmediato. La tarea de Celery hace un POST del trabajo y guarda el id devuelto en la fila.

  4. Paso 4

    Haz polling con backoff

    Haz polling cada 3 segundos hasta que el estado sea completed o failed. Ante un 429, espera los segundos indicados en Retry-After en lugar de reintentar de inmediato.

  5. Paso 5

    Persiste el archivo

    Descarga la imagen terminada en tu propio almacenamiento. La URL entrega un único trabajo; tu bucket es lo que servirá la página de producto durante años.

  6. Paso 6

    Expón el estado

    Tu frontend consulta tu fila mediante polling, no la API de Photta. Así mantienes la clave en el servidor y muestras el progreso como prefieras.

Modelo, tarea, errores, almacenamiento

El patrón de envío y polling junto con la gestión de errores se generan a partir de la documentación de la API.

# store/models.py
from django.db import models

class TryOn(models.Model):
    product = models.ForeignKey("Product", on_delete=models.CASCADE)
    generation_id = models.CharField(max_length=64, db_index=True)
    status = models.CharField(max_length=16, default="processing")
    output_url = models.URLField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
# store/tasks.py
from celery import shared_task

@shared_task(bind=True, max_retries=3)
def run_try_on(self, try_on_id):
    import os, requests
    
    res = requests.post(
        "https://api.photta.app/api/v1/tryon/apparel",
        headers={"Authorization": f"Bearer {os.environ['PHOTTA_API_KEY']}"},
        json={
          "product_type": "top",
          "product_images": [
            "https://cdn.yourstore.com/products/shirt-front.jpg"
          ],
          "product_back_image": "https://cdn.yourstore.com/products/shirt-back.jpg",
          "product_category": "t_shirt",
          "mannequin_id": "mnk_f3a9b1",
          "pose_id": "pose_c2d4e5",
          "background": "https://cdn.yourstore.com/backgrounds/studio-loft.jpg",
          "lock_background": True,
          "resolution": "2K",
          "aspect_ratio": "3:4",
          "body_size": "M",
          "expression": "neutral",
          "user_instruction": "Keep the collar pattern crisp and centered."
        },
    )
    data = res.json()["data"]  # data["id"], data["status"]
    
    import time
    
    def wait_for(job_id):
        for _ in range(120):
            res = requests.get(f"https://api.photta.app/api/v1/tryon/apparel/{job_id}", headers={"Authorization": f"Bearer {os.environ['PHOTTA_API_KEY']}"})
            data = res.json()["data"]
            if data["status"] == "completed":
                return data["output_url"]
            if data["status"] == "failed":
                raise RuntimeError(data.get("error_message", "generation failed"))
            time.sleep(3)
        raise TimeoutError("timed out after 6 minutes")
try:
    photta("POST", "/tryon/apparel", json={
      "product_type": "top",
      "product_images": [
        "https://cdn.yourstore.com/products/shirt-front.jpg"
      ],
      "product_back_image": "https://cdn.yourstore.com/products/shirt-back.jpg",
      "product_category": "t_shirt",
      "mannequin_id": "mnk_f3a9b1",
      "pose_id": "pose_c2d4e5",
      "background": "https://cdn.yourstore.com/backgrounds/studio-loft.jpg",
      "lock_background": True,
      "resolution": "2K",
      "aspect_ratio": "3:4",
      "body_size": "M",
      "expression": "neutral",
      "user_instruction": "Keep the collar pattern crisp and centered."
    })
except PhottaError as e:
    if e.status == 402:
        print("insufficient_credits_error: top up before retrying")
    elif e.status == 429:
        time.sleep(int(e.retry_after or 30))
    else:
        raise
# store/tasks.py — the output URL delivers one job; your storage keeps it
import requests
from django.core.files.base import ContentFile

def persist(try_on, output_url):
    body = requests.get(output_url, timeout=60).content
    try_on.product.gallery.save(f"{try_on.generation_id}.jpg", ContentFile(body))
    try_on.status, try_on.output_url = "completed", output_url
    try_on.save(update_fields=["status", "output_url"])

Preguntas frecuentes de desarrolladores

Preguntas habituales de desarrolladores de Django

¿Por qué usar un worker en lugar de una vista?

Una generación tarda más de lo que debería durar una petición HTTP. Un worker sobrevive a una pestaña cerrada en el navegador, a un despliegue y a un timeout; una vista no sobrevive a ninguno de ellos.

¿Durante cuánto tiempo debe hacer polling la tarea?

El tiempo suficiente para cubrir la latencia documentada del endpoint, estableciendo un límite máximo. Los ejemplos aquí se detienen tras un número fijo de intentos en lugar de hacer polling indefinidamente.

¿Puedo usar Django asíncrono en lugar de Celery?

Sí — las llamadas son HTTP estándar. Lo fundamental es que el trabajo sobreviva al ciclo de vida de la petición, independientemente de cómo lo implementes.

¿Cómo puedo hacer pruebas sin gastar credits?

Graba una respuesta real y reprodúcela en tus tests. Los endpoints de recursos y el análisis detallado de fotos son gratuitos, por lo que verificar la clave y la conectividad no cuesta nada.

¿Puede una sola clave servir para todos los entornos?

Poder, se puede, pero todas las claves consumen el mismo saldo, por lo que un bucle en staging agotaría los credits de producción. Separar las claves hace que el impacto sea visible y fácil de detener.

Un worker, una fila y un archivo

El flujo de envío y polling de esta página se genera a partir de la documentación de la API.