REST API 로 자동화하기

화면의 모든 기능은 /api/v1/… REST API 로도 사용 가능. 화면도 같은 API 사용. 엔드포인트별 요청·응답은 레퍼런스 참고. 서버의 OpenAPI 문서는 /openapi.json, 대화형 문서는 /docs 에서도 제공.

예시의 주소(http://localhost:10000), 테넌트(DEMO), 계정([email protected])은 자기 환경 값으로 변경.

인증: 쿠키 또는 JWT#

로그인 방식은 두 가지, 계정·비밀번호는 동일. 둘 다 폼 필드 username(메일)과 password 사용.

• 로그인 정보가 올바르지 않으면 400 {"detail": "LOGIN_BAD_CREDENTIALS"}. 비활성 계정도 같은 응답.
• 세션 수명은 서버 설정 GEO_MLOPS_AUTH_TOKEN_LIFETIME(초)이 결정. 기본값 0 은 만료 없음. 운영 환경에서는 적절한 수명 설정, 자동화에는 전용 계정 사용 필요.
• 로그아웃은 POST /auth/cookie/logout 또는 POST /auth/jwt/logout. 서버는 세션을 저장하지 않으므로 로그아웃해도 이미 발급된 JWT(쿠키 내부의 값 포함)는 수명이 끝날 때까지 유효. 유출된 JWT 는 계정을 비활성화하거나 서명 키(GEO_MLOPS_AUTH_SECRET)를 변경해야 차단. 서명 키를 변경하면 모든 사용자 로그아웃.
• 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(role DEVELOPER lacks permission …), 속하지 않았거나 정지된 테넌트를 X-Tenant 로 지정하면 403(no membership for the requested tenant). 다른 테넌트의 자원 id 로 요청하면 404. /users/me, /api/v1/stream/notifications 처럼 사용자 단위인 API 는 테넌트 불필요.

오류 형식#

/api/v1/… 의 실패 응답은 모두 같은 형식.

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

/api/v1 외의 경로는 형식이 다름. /auth/… · /users/… 는 {"detail": …}, /mlflow 는 MLflow 의 오류 형식(인증 실패는 {"detail": "Not authenticated"}), /v2 는 OCI 형식 {"errors": [{"code", "message"}]}(인증 · 권한 실패는 {"detail": …}).

페이지네이션 두 가지#

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

page(1부터)와 page_size 를 지정하면 {items, total, page, page_size} 반환. page_size 를 생략하면 서버 기본값(20, GEO_MLOPS_LIST_PAGE_SIZE) 사용. page_size 는 200 까지, 초과하면 422. 대부분의 목록은 검색·정렬 파라미터도 같은 이름 사용.

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"}

청크 업로드#

대용량 파일은 한 요청으로 전송하지 않음. 앞단 프록시의 본문 크기·응답 시간 제한을 초과하기 때문. 작은 데이터셋 파일은 POST /api/v1/datasets/{id}/files(multipart 필드 file) 한 번으로 업로드 가능. 화면은 32 MiB 를 초과하는 파일과 모든 .zip 을 청크로 전송. 플랫폼은 세션 열기 → 청크 전송 → 마감이라는 같은 절차를 두 곳에서 사용. 마감 요청은 202 로 즉시 반환. 조립·검증은 서버가 이어서 수행하므로 상태를 다시 조회해 완료 확인.

청크는 세션 응답의 chunk_size 단위로 분할. 두 경우 모두 앞단 프록시의 본문 한도보다 작아야 함.

"""파일 하나를 청크로 나눠 데이터셋에 올린다 (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 연동