참조 구현

"이런 응답을 주면 된다" 를 코드로 보여 주는 최소 서버입니다. 파이썬 표준 라이브러리만 쓰므로 python3 하나로 돕니다. 디렉터리 하나를 데이터 루트로 삼아, 그 아래 폴더 하나를 데이터셋 하나로 내보냅니다.

data/
└── line-inspect-w38/      # 데이터셋 id
    ├── 00010.png
    ├── 00011.png
    └── 00012.png

이 서버로 플랫폼에 소스를 등록하고 동기화를 두 번 돌려 확인했습니다. 첫 동기화는 신규 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

KINDS = {".png": "image", ".jpg": "image", ".jpeg": "image", ".bmp": "image",
         ".json": "annotation", ".xml": "annotation", ".ply": "pointcloud",
         ".csv": "timeseries", ".parquet": "timeseries"}


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 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 (self.root / parts[1]).is_dir():
            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(), "other"),
                        "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 은 매번 파일에서 계산합니다. 실제 서버라면 계산해 둔 값을 저장해 두세요.
• 오류는 상태 코드 + {"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 에 같은 API 3개가 열립니다. 이 장의 화면은 이 mock 으로 찍었습니다.

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