Geliştirin · Django

Bir Django worker'ından try-on görselleri render edin

Tek cümlede

Bir Django entegrasyonu, iş başına bir satır saklar, bir Celery görevinden gönderir, durum completed olana kadar her 3 saniyede bir polling yapar ve ardından dosyayı kendi depolama alanınıza indirir.

  • Credits: 2K'da 5 credits · 4K'da 6 credits
  • Tipik süre: 1.5–4 min
  • İşleme: asenkron, polling

Bu sayfadaki her veri şuradan alınmıştır: Photta API dokümantasyonu · kontrol edildi

İşi yansıtan bir model, gönderen ve poll eden bir Celery task'ı ve teslimat URL'si kaybolduktan sonra dosyayı saklayan depolama.

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

Nasıl çalışır

Bir satır, bir görev, bir dosya

İşi modelleyin, bir görevden gönderin, backoff ile polling yapın, dosyayı kalıcı hale getirin, durumu dışa aktarın.

  1. Adım 1

    İşi modelleyin

    Üretim başına bir satır: ürün, üretim kimliği, durum ve çıktı URL'si. View'larınızın okuduğu şey satırdır; worker'ın okuduğu ise API'dir.

  2. Adım 2

    Bir Photta API anahtarı alın

    Bunu veritabanından asla değil, settings içindeki ortamdan okuyun ve staging ortamındaki bir sızıntının production credits harcamaması için ortam başına ayrı bir anahtar kullanın.

  3. Adım 3

    Bir görev içinde gönderin

    View'lar anında dönmelidir. Celery görevi işi POST eder ve dönen kimliği satıra yazar.

  4. Adım 4

    Backoff ile polling yapın

    Durum completed veya failed olana kadar her 3 saniyede bir polling yapın. 429 durumunda, hemen yeniden denemek yerine Retry-After saniyesi kadar bekleyin.

  5. Adım 5

    Dosyayı kalıcı hale getirin

    Tamamlanan görseli kendi depolama alanınıza indirin. URL tek bir iş sunar; ürün sayfasına yıllarca hizmet veren ise sizin bucket'ınızdır.

  6. Adım 6

    Durumu dışa aktarın

    Ön yüzünüz Photta API'sini değil, kendi satırınızı polling yapar. Bu, anahtarı sunucuda tutar ve ilerlemeyi dilediğiniz gibi göstermenizi sağlar.

Model, görev, hatalar, depolama

Gönder-ve-polling-yap adımı ile hata yönetimi API belgelerinden derlenmiştir.

# 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"])

Geliştiricilerin sıkça sorduğu sorular

Django geliştiricilerinin sorduğu sorular

Neden bir view yerine bir worker?

Bir üretim, bir isteğin sürmesi gerekenden daha uzun sürer. Bir worker; kapatılan bir tarayıcı sekmesini, bir dağıtımı ve bir zaman aşımını atlatabilir; bir view ise bunların hiçbirini atlatamaz.

Görev ne kadar süre polling yapmalı?

Endpoint'in belgelenmiş gecikme süresini karşılayacak kadar, bir üst sınırla birlikte. Buradaki örnekler, sonsuza kadar polling yapmak yerine sabit sayıda denemeden sonra durur.

Celery yerine async Django kullanabilir miyim?

Evet — çağrılar yalın HTTP'dir. Önemli olan, nasıl kurgularsanız kurgulayın, işin istekten daha uzun süre hayatta kalmasıdır.

Credits harcamadan nasıl test edebilirim?

Gerçek bir yanıtı kaydedin ve testlerde yeniden oynatın. Kaynak endpoint'leri ve ayrıntılı fotoğraf analizi ücretsizdir, bu nedenle anahtar ve bağlantı kontrolleri hiçbir maliyet getirmez.

Tek bir anahtar her ortama hizmet verebilir mi?

Verebilir, ancak her anahtar aynı bakiyeyi harcar; bu nedenle staging'deki bir döngü production credits tüketir. Ayrı anahtarlar, hasarı görünür ve durdurulması kolay hale getirir.

Bir worker, bir satır ve bir dosya

Bu sayfadaki gönder-ve-polling-yap adımı API belgelerinden derlenmiştir.