애플리케이션 연동

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

def validate_paths(source_id, target_id, source_paths, target_path):
    # sourceItems reads filePath, not path
    return api("POST", "/api/transfers/validate-path", {
        "sourceId": source_id,
        "targetId": target_id,
        "sourceItems": [{"filePath": p} for p in source_paths],
        "targetPath": target_path,
    }) or {}


result = validate_paths("device-a", "device-b",
                        ["/data/report.pdf"], "/archive")

if result.get("invalidSourcePaths"):
    raise ValueError(f"missing source paths: {result['invalidSourcePaths']}")

if result.get("validTargetPath") is False:
    raise ValueError("target path not found")

def create_transfer(source_id, target_id, source_paths, target_path,
                    action="numbering"):
    transfer = api("POST", "/api/transfers/manual", {
        "sourceDevice": source_id,
        "targetDevice": target_id,
        "targetPath": target_path,
        "sourceItem": [{"path": p, "isDir": False} for p in source_paths],
        "sendAllFolder": False,
        "transferOptions": {"target-action": action},
    })

    return transfer["monitorId"]

api("POST", "/api/transfers/manual", {
    "sourceDevice": source_id,
    "targetDevice": target_id,
    "targetPath": target_path,
    "sourcePaths": ["/data/reports"],
    "sendAllFolder": True,
    "transferOptions": {"target-action": "numbering"},
})

def start_order_transfer(order_id, source_id, target_id, paths, target_path):
    validate_paths(source_id, target_id, paths, target_path)

    monitor_id = create_transfer(source_id, target_id, paths, target_path)

    db.execute(
        "UPDATE orders SET monitor_id = %s, transfer_state = %s WHERE id = %s",
        (monitor_id, "transferring", order_id),
    )

    return monitor_id

import time


def describe(monitor_id):
    return api("GET", f"/api/transfers/{monitor_id}")


def wait(monitor_id, timeout=1800, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = describe(monitor_id)

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


detail = describe(monitor_id)

print(detail["statusLabel"], detail["percent"], "%")
print(detail["transferSize"], "/", detail["totalSize"])

PAUSED = 3
RUNNING_STATES = {1, 6, 12, 13}
CANCELLED = 5


def control(monitor_id, action, tries=10):
    api("POST", f"/api/transfers/{monitor_id}/{action}", {})

    expected = {
        "pause": {PAUSED},
        "resume": RUNNING_STATES,
        "cancel": {CANCELLED},
    }[action]

    for _ in range(tries):
        time.sleep(1)
        detail = describe(monitor_id)

        if detail.get("status") in expected:
            return detail

    return describe(monitor_id)

result = api("POST", "/api/transfers/bulk-cancel", {"monitorIds": monitor_ids})

print(result.get("cancelled"), result.get("failed"))

def finalize(order_id, monitor_id):
    detail = describe(monitor_id)

    if not is_terminal(detail):
        return None

    status = detail["status"]
    succeeded = status == STATUS_COMPLETE

    db.execute(
        "UPDATE orders SET transfer_state = %s, transfer_status = %s WHERE id = %s",
        ("done" if succeeded else "failed", status, order_id),
    )

    return succeeded

def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

config = api("GET", f"/api/transfers/{monitor_id}/replay-data")
api("POST", f"/api/transfers/{monitor_id}/replay", {"action": "replay"})