Build · Django

Render ảnh try-on từ một Django worker

Tóm tắt trong một câu

Một bản tích hợp Django lưu một hàng cho mỗi job, gửi yêu cầu từ Celery task, polling mỗi 3 giây cho đến khi trạng thái là completed, sau đó tải file về bộ lưu trữ của riêng bạn.

  • Credits: 5 credits ở 2K · 6 credits ở 4K
  • Thời gian điển hình: 1.5–4 min
  • Xử lý: async, polling

Mọi số liệu trên trang này đều lấy từ tài liệu Photta API · đã kiểm tra

Một model ánh xạ job, một Celery task chịu trách nhiệm gửi và polling, cùng bộ lưu trữ giữ lại file sau khi URL phân phối hết hạn.

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

Cơ chế hoạt động

Một hàng, một task, một file

Tạo model cho job, gửi từ task, polling kèm backoff, lưu trữ file, hiển thị trạng thái.

  1. Bước 1

    Tạo model cho job

    Một hàng cho mỗi lượt tạo: sản phẩm, generation id, trạng thái và output URL. Hàng này là dữ liệu views đọc; API là nơi worker đọc.

  2. Bước 2

    Lấy Photta API key

    Đọc key từ biến môi trường trong settings, không bao giờ đọc từ cơ sở dữ liệu, và dùng key riêng cho từng môi trường để việc rò rỉ ở staging không làm tiêu hao credits của production.

  3. Bước 3

    Gửi yêu cầu trong task

    Views cần trả về phản hồi ngay lập tức. Celery task thực hiện POST job và ghi id nhận được vào hàng dữ liệu.

  4. Bước 4

    Polling kèm backoff

    Polling mỗi 3 giây cho đến khi trạng thái là completed hoặc failed. Khi gặp lỗi 429, hãy chờ số giây theo Retry-After thay vì thử lại ngay lập tức.

  5. Bước 5

    Lưu trữ file

    Tải ảnh hoàn thiện về bộ lưu trữ của riêng bạn. URL chỉ phục vụ cho một job; bucket của bạn mới là nơi phân phối cho trang sản phẩm trong nhiều năm.

  6. Bước 6

    Hiển thị trạng thái

    Front-end của bạn polling hàng dữ liệu nội bộ, không phải Photta API. Cách này giúp giữ key an toàn trên server và cho phép bạn hiển thị tiến trình theo bất kỳ cách nào tùy thích.

Model, task, lỗi, lưu trữ

Quy trình submit-and-poll và xử lý lỗi được xây dựng dựa trên tài liệu 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"])

Câu hỏi thường gặp từ developer

Câu hỏi lập trình viên Django thường gặp

Tại sao nên dùng worker thay vì view?

Một lượt tạo tốn nhiều thời gian hơn mức cho phép của một request. Worker vẫn duy trì khi người dùng đóng tab trình duyệt, khi deploy hoặc khi gặp timeout; còn view thì không.

Task nên polling trong bao lâu?

Đủ lâu theo độ trễ được ghi nhận của endpoint, kèm theo một giới hạn trần. Các ví dụ ở đây sẽ dừng lại sau một số lần thử cố định thay vì polling vô hạn.

Tôi có thể dùng async Django thay cho Celery không?

Có — các lệnh gọi chỉ là HTTP thuần túy. Điều quan trọng là tác vụ phải tồn tại độc lập ngoài vòng đời request, bất kể bạn triển khai theo cách nào.

Làm thế nào để kiểm thử mà không tốn credits?

Ghi lại một response thực tế và replay trong các bài test. Các resource endpoint và tính năng phân tích ảnh chi tiết hoàn toàn miễn phí, nên việc kiểm tra key và kết nối không tốn chi phí.

Một key có thể dùng cho mọi môi trường không?

Được, nhưng mọi key đều trừ vào cùng một số dư, do đó một vòng lặp lỗi ở staging sẽ làm cạn kiệt credits của production. Sử dụng các key riêng biệt giúp dễ nhận biết thiệt hại và chặn kịp thời.

Một worker, một hàng và một file

Quy trình submit-and-poll trên trang này được xây dựng dựa trên tài liệu API.