Construa · Django

Renderize imagens de try-on a partir de um worker Django

Em uma frase

Uma integração com Django armazena uma linha por job, faz o envio a partir de uma task do Celery, realiza o polling a cada 3 segundos até que o status seja completed e, em seguida, baixa o arquivo para o seu próprio storage.

Obter uma chave de APILeia a documentação do endpoint de try-on
  • Credits: 5 credits em 2K · 6 credits em 4K
  • Tempo típico: 1.5–4 min
  • Processamento: assíncrono, polling

Todos os números nesta página vêm de a documentação da API da Photta · verificado

Um model que espelha o job, uma task Celery que envia e faz polling, e um storage que mantém o arquivo após a expiração da URL de entrega.

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

Como funciona

Uma linha, uma task, um arquivo

Modele o job, envie a partir de uma task, faça o polling com backoff, persista o arquivo, exponha o status.

  1. Passo 1

    Modele o job

    Uma linha por geração: o produto, o id da geração, o status e a URL de saída. A linha é o que suas views leem; a API é o que o worker lê.

  2. Passo 2

    Obtenha uma chave de API da Photta

    Leia-a a partir do ambiente no settings, nunca do banco de dados, e use uma chave separada por ambiente para que um vazamento em staging não consuma credits de produção.

  3. Passo 3

    Envie em uma task

    As views devem responder imediatamente. A task do Celery faz o POST do job e grava o id retornado na linha.

  4. Passo 4

    Faça o polling com backoff

    Faça o polling a cada 3 segundos até que o status seja completed ou failed. Em um 429, aguarde os segundos informados no Retry-After em vez de tentar novamente de imediato.

  5. Passo 5

    Persista o arquivo

    Baixe a imagem finalizada para o seu próprio storage. A URL entrega um job; o seu bucket é o que serve a página do produto por anos.

  6. Passo 6

    Exponha o status

    Seu front-end faz o polling na sua linha, não na API da Photta. Isso mantém a chave no servidor e permite exibir o progresso como você preferir.

Model, task, erros, storage

O fluxo de envio e polling e o tratamento de erros são gerados a partir da documentação da 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"])

Dúvidas frequentes de desenvolvedores

Perguntas que desenvolvedores Django fazem

Por que usar um worker em vez de uma view?

Uma geração leva mais tempo do que uma requisição deveria durar. Um worker sobrevive a uma aba do navegador fechada, a um deploy e a um timeout; uma view não sobrevive a nenhum deles.

Por quanto tempo a task deve fazer o polling?

O tempo suficiente para a latência documentada do endpoint, com um teto. Os exemplos aqui param após um número fixo de tentativas em vez de fazer polling indefinidamente.

Posso usar Django assíncrono em vez do Celery?

Sim — as chamadas são HTTP puro. O que importa é que o trabalho sobreviva à requisição, independentemente de como você o estruture.

Como posso testar sem gastar credits?

Grave uma resposta real e a reproduza nos testes. Os endpoints de recursos e a análise de fotos de detalhes são gratuitos, portanto verificações de chave e conectividade não custam nada.

Uma única chave pode atender a todos os ambientes?

Pode, mas qualquer chave consome o mesmo saldo, então um loop em staging drena credits de produção. Chaves separadas tornam o dano visível e fácil de conter.

Páginas relacionadas

Leituras relacionadas

Todos os números nesta página vêm de a documentação da API da Photta · verificado

Um worker, uma linha e um arquivo

O fluxo de envio e polling nesta página é gerado a partir da documentação da API.

Obter uma chave de APILeia o quickstart