Integrazione · Django

Esegui il rendering delle immagini try-on da un worker Django

In sintesi

Un'integrazione Django memorizza una riga per job, invia la richiesta da un task Celery, esegue il polling ogni 3 secondi fino allo stato completed, quindi scarica il file nel tuo storage.

Ottieni una chiave APILeggi la documentazione dell'endpoint di try-on
  • Credits: 5 credits a 2K · 6 credits a 4K
  • Tempo tipico: 1.5–4 min
  • Elaborazione: asincrono, polling

Ogni dato in questa pagina proviene da la documentazione dell'API di Photta · verificato

Un modello che riflette il job, un task Celery che invia ed esegue il polling, e uno storage per conservare il file una volta scaduto l'URL di consegna.

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

Come funziona

Una riga, un task, un file

Modella il job, invia la richiesta da un task, esegui il polling con backoff, salva il file in modo persistente ed esponi lo stato.

  1. Passaggio 1

    Modella il job

    Una riga per ogni generazione: il prodotto, l'ID di generazione, lo stato e l'URL di output. La riga è ciò che leggono le tue view; l'API è ciò che legge il worker.

  2. Passaggio 2

    Ottieni una chiave API di Photta

    Leggila dall'ambiente nei settings, mai dal database, e usa una chiave separata per ogni ambiente: un leak in staging non deve consumare crediti di produzione.

  3. Passaggio 3

    Invia la richiesta in un task

    Le view devono rispondere immediatamente. Il task Celery invia il job con una POST e scrive l'ID restituito nella riga corrispondente.

  4. Passaggio 4

    Esegui il polling con backoff

    Esegui il polling ogni 3 secondi finché lo stato non è completed o failed. In caso di errore 429, attendi i secondi specificati in Retry-After invece di riprovare subito.

  5. Passaggio 5

    Salva il file in modo persistente

    Scarica l'immagine finale nel tuo storage personale. L'URL serve per consegnare un singolo job; il tuo bucket è ciò che servirà la pagina prodotto per anni.

  6. Passaggio 6

    Esponi lo stato

    Il tuo frontend interroga la tua riga, non l'API di Photta. In questo modo la chiave rimane protetta sul server e puoi mostrare l'avanzamento come preferisci.

Modello, task, errori, storage

Il flusso di invio e polling e la gestione degli errori sono basati sulla documentazione dell'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"])

Domande frequenti degli sviluppatori

Domande frequenti degli sviluppatori Django

Perché un worker invece di una view?

Una generazione richiede più tempo di quanto dovrebbe durare una richiesta HTTP. Un worker resiste alla chiusura del tab del browser, a un deploy e a un timeout; una view non sopravvive a nessuno di questi eventi.

Per quanto tempo deve proseguire il polling del task?

Abbastanza da coprire la latenza documentata dell'endpoint, con un limite massimo. Gli esempi qui riportati si arrestano dopo un numero fisso di tentativi anziché proseguire all'infinito.

Posso usare Django asincrono al posto di Celery?

Sì — le chiamate sono normali richieste HTTP. Ciò che conta è che l'elaborazione sopravviva al ciclo di vita della richiesta, indipendentemente da come la strutturi.

Come posso fare test senza consumare crediti?

Registra una risposta reale ed eseguine il replay nei test. Gli endpoint delle risorse e l'analisi dei dettagli della foto sono gratuiti, quindi le verifiche di connettività e della chiave non costano nulla.

Una singola chiave può bastare per tutti gli ambienti?

Tecnicamente sì, ma ogni chiave attinge allo stesso saldo: un loop incontrollato in staging prosciugherebbe i crediti di produzione. Chiavi separate rendono l'impatto visibile e facile da bloccare.

Pagine correlate

Letture correlate

Ogni dato in questa pagina proviene da la documentazione dell'API di Photta · verificato

Un worker, una riga e un file

Il flusso di invio e polling presente in questa pagina è basato sulla documentazione dell'API.

Ottieni una chiave APILeggi la guida rapida