참조 구현

플랫폼이 기대하는 응답을 구현한 최소 서버. 파이썬 표준 라이브러리만 사용하므로 python3 만으로 실행 가능. 디렉터리 하나를 데이터 루트로 지정하고, 그 아래 폴더 하나를 데이터셋 하나로 제공.

이 데이터의 첫 동기화 결과는 신규 3. 파일 하나를 변경하고 하나를 삭제한 뒤 다시 동기화하면 받음 1 · 지움 1 · 유지 1.

코드#

"""DataOps 최소 서버: 디렉터리 하나를 데이터셋 목록·매니페스트·파일로 내보낸다.

    python3 dataops_server.py ./data --token <your-secret> --port 18080

./data/<데이터셋 id>/ 아래 파일이 그 데이터셋의 파일이다. 표준 라이브러리만 쓴다.
"""

import argparse
import hashlib
import json
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlsplit

# CSV · Parquet 는 시계열인지 표인지 알 때만 kind 를 "timeseries" 나 "tabular" 로 준다. 플랫폼이 그 값을 따른다
KINDS = {".png": "image", ".jpg": "image", ".jpeg": "image", ".bmp": "image",
         ".json": "annotation", ".xml": "annotation", ".ply": "pointcloud"}


def iso(ts: float) -> str:
    return datetime.fromtimestamp(ts, timezone.utc).isoformat().replace("+00:00", "Z")


def file_id(name: str) -> str:
    # 파일 이름에서 만든 불변 id. 이웃 파일이 늘어도 바뀌지 않는다
    return "f-" + hashlib.sha1(name.encode()).hexdigest()[:12]


def files_of(root: Path, dataset: str) -> list[Path]:
    return sorted(p for p in (root / dataset).iterdir() if p.is_file())


def is_dataset(root: Path, name: str) -> bool:
    # 데이터 루트 바로 아래 폴더만 데이터셋이다. "../" 같은 경로 탈출을 막는다
    d = (root / name).resolve()
    return d.parent == root.resolve() and d.is_dir()


def dataset_entry(root: Path, d: Path) -> dict:
    files = files_of(root, d.name)
    # 파일이 추가·변경·삭제되면 반드시 바뀌어야 한다. 디렉터리와 파일 중 가장 늦은 시각을 쓴다
    updated = max([d.stat().st_mtime] + [f.stat().st_mtime for f in files])
    return {"id": d.name, "name": d.name, "modality": "image",
            "file_count": len(files), "size_bytes": sum(f.stat().st_size for f in files),
            "updated_at": iso(updated)}


class Handler(BaseHTTPRequestHandler):
    root: Path
    token: str

    def send_json(self, status: int, body: dict) -> None:
        data = json.dumps(body, ensure_ascii=False).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self) -> None:
        if self.headers.get("Authorization") != f"Bearer {self.token}":
            return self.send_json(401, {"error": "invalid_token", "message": "토큰이 올바르지 않습니다."})
        url = urlsplit(self.path)
        parts = [unquote(p) for p in url.path.strip("/").split("/")]
        if parts[:1] != ["api"]:
            return self.send_json(404, {"error": "not_found", "message": url.path})
        parts = parts[1:]
        query = parse_qs(url.query)

        if parts == ["datasets"]:                                     # 1) 목록
            items = [dataset_entry(self.root, d) for d in sorted(self.root.iterdir()) if d.is_dir()]
            after = query.get("updated_after", [None])[0]
            if after:
                items = [i for i in items if i["updated_at"] > after]
            return self.send_json(200, {"items": items, "page": 1, "total": len(items)})

        if len(parts) >= 3 and parts[0] == "datasets" and is_dataset(self.root, parts[1]):
            dataset = parts[1]
            if parts[2:] == ["manifest"]:                             # 2) 매니페스트
                return self.send_json(200, {
                    "dataset_id": dataset,
                    "annotation_format": None,
                    "classes": [],
                    "files": [{
                        "file_id": file_id(f.name),
                        "filename": f.name,
                        "kind": KINDS.get(f.suffix.lower()),
                        "size_bytes": f.stat().st_size,
                        "sha256": hashlib.sha256(f.read_bytes()).hexdigest(),
                        "meta": {},
                    } for f in files_of(self.root, dataset)],
                })
            if len(parts) == 4 and parts[2] == "files":               # 3) 파일
                for f in files_of(self.root, dataset):
                    if file_id(f.name) == parts[3]:
                        data = f.read_bytes()
                        self.send_response(200)
                        self.send_header("Content-Type", "application/octet-stream")
                        self.send_header("Content-Length", str(len(data)))
                        self.end_headers()
                        self.wfile.write(data)
                        return
        return self.send_json(404, {"error": "not_found", "message": f"{url.path} 를 찾을 수 없습니다."})


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("root", type=Path)
    parser.add_argument("--token", required=True)
    parser.add_argument("--host", default="0.0.0.0")
    parser.add_argument("--port", type=int, default=18080)
    args = parser.parse_args()
    Handler.root, Handler.token = args.root, args.token
    print(f"http://{args.host}:{args.port}/api")
    ThreadingHTTPServer((args.host, args.port), Handler).serve_forever()


