构建指南 · Django

通过 Django worker 渲染 try-on 图片

一句话总结

Django 集成方案：每个任务存储一行数据，通过 Celery 任务提交，每 3 秒轮询一次直至状态变为 completed，随后将文件下载至你自有的存储中。

获取 API Key查阅 try-on endpoint
  • Credits: 2K 分辨率 5 credits · 4K 分辨率 6 credits
  • 常规耗时: 1.5–4 min
  • 处理方式: 异步，轮询

本页所有数据均来自 Photta API 文档 · 已验证

一个与任务状态映射的 Model，一个负责提交与轮询的 Celery task，以及在交付 URL 失效后用于持久保存文件的 storage。

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

    对任务建模

    每次生成对应一行记录：商品、生成 ID、状态以及输出 URL。视图只读取该行记录；worker 负责与 API 通信。

  2. 步骤 2

    获取 Photta API key

    在 settings 中从环境变量读取，切勿存入数据库；并且每个环境使用独立的 key，防止预发布环境泄露消耗生产环境的 credits。

  3. 步骤 3

    在任务中提交

    视图应当立即返回响应。Celery 任务负责 POST 提交任务，并将返回的 ID 写入数据行。

  4. 步骤 4

    带退避机制的轮询

    每 3 秒轮询一次，直至状态为 completed 或 failed。若遇到 429，等待 Retry-After 指定的秒数，切勿立即重试。

  5. 步骤 5

    持久化文件

    将生成的图片下载至你自有的存储中。URL 仅用于单次交付任务；而你的 bucket 才是未来数年为商品详情页提供服务的地方。

  6. 步骤 6

    暴露状态

    前端轮询你自己的数据行，而非 Photta API。这样既能将 key 保留在服务端，又能让你按需自由展示进度。

模型、任务、错误处理、存储

提交与轮询以及错误处理逻辑均根据 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 开发者常见问题

为什么使用 worker 而不是视图？

单次生成耗时远超普通请求所允许的长度。Worker 能够耐受浏览器标签页关闭、服务部署和超时中断；而视图在这些情况下都会直接失败。

任务应该轮询多久？

足够覆盖 endpoint 文档中标明的延迟即可，同时设置上限。此处的示例代码在达到固定尝试次数后即停止，而非无休止轮询。

我可以用异步 Django 代替 Celery 吗？

可以——底层调用都是标准 HTTP。关键在于无论如何组织，该任务的生命周期都必须独立于当前请求。

如何在不消耗 credits 的情况下进行测试？

录制一次真实的响应并在测试中回放。资源类 endpoint 和细节图分析均免费，因此 key 验证和网络连通性检查不消耗任何费用。

所有环境可以共用一个 key 吗？

可以，但所有 key 扣减的是同一个账户余额，预发布环境的死循环会耗尽生产环境的 credits。隔离 key 能让损失一目了然且易于及时止损。

相关页面

相关阅读

本页所有数据均来自 Photta API 文档 · 已验证

一个 worker、一行记录和一个文件

本页中的提交与轮询逻辑均根据 API 文档生成。

获取 API Key阅读快速入门