Reference implementation

This minimal server shows in code the responses the platform expects. 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.

The first sync of this data gives 3 new. After you replace one file and delete one, the next sync gives 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

# give CSV · Parquet a kind of "timeseries" or "tabular" only when you know which — the platform follows it
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:
    # 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 is_dataset(root: Path, name: str) -> bool:
    # only folders directly under the data root are datasets — blocks path escapes such as "../"
    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)
    # 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 is_dataset(self.root, parts[1]):
            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()),
                        "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.
• kind is given only when certain. If a CSV · Parquet file carries timeseries or tabular, the platform follows it over the dataset modality.
• Only folders directly under the data root are served as datasets (a path such as ../ gets 404).
• 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.

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 as a real server open at {platform address}/mock-dataops.

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

Register http://localhost:10000/mock-dataops as the base URL of an external source, with mock-dataops-token as the token, to test everything from registration to sync without a real DataOps server. The mock's list is split by page · page_size (default 20) and returns total.