if __name__ == "__main__":
    main()

구현 규칙:

• file_id 는 파일 이름으로 생성한 불변 값. 파일 내용이 변경되어도 id 는 유지되므로, 플랫폼이 "변경" 으로 판정해 같은 행을 교체.
• updated_at 은 파일 변경 시 함께 갱신(디렉터리와 파일의 수정 시각 중 가장 늦은 것).
• sha256 은 매번 파일에서 계산. 실제 서버는 미리 계산해 저장한 값 사용 필요.
• kind 는 확실할 때만 지정. CSV · Parquet 에 timeseries 나 tabular 를 지정하면 플랫폼은 데이터셋 모달리티보다 그 값을 따름.
• 데이터 루트 바로 아래 폴더만 데이터셋으로 취급(../ 같은 경로는 404).
• 오류는 상태 코드 + {"error", "message"} JSON.

실행 예시#

python3 dataops_server.py ./data --token '<your-secret>' --port 18080

# 다른 셸에서
TOKEN='<your-secret>'
BASE=http://127.0.0.1:18080/api
curl -H "Authorization: Bearer $TOKEN" "$BASE/datasets"
curl -H "Authorization: Bearer $TOKEN" "$BASE/datasets/line-inspect-w38/manifest"
curl -H "Authorization: Bearer $TOKEN" -o out.png "$BASE/datasets/line-inspect-w38/files/<file_id>"
sha256sum out.png            # 매니페스트의 sha256 과 같아야 한다
curl "$BASE/datasets"         # 토큰이 없으면 401

{"items": [{"id": "line-inspect-w38", "name": "line-inspect-w38", "modality": "image", "file_count": 3, "size_bytes": 403, "updated_at": "2026-09-21T07:10:36.166989Z"}], "page": 1, "total": 1}
{"error": "invalid_token", "message": "토큰이 올바르지 않습니다."}

플랫폼이 이 서버에 접근하려면 플랫폼 서버에서 접근하는 주소를 기준 URL 로 등록 필요. 같은 PC 면 http://127.0.0.1:18080/api, 다른 호스트면 그 호스트 주소.

변경 감지 시험하기#

파일을 수동으로 변경하면 다음 동기화의 판정이 달라짐.

플랫폼 내장 mock 서버 (개발용)#

플랫폼 서버에 개발·시험용 DataOps mock(실제 서버를 모방한 가짜 서버) 포함. 서버 설정으로 활성화하면 {플랫폼 주소}/mock-dataops 에 실제 서버와 같은 3개의 API 제공.

curl -H "Authorization: Bearer mock-dataops-token" http://localhost:10000/mock-dataops/datasets

외부 소스의 기준 URL 로 http://localhost:10000/mock-dataops, 토큰으로 mock-dataops-token 을 등록하면 실제 DataOps 서버 없이 등록부터 동기화까지 시험 가능. mock 의 목록은 page · page_size(기본 20) 로 나누고 total 반환.