Разработка · Django

Генерация try-on изображений из воркера Django

В двух словах

Интеграция с Django сохраняет одну запись на задачу, отправляет запрос из Celery-таски, выполняет polling каждые 3 секунды до статуса completed, затем скачивает файл в ваше хранилище.

Получить API-ключДокументация эндпоинта try-on
  • Кредиты: 5 credits при 2K · 6 credits при 4K
  • Типичное время: 1.5–4 min
  • Обработка: async, polling

Все показатели на этой странице взяты из документации API Photta · проверено

Модель, отражающая задачу, задача Celery для отправки и polling, а также хранилище, сохраняющее файл после того, как временный 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')

Как это работает

Одна запись, одна таска, один файл

Опишите модель задачи, отправляйте из таски, опрашивайте с backoff, сохраняйте файл, отдавайте статус.

  1. Шаг 1

    Опишите модель задачи

    Одна строка на генерацию: товар, id генерации, статус и URL результата. Представления читают строку из базы; воркер работает с API.

  2. Шаг 2

    Получите API-ключ Photta

    Считывайте его из переменных окружения в settings, ни в коем случае не из базы данных, и используйте отдельный ключ для каждого окружения, чтобы утечка со staging не сожгла кредиты с продакшена.

  3. Шаг 3

    Отправляйте задачу через воркер

    Views должны отвечать мгновенно. Таска Celery отправляет POST-запрос с задачей и записывает полученный id в строку базы данных.

  4. Шаг 4

    Опрашивайте с экспоненциальной задержкой

    Выполняйте polling каждые 3 секунды, пока статус не сменится на completed или failed. При получении 429 подождите указанное в Retry-After количество секунд вместо немедленного повтора.

  5. Шаг 5

    Сохраняйте файл

    Скачайте готовое изображение в собственное хранилище. URL возвращает результат одной задачи; ваш бакет будет отдавать картинку на странице товара годами.

  6. Шаг 6

    Отдавайте статус клиенту

    Ваш фронтенд опрашивает вашу запись в базе, а не Photta API. Это защищает ключ на сервере и позволяет отображать прогресс в любом удобном виде.

Модель, таска, ошибки, хранилище

Логика отправки с polling и обработка ошибок взяты напрямую из документации 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"])

Частые вопросы разработчиков

Вопросы Django-разработчиков

Зачем нужен воркер вместо view?

Генерация длится дольше допустимого времени HTTP-запроса. Воркер продолжит работу при закрытии вкладки, деплое и таймауте; view прервется в любом из этих случаев.

Как долго таска должна выполнять polling?

Достаточно долго с учетом документированного времени ответа эндпоинта, но с фиксированным лимитом. Примеры в документации прекращают опрос после заданного числа попыток, избегая бесконечного цикла.

Можно ли использовать асинхронный Django вместо Celery?

Да — все вызовы представляют собой обычный HTTP. Главное — чтобы выполнение задачи не зависело от жизненного цикла входящего запроса, независимо от архитектуры.

Как тестировать интеграцию без списания кредитов?

Запишите один реальный ответ и воспроизводите его в тестах. Эндпоинты ресурсов и анализ детальных фото бесплатны, поэтому проверка ключа и сетевой доступности не расходует кредиты.

Можно ли использовать один ключ для всех окружений?

Можно, но все ключи списывают один и тот же баланс, поэтому зациклившийся тест на staging обнулит кредиты продакшена. Раздельные ключи изолируют риски и позволяют быстро локализовать проблему.

Связанные страницы

Материалы по теме

Все показатели на этой странице взяты из документации API Photta · проверено

Воркер, запись в БД и файл

Логика отправки и polling на этой странице взята напрямую из документации API.

Получить API-ключОзнакомьтесь с быстрым стартом