Sync from External Sources

import os

from honeyhive import HoneyHive
from honeyhive.models import (
    CreateDatapointRequest,
    UpdateDatapointRequest,
    UpdateDatasetRequest,
)

client = HoneyHive(api_key=os.environ["HH_API_KEY"])
DATASET_NAME = os.environ["HH_DATASET_NAME"]

datasets = client.datasets.list(name=DATASET_NAME).datasets
if not datasets:
    raise ValueError(f"Dataset '{DATASET_NAME}' not found")

dataset = datasets[0]
dataset_id = dataset.id


def sync_dataset(rows: list[dict], id_field: str = "id"):
    """Sync source rows into a HoneyHive dataset by external_id."""

    # 1. Load existing datapoints and index by external_id
    existing = client.datapoints.list(dataset_name=DATASET_NAME)
    current_dataset_datapoints = [dp.id for dp in existing.datapoints]
    id_map: dict[str, str] = {}  # external_id -> HoneyHive datapoint id
    for dp in existing.datapoints:
        ext_id = (dp.metadata or {}).get("external_id")
        if ext_id:
            id_map[str(ext_id)] = dp.id

    # 2. Create or update each source row
    source_ids = set()
    created_ids: list[str] = []
    for row in rows:
        ext_id = str(row[id_field])
        source_ids.add(ext_id)

        # --- Adapt this mapping to your schema ---
        datapoint = {
            "inputs": {"question": row["question"]},
            "ground_truth": {"answer": row["answer"]},
            "metadata": {"external_id": ext_id},
        }

        if ext_id in id_map:
            client.datapoints.update(
                id_map[ext_id], UpdateDatapointRequest(**datapoint)
            )
        else:
            response = client.datapoints.create(CreateDatapointRequest(**datapoint))
            new_datapoint_id = response.result["insertedId"]
            created_ids.append(new_datapoint_id)
            id_map[ext_id] = new_datapoint_id

    # 3. Delete datapoints no longer in source
    stale_ids = set()
    for ext_id, dp_id in id_map.items():
        if ext_id not in source_ids:
            client.datapoints.delete(dp_id)
            stale_ids.add(dp_id)

    # 4. Update dataset membership once per sync run
    if created_ids or stale_ids:
        next_datapoints = [
            dp_id
            for dp_id in [*current_dataset_datapoints, *created_ids]
            if dp_id not in stale_ids
        ]
        client.datasets.update(
            UpdateDatasetRequest(
                dataset_id=dataset_id,
                datapoints=next_datapoints,
            )
        )


# Load rows from your source (S3, database, internal tool, etc.)
rows = [
    {"id": "q1", "question": "What is Python?", "answer": "A programming language"},
    {"id": "q2", "question": "What is tracing?", "answer": "Recording app execution"},
]

sync_dataset(rows)

import hashlib, json

def content_hash(data: dict) -> str:
    """Deterministic hash for change detection."""
    return hashlib.sha256(
        json.dumps(data, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()

content = {
    "inputs": {"question": row["question"]},
    "ground_truth": {"answer": row["answer"]},
}
row_hash = content_hash(content)

datapoint = {
    **content,
    "metadata": {"external_id": ext_id, "content_hash": row_hash},
}

# Skip update if hash matches
existing_hash = (dp.metadata or {}).get("content_hash")
if ext_id in id_map and existing_hash == row_hash:
    continue  # unchanged, skip

{
  "dataset_name": "my-dataset",
  "content_hash": "3f2d1a9b7c42",
  "row_count": 250,
  "synced_at": "2026-02-13T10:22:00Z",
  "source": "data/test_cases.json"
}

import datetime, hashlib, json

def dataset_hash(rows: list[dict], id_field: str = "id") -> str:
    """Deterministic hash over all rows."""
    return hashlib.sha256(
        json.dumps(sorted(rows, key=lambda r: str(r[id_field])),
                   sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()[:12]

manifest = {
    "dataset_name": DATASET_NAME,
    "content_hash": dataset_hash(rows),
    "row_count": len(rows),
    "synced_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    "source": "data/test_cases.json",
}

with open("dataset-version.json", "w") as f:
    json.dump(manifest, f, indent=2)

# .github/workflows/sync-dataset.yml
name: Sync dataset
on:
  push:
    paths:
      - "data/test_cases.json"  # trigger when source file changes
  workflow_dispatch:             # manual trigger

jobs:
  sync:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install honeyhive
      - run: python sync_dataset.py
        env:
          HH_API_KEY: ${{ secrets.HH_API_KEY }}
          HH_DATASET_NAME: my-dataset