Midjourney API Python 사용법: 이미지 생성과 저장

export OMNIAKEY_API_KEY="your-key"
python3 midjourney_grid.py

{
  "model": "midjourney",
  "type": "image",
  "operation": "generate",
  "input": {
    "prompt": "A small red paper boat floating on a quiet blue lake at sunrise, cinematic lighting, elegant composition, no text"
  },
  "parameters": {"aspectRatio": "1:1", "quality": "1", "n": 1}
}

import hashlib
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path

API = "https://api.omniakey.com/v1"
KEY = os.environ["OMNIAKEY_API_KEY"]
STATE = Path("midjourney-task.json")
OUT = Path("midjourney-grid.png")
PROMPT = ("A small red paper boat floating on a quiet blue lake at sunrise, "
          "cinematic lighting, elegant composition, no text")


def save(state):
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2))
    tmp.replace(STATE)


def read_json(response):
    raw = response.read()
    try:
        return json.loads(raw)
    except ValueError:
        return {"raw": raw[:300].decode("utf-8", "replace")}


def call(method, path, body=None, headers=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(
        API + path, data=data, method=method,
        headers={"Authorization": f"Bearer {KEY}", **(headers or {})})
    if body is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            return response.status, read_json(response)
    except urllib.error.HTTPError as error:
        return error.code, read_json(error)
    except (urllib.error.URLError, TimeoutError) as error:
        return None, {"error": str(error)}


def main():
    state = json.loads(STATE.read_text()) if STATE.exists() else None
    if state is None:
        body = {
            "model": "midjourney", "type": "image", "operation": "generate",
            "input": {"prompt": PROMPT},
            "parameters": {"aspectRatio": "1:1", "quality": "1", "n": 1},
        }
        state = {"idempotency_key": str(uuid.uuid4()), "body": body,
                 "task_id": None, "status": "creating"}
        save(state)  # no API key here; write before POST
    elif state["body"]["input"]["prompt"] != PROMPT:
        sys.exit("The state file holds a different prompt. Move it aside to start a new job.")

    if OUT.exists() and state.get("status") == "downloaded":
        saved_sha = state.get("download_sha")
        if not saved_sha or hashlib.sha256(OUT.read_bytes()).hexdigest() == saved_sha:
            print("already saved", OUT)
            return

    if not state["task_id"]:
        code, task = call("POST", "/media/tasks", state["body"],
                          {"Idempotency-Key": state["idempotency_key"]})
        if code is None:
            sys.exit(f"Create not confirmed: {task['error']}. Rerun to retry the same key and body.")
        if code == 409:
            sys.exit("409 idempotency_key_reused: same key, different body. Restore the saved body.")
        if code == 503:
            sys.exit("503 no_available_channel: check model access for this key/account; "
                     "contact support with the request id if it persists.")
        if code not in (200, 202) or "id" not in task:
            sys.exit(f"Create failed: HTTP {code} {task}")
        state.update(task_id=task["id"], status=task["status"])
        save(state)

    deadline = time.monotonic() + 900
    while True:
        code, task = call("GET", f"/media/tasks/{state['task_id']}")
        if code is None:
            sys.exit(f"Polling stopped: {task['error']}. The job is still saved; rerun this script.")
        if code != 200 or "status" not in task:
            sys.exit(f"Poll failed: HTTP {code} {task}")
        state["status"] = task["status"]
        save(state)
        if task["status"] in ("failed", "canceled"):
            sys.exit(f"Job {task['status']}: {task.get('error')}")
        if task["status"] == "succeeded":
            break
        if time.monotonic() > deadline:
            sys.exit(f"Still {task['status']} after 900 s. Rerun with the same state file.")
        time.sleep(min(8, 2 + task.get("progress", {}).get("percent", 0) // 20))

    asset = task["assets"][0]
    if asset.get("mime_type") != "image/png":
        sys.exit(f"Expected a PNG grid, got {asset.get('mime_type')}")
    try:
        with urllib.request.urlopen(asset["url"], timeout=120) as response:
            data = response.read()  # signed URL: no Authorization header
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as error:
        sys.exit(f"Download failed: {error}. Rerun to get a fresh signed URL and retry.")
    digest = hashlib.sha256(data).hexdigest()
    if asset.get("sha256") and digest != asset["sha256"]:
        sys.exit("SHA-256 mismatch. Rerun to download the asset again.")
    part = OUT.with_suffix(".part")
    part.write_bytes(data)
    part.replace(OUT)
    state.update(status="downloaded", download_sha=digest)
    save(state)
    print("saved", OUT, len(data), "bytes")


if __name__ == "__main__":
    main()

{
  "id": "task_DbCWJPfPL1jyhSD38IihNHiiFQao8hjV",
  "status": "succeeded",
  "progress": {"percent": 100, "stage": "completed"},
  "assets": [
    {
      "kind": "image",
      "mime_type": "image/png",
      "width": 2048,
      "height": 2048,
      "byte_size": 5834301,
      "sha256": "f503cef038d15fa150b36b5fe7c41874febad3aa1167698d790781e47a4a10b9",
      "expires_at": "2026-11-12T04:16:41Z",
      "url": "https://…redacted-signed-url…"
    }
  ]
}