모범 사례

import time, httpx

def wait_task(task_id, timeout=300):
    deadline = time.time() + timeout
    while time.time() < deadline:
        resp = httpx.get(f"{HOST}/v1/midjourney/{task_id}",
                         headers={"Authorization": f"Bearer {API_KEY}"}).json()
        if resp["status"] in ("SUCCESS", "FAILURE"):
            return resp
        if resp["status"] == "MODAL":
            raise RuntimeError(f"task {task_id} 작업을 완료하려면 /modal 호출이 필요합니다")
        time.sleep(3)
    raise TimeoutError(task_id)

a serene mountain lake at sunrise, photorealistic, soft golden light,
mist rising from water, snow-capped peaks in distance --ar 16:9 --v 8.1 --s 100

import time, random, httpx

def submit_with_retry(payload, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            r = httpx.post(f"{HOST}/v1/midjourney/generations/imagine",
                           json=payload,
                           headers={"Authorization": f"Bearer {API_KEY}"},
                           timeout=30)
            data = r.json()
            if r.status_code == 200 and data["code"] in (1, 200):
                return data
            if data["code"] in (4, 24):
                raise ValueError(data["description"])      # 재시도 불가
            if data["code"] == 3 and "task" in data["description"]:
                raise ValueError(data["description"])      # task_id 미존재
            # 그 외(9 / 429 / 5xx)는 재시도 가능
        except httpx.RequestError:
            pass
        time.sleep((4 ** attempt) + random.uniform(0, 1))  # 1s / 4s / 16s ...
    raise RuntimeError(f"최대 재시도 횟수 {max_attempts}에 도달")

# imagine → 폴링 → upscale
imagine_id = submit({"prompt": "a cat"})["data"][0]["task_id"]
result = wait_task(imagine_id)           # grid_image_url + 이미지 4장 image_urls + buttons
upscale_id = submit_to("/upscale", {"task_id": imagine_id, "index": 2})["data"][0]["task_id"]
final = wait_task(upscale_id)            # upscale 로컬 합성, 1~2s
single_image = final["image_urls"][0]

imagine_id = submit({"prompt": "a portrait"})["data"][0]["task_id"]; wait_task(imagine_id)
upscale_id = submit_to("/upscale", {"task_id": imagine_id, "index": 1})["data"][0]["task_id"]; wait_task(upscale_id)

inpaint_id = submit_to("/inpaint", {"task_id": upscale_id})["data"][0]["task_id"]  # status=modal
# 프런트엔드에서 mask 그리기(투명=리페인트 영역), 자체 OSS에 업로드하여 mask_url 획득
final = submit_to("/modal", {
    "task_id": inpaint_id,
    "prompt": "replace the eyes with cybernetic blue eyes",
    "mask_url": "https://your-oss.com/mask.png"
})
wait_task(final["data"][0]["task_id"])

import asyncio
sem = asyncio.Semaphore(10)  # 클라이언트 최대 10개 동시 제출

async def submit_one(prompt):
    async with sem:
        return await submit({"prompt": prompt})