구현 가이드 · Django

Django 워커에서 try-on 이미지 렌더링하기

한 줄 요약

Django 연동 방식은 작업당 1개의 행을 저장하고, Celery 작업에서 요청을 제출하며, 상태가 completed가 될 때까지 3초마다 폴링한 다음 자체 스토리지로 파일을 다운로드합니다.

  • 크레딧: 2K에서 5 크레딧 · 4K에서 6 크레딧
  • 일반적인 소요 시간: 1.5–4 min
  • 처리 방식: 비동기, 폴링

이 페이지의 모든 수치는 다음 출처를 기반으로 합니다: Photta API 문서 · 확인 완료

작업을 반영하는 모델, 제출 및 polling을 담당하는 Celery 태스크, 전달 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')

동작 원리

행 1개, 작업 1개, 파일 1개

작업 모델링, 백그라운드 작업에서 제출, 백오프를 적용한 폴링, 파일 영속화, 상태 노출.

  1. 단계 1

    작업 모델링

    생성 건당 행 1개: 상품, 생성 id, 상태, output_url. 뷰는 이 행을 읽고, 워커는 API를 읽습니다.

  2. 단계 2

    Photta API 키 발급받기

    데이터베이스가 아닌 settings의 환경 변수에서 읽어오고, 환경별로 별도의 키를 사용하여 스테이징 환경의 누출로 프로덕션 credits가 소진되지 않도록 하세요.

  3. 단계 3

    백그라운드 작업에서 제출

    뷰는 즉시 반환되어야 합니다. Celery 작업이 작업을 POST하고 반환된 id를 해당 행에 기록합니다.

  4. 단계 4

    백오프를 적용한 폴링

    상태가 completed 또는 failed가 될 때까지 3초마다 폴링하세요. 429 응답을 받으면 즉시 재시도하는 대신 Retry-After 초만큼 대기하세요.

  5. 단계 5

    파일 영속화

    완성된 이미지를 자체 스토리지로 다운로드하세요. URL은 단일 작업을 제공할 뿐이며, 수년간 상품 페이지에 이미지를 서빙하는 것은 여러분의 버킷입니다.

  6. 단계 6

    상태 노출

    프론트엔드는 Photta API가 아닌 데이터베이스 행을 폴링합니다. 이렇게 하면 키를 서버에 안전하게 보관하고 진행 상황을 원하는 방식으로 표시할 수 있습니다.

모델, 작업, 에러, 스토리지

제출 및 폴링, 에러 처리는 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 개발자들이 자주 묻는 질문

뷰 대신 워커를 사용하는 이유는 무엇인가요?

생성 작업은 일반적인 HTTP 요청 시간보다 오래 걸립니다. 브라우저 탭이 닫히거나, 배포가 진행되거나, 타임아웃이 발생해도 워커는 안전하게 유지되지만 뷰는 그렇지 못합니다.

작업이 얼마나 오래 폴링해야 하나요?

엔드포인트 문서에 명시된 레이턴시를 고려해 상한을 두고 충분한 시간 동안 폴링해야 합니다. 여기에 제공된 예제는 무한정 폴링하는 대신 정해진 횟수만큼 시도한 후 중단합니다.

Celery 대신 비동기 Django를 사용할 수 있나요?

네 — 호출 자체는 단순한 HTTP 요청입니다. 어떤 방식을 선택하든 중요한 것은 작업의 수명이 요청 수명보다 오래 유지되어야 한다는 점입니다.

credits를 소진하지 않고 테스트하려면 어떻게 해야 하나요?

실제 응답 1개를 기록한 후 테스트에서 재사용하세요. 리소스 엔드포인트와 상세 사진 분석은 무료이므로 키 유효성 검사 및 연결 확인에는 비용이 전혀 들지 않습니다.

하나의 키를 모든 환경에서 사용할 수 있나요?

가능하지만, 모든 키가 동일한 잔액을 사용하므로 스테이징의 무한 루프로 인해 프로덕션 credits가 고갈될 수 있습니다. 키를 분리하면 피해 범위를 명확히 파악하고 신속하게 차단할 수 있습니다.

워커, 행, 그리고 파일

이 페이지의 제출 및 폴링 코드는 API 문서를 바탕으로 구현되었습니다.