#!/usr/bin/env python3 """ Atomic Dataset Persistence for HermesFace Save state to Hugging Face Dataset with atomic commit, checksum, and backup. Usage: python3 save_to_dataset_atomic.py [source_path2...] Env vars: HF_TOKEN HF access token (read from env by huggingface_hub) """ import hashlib import json import logging import os import shutil import sys import tempfile import time from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download from huggingface_hub.utils import RepositoryNotFoundError logging.basicConfig( level=logging.INFO, format='{"timestamp": "%(asctime)s", "level": "%(levelname)s", "module": "atomic-save", "message": "%(message)s"}', ) logger = logging.getLogger(__name__) class AtomicDatasetSaver: def __init__(self, repo_id: str, dataset_path: str = "state"): self.repo_id = repo_id self.dataset_path = Path(dataset_path) self.api = HfApi() self.max_retries = 3 self.base_delay = 1.0 self.max_backups = 3 logger.info(f"init repo_id={repo_id} dataset_path={dataset_path}") def calculate_checksum(self, file_path: Path) -> str: h = hashlib.sha256() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): h.update(chunk) return h.hexdigest() def create_backup(self, current_commit: Optional[str]) -> Optional[str]: if not current_commit: return None try: files = self.api.list_repo_files( repo_id=self.repo_id, repo_type="dataset", revision=current_commit ) state_files = [f for f in files if f.startswith(str(self.dataset_path))] if not state_files: return None timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = f"backups/state_{timestamp}" logger.info(f"creating_backup path={backup_path} files={len(state_files)}") with tempfile.TemporaryDirectory() as tmpdir: tmpdir_path = Path(tmpdir) ops = [] for file_path in state_files: local = hf_hub_download( repo_id=self.repo_id, repo_type="dataset", filename=file_path, revision=current_commit, ) if local: dst = tmpdir_path / Path(file_path).name shutil.copy2(local, dst) ops.append(CommitOperationAdd( path_in_repo=f"{backup_path}/{Path(file_path).name}", path_or_fileobj=str(dst), )) if ops: info = self.api.create_commit( repo_id=self.repo_id, repo_type="dataset", operations=ops, commit_message=f"Backup state before update - {timestamp}", parent_commit=current_commit, ) logger.info(f"backup_created commit={info.oid}") return info.oid except Exception as e: logger.error(f"backup_failed error={e}") return None def save_state_atomic( self, state_data: Dict[str, Any], source_paths: List[str] ) -> Dict[str, Any]: operation_id = f"save_{int(time.time())}" logger.info(f"starting_atomic_save op={operation_id} sources={source_paths}") try: try: repo_info = self.api.repo_info(repo_id=self.repo_id, repo_type="dataset") current_commit = repo_info.sha except RepositoryNotFoundError: current_commit = None backup_commit = self.create_backup(current_commit) with tempfile.TemporaryDirectory() as tmpdir: state_dir = Path(tmpdir) / self.dataset_path state_dir.mkdir(parents=True, exist_ok=True) metadata = { "timestamp": datetime.now().isoformat(), "operation_id": operation_id, "checksum": None, "backup_commit": backup_commit, "state_data": state_data, } metadata_path = state_dir / "metadata.json" with open(metadata_path, "w") as f: json.dump(metadata, f, indent=2) ops = [CommitOperationAdd( path_in_repo="state/metadata.json", path_or_fileobj=str(metadata_path), )] for source_path in source_paths: src = Path(source_path) if src.exists(): dst = state_dir / src.name shutil.copy2(src, dst) checksum = self.calculate_checksum(dst) ops.append(CommitOperationAdd( path_in_repo=f"state/{src.name}", path_or_fileobj=str(dst), )) logger.info(f"file_added source={source_path} sha256={checksum[:12]}") metadata["checksum"] = hashlib.sha256( json.dumps(state_data, sort_keys=True).encode() ).hexdigest() with open(metadata_path, "w") as f: json.dump(metadata, f, indent=2) info = self.api.create_commit( repo_id=self.repo_id, repo_type="dataset", operations=ops, commit_message=f"Atomic state update - {operation_id}", parent_commit=current_commit, ) result = { "success": True, "operation_id": operation_id, "commit_id": info.oid, "backup_commit": backup_commit, "timestamp": datetime.now().isoformat(), "files_count": len(source_paths), } logger.info(f"atomic_save_completed {result}") return result except Exception as e: logger.error(f"atomic_save_failed error={e}") raise def main() -> None: if len(sys.argv) < 3: print(json.dumps({ "error": "Usage: python save_to_dataset_atomic.py [source_path2...]", "status": "error", }, indent=2)) sys.exit(1) repo_id = sys.argv[1] source_paths = sys.argv[2:] for p in source_paths: if not os.path.exists(p): print(json.dumps({"error": f"Source path does not exist: {p}", "status": "error"}, indent=2)) sys.exit(1) state_data = { "environment": "production", "version": "1.0.0", "platform": "huggingface-spaces", "app": "hermesface", "timestamp": datetime.now().isoformat(), } try: saver = AtomicDatasetSaver(repo_id) result = saver.save_state_atomic(state_data, source_paths) print(json.dumps(result, indent=2)) except Exception as e: print(json.dumps({"error": str(e), "status": "error"}, indent=2)) sys.exit(1) if __name__ == "__main__": main()