REST API 로 자동화하기

화면에서 하는 일은 모두 /api/v1/… REST API 로도 할 수 있습니다. 화면도 같은 API 를 씁니다. 이 페이지는 모든 API 에 공통인 규칙만 다루고, 엔드포인트별 요청·응답은 레퍼런스 로 넘깁니다. 서버의 OpenAPI 문서는 /openapi.json, 대화형 문서는 /docs 에도 있습니다.

예시는 모두 촬영 스택(http://localhost:10000, 테넌트 DEMO, 계정 [email protected])에 실제로 실행한 것입니다. 여러분 환경에서는 주소와 계정만 바꾸세요.

인증: 쿠키 또는 JWT#

로그인은 같은 계정·비밀번호로 두 가지 방식이 있습니다. 둘 다 폼 필드 username(메일)과 password 를 받습니다.

• 세션 수명은 서버 설정 GEO_MLOPS_AUTH_TOKEN_LIFETIME(초)이 정합니다. 기본값 0 은 만료 없음입니다. 운영에서는 적당한 수명을 두고, 자동화에는 전용 계정을 쓰세요.
• 로그아웃은 POST /auth/cookie/logout 또는 POST /auth/jwt/logout 입니다.
• MLflow 토큰과 컨테이너 토큰은 REST API 에 쓰지 못합니다. 각각 /mlflow, /v2 전용입니다.

테넌트 지정: X-Tenant#

한 계정이 여러 테넌트에 속할 수 있으므로, 테넌트 범위의 API 는 요청마다 어느 테넌트인지 알려야 합니다. X-Tenant 헤더 또는 ?tenant= 쿼리입니다(대소문자 무관, 헤더가 우선).

curl -sS -b cookies.txt "$API/api/v1/datasets?page_size=2"
# {"error_id":"fb61…","code":"bad_request","message":"tenant context required (X-Tenant header or ?tenant=)","detail":null}

curl -sS -b cookies.txt -H "X-Tenant: DEMO" "$API/api/v1/datasets?page_size=2"
# {"items":[…],"total":9,"page":1,"page_size":2}

권한은 그 테넌트에서의 역할로 판정합니다. 역할이 모자라면 403, 속하지 않은 테넌트의 자원은 404 입니다. /users/me, /api/v1/stream/notifications 처럼 사람 단위인 API 는 테넌트가 필요 없습니다.

오류 형식#

실패는 모두 같은 모양입니다.

{"error_id": "7bea34c0…", "code": "bad_request", "message": "unknown sort 'bogus'",
 "detail": {"allowed": ["created_at", "name", "records", "size", "validation"]}}

페이지네이션 두 가지#

번호 페이지 (대부분의 목록)#

page(1부터)와 page_size 를 보내면 {items, total, page, page_size} 가 옵니다. page_size 를 빼면 서버 기본값을 씁니다. 목록 대부분은 검색·정렬도 같은 이름으로 받습니다.

curl -sS -b cookies.txt -H "X-Tenant: DEMO" \
  "$API/api/v1/datasets?page=2&page_size=2&sort=name&order=asc"
# {"items":[…2개…],"total":9,"page":2,"page_size":2}

커서 (시간순 피드)#

엣지 디바이스의 로그 · 텔레메트리 · 추론 기록처럼 최신순으로 계속 쌓이는 피드는 커서를 씁니다. limit 과 cursor 를 보내고, 응답의 next_cursor 를 다음 요청의 cursor 로 넘깁니다. next_cursor 가 null 이면 끝입니다. 이 피드의 total 은 전체 개수가 아니라 이번 응답의 항목 수입니다.

curl -sS -b cookies.txt -H "X-Tenant: DEMO" \
  "$API/api/v1/edge/devices/edge-demo-01/logs?limit=2"
# {"items":[…2개…],"total":2,"next_cursor":"Mg=="}
curl -sS -b cookies.txt -H "X-Tenant: DEMO" \
  "$API/api/v1/edge/devices/edge-demo-01/logs?limit=2&cursor=Mg=="
# {"items":[…],"total":2,"next_cursor":"NA=="}

커서 문자열은 해석하거나 직접 만들지 말고 받은 그대로 넘기세요. 모양은 바뀔 수 있습니다.

실시간 스트림: SSE#

진행 상황과 알림은 Server-Sent Events(text/event-stream)로 받습니다. 서버가 연결을 열어 둔 채 이벤트를 계속 보내 주는 방식입니다. 연결하면 먼저 connected 이벤트가 오고, 그 뒤로 일이 생길 때마다 이벤트가 옵니다.

curl -sS -N -b cookies.txt -H "X-Tenant: DEMO" "$API/api/v1/stream/alerts"
# event: connected
# data: {"channel": "alerts:DEMO"}

• 지난 이벤트는 다시 보내지 않습니다. 먼저 REST 로 현재 상태를 읽고, 그 뒤 스트림을 이어 붙이세요. 학습 곡선이라면 지표 API 로 전체를 받고 metrics 증분을 덧붙입니다.
• 이벤트 본문은 "무엇이 바뀌었는지" 정도만 담습니다. 자세한 내용은 REST 로 다시 읽습니다.
• 브라우저의 EventSource 는 헤더를 붙일 수 없습니다. 쿠키로 로그인한 뒤 테넌트는 ?tenant=DEMO 쿼리로 넘기세요.
• 끊기면 다시 연결하면 됩니다. 앞단 프록시가 긴 연결을 자르지 않도록(버퍼링 끄기, 시간 제한 늘리기) 운영자가 설정해야 합니다.

Python 예시: 로그인 · 페이지 훑기 · SSE#

import os
import requests

API = os.environ.get("API", "http://localhost:10000")

# 1) 로그인: JWT 를 받아 Authorization 헤더로 쓴다
r = requests.post(f"{API}/auth/jwt/login",
                  data={"username": os.environ["EMAIL"], "password": os.environ["PASSWORD"]})
r.raise_for_status()
s = requests.Session()
s.headers["Authorization"] = f"Bearer {r.json()['access_token']}"
s.headers["X-Tenant"] = "DEMO"          # 테넌트 범위 API 는 모두 필요

# 2) 번호 페이지네이션: 끝까지 훑기
page, names = 1, []
while True:
    body = s.get(f"{API}/api/v1/datasets",
                 params={"page": page, "page_size": 50, "sort": "name"}).json()
    names += [d["name"] for d in body["items"]]
    if page * body["page_size"] >= body["total"]:
        break
    page += 1
print(len(names), "datasets")

# 3) SSE: 이벤트 몇 개만 읽고 닫는다
with s.get(f"{API}/api/v1/stream/alerts", stream=True, timeout=(5, 30)) as resp:
    event = None
    for line in resp.iter_lines(decode_unicode=True):
        if line.startswith("event:"):
            event = line.split(":", 1)[1].strip()
        elif line.startswith("data:"):
            print(event, line.split(":", 1)[1].strip())
            break                      # 예시이므로 첫 이벤트에서 멈춘다

9 datasets
connected {"channel": "alerts:DEMO"}

청크 업로드#

큰 파일은 한 요청으로 보내지 않습니다. 앞단 프록시의 본문 크기·응답 시간 제한에 걸리기 때문입니다. 플랫폼은 세션 열기 → 청크 보내기 → 마감이라는 같은 틀을 두 곳에서 씁니다. 마감은 202 로 바로 돌아오고, 조립·검증은 서버가 이어서 하므로 상태를 다시 읽어 끝을 확인합니다.

청크 크기는 서버가 정해 돌려준 값을 쓰세요. 그보다 큰 청크는 413 입니다.

"""파일 하나를 청크로 나눠 데이터셋에 올린다 (REST API 예시)."""

import hashlib
import os
import sys
import time

import requests

API = os.environ.get("API", "http://localhost:10000")
TENANT = os.environ.get("TENANT", "DEMO")
dataset_id, path = sys.argv[1], sys.argv[2]
size = os.path.getsize(path)
sha256 = hashlib.sha256(open(path, "rb").read()).hexdigest()

s = requests.Session()
s.headers["X-Tenant"] = TENANT
s.post(f"{API}/auth/cookie/login",
       data={"username": os.environ["EMAIL"],
             "password": os.environ["PASSWORD"]}).raise_for_status()

# 1) 세션 열기: chunk_size 는 서버가 정해서 돌려준다
r = s.post(f"{API}/api/v1/datasets/{dataset_id}/uploads",
           json={"filename": os.path.basename(path), "size": size, "sha256": sha256})
r.raise_for_status()
up = r.json()
upload_id, chunk = up["upload_id"], up["chunk_size"]

# 2) 청크 보내기: 인덱스로 보내므로 순서가 달라도, 같은 청크를 또 보내도 된다
with open(path, "rb") as f:
    index = 0
    while data := f.read(chunk):
        s.put(f"{API}/api/v1/datasets/{dataset_id}/uploads/{upload_id}/chunks/{index}",
              data=data,
              headers={"Content-Type": "application/octet-stream"}).raise_for_status()
        index += 1

# 3) 마감: 202 로 즉시 돌아오고, 조립·검증은 서버가 이어서 한다
s.post(f"{API}/api/v1/datasets/{dataset_id}/uploads/{upload_id}:complete").raise_for_status()
while True:
    st = s.get(f"{API}/api/v1/datasets/{dataset_id}/uploads/{upload_id}").json()
    if st["state"] in ("done", "failed"):
        print(st["state"], st.get("result") or st.get("error"))
        break
    time.sleep(1)

EMAIL=[email protected] PASSWORD='<your-password>' \
  python3 upload_file.py ds-d3dd6730ad13 20260721_line3_0002.png
# done {'created': 1, 'skipped': 0, 'file_ids': ['b2597c97-…']}

"""이미지 tar 를 청크로 나눠 레지스트리에 반입한다 (REST API 예시)."""

import os
import sys
import time

import requests

API = os.environ.get("API", "http://localhost:10000")
TENANT = os.environ.get("TENANT", "DEMO")
path = sys.argv[1]
size = os.path.getsize(path)

s = requests.Session()
s.headers["X-Tenant"] = TENANT
# 1) 로그인: 쿠키 세션 (JWT 를 쓰려면 /auth/jwt/login 후 Authorization 헤더)
r = s.post(f"{API}/auth/cookie/login",
           data={"username": os.environ["EMAIL"], "password": os.environ["PASSWORD"]})
r.raise_for_status()

# 2) 반입 세션 만들기: 전체 크기를 먼저 알린다
r = s.post(f"{API}/api/v1/registry/imports",
           json={"size_bytes": size, "filename": os.path.basename(path)})
r.raise_for_status()
job = r.json()
chunk = job["chunk_size"]
print("import", job["id"], "chunk", chunk)

# 3) 청크를 순서대로 보낸다: Content-Range 로 위치를 알린다
with open(path, "rb") as f:
    offset = 0
    while offset < size:
        data = f.read(chunk)
        end = offset + len(data) - 1
        r = s.patch(
            f"{API}/api/v1/registry/imports/{job['id']}/chunks",
            data=data,
            headers={
                "Content-Type": "application/octet-stream",
                "Content-Range": f"bytes {offset}-{end}/{size}",
            },
        )
        if r.status_code == 416:  # 서버가 가진 위치에서 다시 시작
            offset = r.json()["detail"]["offset"]
            f.seek(offset)
            continue
        r.raise_for_status()
        offset = r.json()["received_bytes"]
        print(f"\r{offset}/{size}", end="", flush=True)
print()

# 4) 반입 시작: 즉시 돌아오고 등록은 서버가 이어서 한다
s.post(f"{API}/api/v1/registry/imports/{job['id']}:start").raise_for_status()
while True:
    row = s.get(f"{API}/api/v1/registry/imports/{job['id']}").json()
    print(row["status"], row["phase"])
    if row["status"] in ("READY", "FAILED", "CANCELED"):
        print(row["image"] or row["error"])
        break
    time.sleep(3)

EMAIL=[email protected] PASSWORD='<your-password>' \
  python3 import_image.py mnist-trainer-v1.tar.gz
# import <반입 id> chunk 33554432
# 521930649/521930649
# RUNNING assemble
# RUNNING inspect
# READY done
# localhost:10000/demo/example/mnist-trainer:v1

더 보기#

• 엔드포인트 전체 목록: 레퍼런스
• 학습 컨테이너 반입과 변형 등록의 화면 절차: 플랫폼에 이미지 올리기
• 외부 데이터 소스 API: DataOps 연동