Reference implementation

This is a minimal server that shows "respond like this" in code. It uses only the Python standard library, so python3 alone runs it. It takes one directory as the data root and exposes each folder under it as one dataset.

data/
└── line-inspect-w38/      # dataset id
    ├── 00010.png
    ├── 00011.png
    └── 00012.png

We registered this server as a source on the platform and ran sync twice, confirming the result 3 new → (after replacing one file and deleting one) 1 fetched · 1 deleted · 1 kept.

Code#

"""Minimal DataOps server — exposes one directory as a dataset list, manifests and files.

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

The files under ./data/<dataset id>/ are that dataset's files. Uses only the standard library.
"""

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:
    # immutable id made from the file name — does not change when neighboring files are added
    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)
    # must change whenever a file is added, changed or deleted — the latest time among the directory and its files
    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": "The token is invalid."})
        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) list
            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) manifest
                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) file
                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} was not found."})


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()

What it gets right:

• file_id is an immutable value made from the file name. The id stays the same when the file contents change, so the platform sees it as "changed" and replaces the same row.
• updated_at changes whenever a file changes (the latest modification time among the directory and its files).
• sha256 is computed from the file every time. A real server should store precomputed values.
• Errors are a status code + {"error", "message"} JSON.

Try it#

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

# in another shell
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            # must equal the sha256 in the manifest
curl "$BASE/datasets"         # 401 without a token

{"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": "The token is invalid."}

For the platform to reach this server, register the address as seen from the platform server as the base URL. On the same PC that is http://127.0.0.1:18080/api; on another host, use that host's address.

Test change detection#

If you change files by hand, the verdict of the next sync changes right away.

The platform's built-in mock server (for development)#

The platform server includes a DataOps mock (a fake server that imitates a real one) for development and testing. When turned on in the server settings, the same three APIs open at {platform address}/mock-dataops. The screenshots in this chapter were taken with this mock.

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