Wdrożenie · Django

Renderowanie obrazów try-on z poziomu workera Django

W jednym zdaniu

Integracja z Django zapisuje jeden wiersz na zadanie, wysyła żądanie z poziomu taska Celery, wykonuje polling co 3 sekundy, aż status zmieni się na completed, a następnie pobiera plik do Twojego własnego storage'u.

Pobierz klucz APIDokumentacja endpointu try-on
  • Credits: 5 credits w 2K · 6 credits w 4K
  • Typowy czas: 1.5–4 min
  • Przetwarzanie: asynchroniczne, polling

Każda liczba na tej stronie pochodzi z dokumentacji API Photta · sprawdzono

Model odzwierciedlający zadanie, zadanie Celery wysyłające żądanie i wykonujące polling oraz magazyn danych zachowujący plik po wygaśnięciu tymczasowego adresu URL.

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

Jak to działa

Jeden wiersz, jeden task, jeden plik

Zamodeluj zadanie, wyślij je z taska, wykonuj polling z backoffem, zapisz plik na stałe, udostępnij status.

  1. Krok 1

    Zamodeluj zadanie

    Jeden wiersz na generację: produkt, id generacji, status i URL wyjściowy. Widoki odczytują ten wiersz; worker odczytuje API.

  2. Krok 2

    Pobierz klucz Photta API

    Odczytuj go ze zmiennych środowiskowych w settings, nigdy z bazy danych, i używaj osobnego klucza na każde środowisko, aby wyciek na stagingu nie zużył produkcyjnych credits.

  3. Krok 3

    Wysyłaj w tasku

    Widoki powinny zwracać odpowiedź natychmiast. Task Celery wykonuje POST z zadaniem i zapisuje zwrócone id w wierszu.

  4. Krok 4

    Polling z backoffem

    Wykonuj polling co 3 sekundy, aż status zmieni się na completed lub failed. Przy kodzie 429 odczekaj liczbę sekund z nagłówka Retry-After, zamiast ponawiać próbę od razu.

  5. Krok 5

    Zapisz plik na stałe

    Pobierz gotowy obraz do własnego storage'u. Otrzymany URL obsługuje jedno zadanie; Twój bucket będzie serwować stronę produktu przez lata.

  6. Krok 6

    Udostępnij status

    Twój frontend odpytuje Twój wiersz w bazie, a nie API Photta. Dzięki temu klucz pozostaje na serwerze, a Ty możesz wyświetlać postęp w dowolny sposób.

Model, task, błędy, storage

Wzorzec submit-and-poll oraz obsługa błędów zostały wygenerowane na podstawie dokumentacji 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"])

Pytania zadawane przez programistów

Pytania zadawane przez programistów Django

Dlaczego worker zamiast widoku?

Generowanie trwa dłużej, niż powinno trwać żądanie HTTP. Worker przetrwa zamkniętą kartę przeglądarki, wdrożenie i timeout; widok nie przetrwa żadnego z nich.

Jak długo task powinien wykonywać polling?

Wystarczająco długo, by pokryć udokumentowane opóźnienie endpointu, z ustawionym limitem górnym. Tutejsze przykłady zatrzymują się po ustalonej liczbie prób, zamiast odpytywać w nieskończoność.

Czy mogę użyć asynchronicznego Django zamiast Celery?

Tak — to zwykłe wywołania HTTP. Liczy się to, aby zadanie działało dłużej niż cykl życia żądania, niezależnie od przyjętego rozwiązania.

Jak testować bez zużywania credits?

Zarejestruj jedną prawdziwą odpowiedź i odtwarzaj ją w testach. Endpointy zasobów oraz szczegółowa analiza zdjęć są bezpłatne, więc sprawdzanie klucza i łączności nic nie kosztuje.

Czy jeden klucz może obsługiwać wszystkie środowiska?

Może, ale każdy klucz korzysta z tego samego salda, więc pętla na stagingu wyczyści produkcyjne credits. Osobne klucze sprawiają, że szkody są widoczne i łatwe do zatrzymania.

Powiązane strony

Powiązane materiały

Każda liczba na tej stronie pochodzi z dokumentacji API Photta · sprawdzono

Worker, wiersz i plik

Kod submit-and-poll na tej stronie został wygenerowany na podstawie dokumentacji API.

Pobierz klucz APIPrzeczytaj quickstart