実装ガイド · Django

Djangoワーカーからtry-on画像をレンダリングする

要約

Django連携では、ジョブごとに1行を保存し、Celeryタスクから送信し、ステータスがcompletedになるまで3秒ごとにポーリングして、ファイルを独自のストレージにダウンロードします。

APIキーを取得try-on endpointを見る
  • Credits: 2Kで5 credits · 4Kで6 credits
  • 所要時間の目安: 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、ステータス、そして出力URL。ビューが読み取るのはその行であり、ワーカーが読み取るのはAPIです。

  2. ステップ 2

    Photta APIキーの取得

    データベースからではなく、設定の環境変数から読み取ってください。また、ステージング環境の漏洩によって本番環境のcreditsが消費されないよう、環境ごとに個別のキーを使用してください。

  3. ステップ 3

    タスク内での送信

    ビューは即座に応答を返す必要があります。CeleryタスクがジョブをPOSTし、返されたIDを行に書き込みます。

  4. ステップ 4

    バックオフ付きでポーリング

    ステータスがcompletedまたはfailedになるまで3秒ごとにポーリングします。429が発生した場合は、すぐに再試行せずRetry-After秒待機してください。

  5. ステップ 5

    ファイルの永続化

    完成した画像を独自のストレージにダウンロードします。URLが提供するのは1回分のジョブですが、製品ページに何年も配信し続けるのはあなたのバケットです。

  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開発者からのよくある質問

なぜビューではなくワーカーなのか？

生成には、リクエストが許容する以上の時間がかかります。ワーカーはブラウザのタブ閉じ、デプロイ、タイムアウトが発生しても存続しますが、ビューはそのどれにも耐えられません。

タスクはどのくらいの時間ポーリングすべきか？

上限を設けた上で、endpointのドキュメントに記載されたレイテンシに十分な長さです。ここでのサンプルは、永久にポーリングし続けるのではなく、一定の試行回数後に停止します。

Celeryの代わりに非同期Djangoを使用できますか？

はい — 呼び出しは単純なHTTPです。どのような構成にするにせよ、重要なのは処理がリクエストのライフサイクルを超えて存続することです。

creditsを消費せずにテストするには？

実際のリクエスト応答を1つ記録し、テストでそれを再生してください。リソースendpointと詳細写真の分析は無料であるため、キーの確認や接続チェックには費用がかかりません。

1つのキーをすべての環境で共用できますか？

共用自体は可能ですが、すべてのキーで同じ残高が消費されるため、ステージングの無限ループによって本番のcreditsが枯渇します。環境ごとにキーを分けることで、影響を可視化し、停止を容易にできます。

関連ページ

関連ドキュメント

このページに記載されている数値はすべて以下に基づきます: Photta APIドキュメント · 確認済み

ワーカー、行、そしてファイル

このページの送信とポーリングは、APIドキュメントに基づいて作成されています。

APIキーを取得クイックスタートを見る