import atexit import hashlib import json as json_mod import os import shutil import sqlite3 import time import uuid from collections.abc import Iterator from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from threading import Lock, local from typing import Any try: import fcntl except ImportError: fcntl = None try: import msvcrt as _msvcrt except ImportError: _msvcrt = None import huggingface_hub as hf import orjson from trackio import cas from trackio.commit_scheduler import CommitScheduler from trackio.dummy_commit_scheduler import DummyCommitScheduler from trackio.typehints import Manifest, Sha256Digest from trackio.utils import ( TRACKIO_DIR, _emit_nonfatal_warning, canonical_project_name, deserialize_values, get_color_palette, on_spaces, project_media_dir, serialize_values, ) DB_EXT = ".db" _JOURNAL_MODE_WHITELIST = frozenset( {"wal", "delete", "truncate", "persist", "memory", "off"} ) _SYNCHRONOUS_WHITELIST = frozenset({"off", "normal", "full", "extra"}) _LOCKING_MODE_WHITELIST = frozenset({"normal", "exclusive"}) _TEMP_STORE_WHITELIST = frozenset({"default", "file", "memory"}) _READ_ONLY_QUERY_PREFIXES = ("select", "with", "pragma") _QUERY_MAX_ROWS = 10_000 _READ_ONLY_PRAGMAS = frozenset( {"table_info", "table_xinfo", "index_list", "index_info", "index_xinfo"} ) def _env_pragma_choice(name: str, whitelist: frozenset[str]) -> str | None: value = os.environ.get(name, "").strip().lower() if value in whitelist: return value.upper() return None def _env_pragma_int(name: str) -> int | None: value = os.environ.get(name, "").strip() if not value: return None try: return int(value) except ValueError: return None def _configure_sqlite_pragmas(conn: sqlite3.Connection) -> None: override = os.environ.get("TRACKIO_SQLITE_JOURNAL_MODE", "").strip().lower() if override in _JOURNAL_MODE_WHITELIST: journal = override.upper() elif on_spaces(): journal = "DELETE" else: journal = "WAL" conn.execute(f"PRAGMA journal_mode = {journal}") synchronous = ( _env_pragma_choice("TRACKIO_SQLITE_SYNCHRONOUS", _SYNCHRONOUS_WHITELIST) or "NORMAL" ) conn.execute(f"PRAGMA synchronous = {synchronous}") temp_store = ( _env_pragma_choice("TRACKIO_SQLITE_TEMP_STORE", _TEMP_STORE_WHITELIST) or "MEMORY" ) conn.execute(f"PRAGMA temp_store = {temp_store}") conn.execute("PRAGMA cache_size = -20000") mmap_size = _env_pragma_int("TRACKIO_SQLITE_MMAP_SIZE") conn.execute(f"PRAGMA mmap_size = {0 if mmap_size is None else mmap_size}") locking_mode = _env_pragma_choice( "TRACKIO_SQLITE_LOCKING_MODE", _LOCKING_MODE_WHITELIST ) if locking_mode is not None: conn.execute(f"PRAGMA locking_mode = {locking_mode}") elif on_spaces(): conn.execute("PRAGMA locking_mode = EXCLUSIVE") _persistent_connections: dict[str, sqlite3.Connection] = {} _persistent_lock = Lock() _db_access_locks: dict[str, Lock] = {} def _get_db_access_lock(db_path: Path) -> Lock: key = str(db_path) with _persistent_lock: if key not in _db_access_locks: _db_access_locks[key] = Lock() return _db_access_locks[key] def _get_or_create_persistent_conn( db_path: Path, timeout: float = 30.0 ) -> sqlite3.Connection: key = str(db_path) with _persistent_lock: conn = _persistent_connections.get(key) if conn is not None: try: conn.execute("SELECT 1") return conn except sqlite3.Error: try: conn.close() except sqlite3.Error: pass _persistent_connections.pop(key, None) conn = sqlite3.connect(str(db_path), timeout=timeout, check_same_thread=False) _configure_sqlite_pragmas(conn) conn.execute("SELECT 1") _persistent_connections[key] = conn return conn def _close_all_persistent_connections() -> None: with _persistent_lock: for conn in _persistent_connections.values(): try: conn.close() except sqlite3.Error: pass _persistent_connections.clear() atexit.register(_close_all_persistent_connections) class ProcessLock: """Lock used to coordinate database access. Normally uses file-based locking for cross-process coordination. When running on a bucket-mounted filesystem where file locks are unreliable, falls back to an in-memory threading Lock (single-process only).""" _thread_locks: dict[str, Lock] = {} _meta_lock = Lock() def __init__(self, lockfile_path: Path): self.lockfile_path = lockfile_path self.lockfile = None self._use_thread_lock = on_spaces() if self._use_thread_lock: key = str(lockfile_path) with ProcessLock._meta_lock: if key not in ProcessLock._thread_locks: ProcessLock._thread_locks[key] = Lock() self._thread_lock = ProcessLock._thread_locks[key] def __enter__(self): if self._use_thread_lock: self._thread_lock.acquire() return self if fcntl is None and _msvcrt is None: return self self.lockfile_path.parent.mkdir(parents=True, exist_ok=True) self.lockfile = open(self.lockfile_path, "w") max_retries = 100 for attempt in range(max_retries): try: if fcntl is not None: fcntl.flock(self.lockfile.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) else: _msvcrt.locking(self.lockfile.fileno(), _msvcrt.LK_NBLCK, 1) return self except (IOError, OSError): if attempt < max_retries - 1: time.sleep(0.1) else: raise IOError("Could not acquire database lock after 10 seconds") def __exit__(self, exc_type, exc_val, exc_tb): if self._use_thread_lock: self._thread_lock.release() return if self.lockfile: try: if fcntl is not None: fcntl.flock(self.lockfile.fileno(), fcntl.LOCK_UN) elif _msvcrt is not None: _msvcrt.locking(self.lockfile.fileno(), _msvcrt.LK_UNLCK, 1) except (IOError, OSError): pass self.lockfile.close() _LOGS_READ_CACHE: dict[tuple[Any, ...], tuple[int, list[dict[str, Any]]]] = {} _LOGS_READ_CACHE_LOCK = Lock() _LOGS_READ_CACHE_MAX_KEYS = 512 _LOGS_READ_CACHE_MAX_ROWS_PER_ENTRY = 4000 def _spaces_logs_read_cache_enabled() -> bool: if not on_spaces(): return False v = os.environ.get("TRACKIO_DISABLE_LOGS_CACHE", "").strip().lower() return v not in ("1", "true", "yes") def _sqlite_db_invalidation_mtime_ns(db_path: Path) -> int | None: try: m = db_path.stat().st_mtime_ns except OSError: return None wal_path = db_path.with_name(db_path.name + "-wal") if wal_path.is_file(): try: m = max(m, wal_path.stat().st_mtime_ns) except OSError: pass return m def _logs_read_cache_key( project: str, run: str | None, run_id: str | None, max_points: int | None, ) -> tuple[Any, ...]: return ( project, run or "", run_id or "", max_points if max_points is not None else -1, ) def _logs_read_cache_get( db_path: Path, key: tuple[Any, ...] ) -> list[dict[str, Any]] | None: if not _spaces_logs_read_cache_enabled(): return None mtime_ns = _sqlite_db_invalidation_mtime_ns(db_path) if mtime_ns is None: return None with _LOGS_READ_CACHE_LOCK: item = _LOGS_READ_CACHE.get(key) if item is None: return None cached_mtime, logs = item if cached_mtime != mtime_ns: del _LOGS_READ_CACHE[key] return None return [{**d} for d in logs] def _logs_read_cache_put( db_path: Path, key: tuple[Any, ...], logs: list[dict[str, Any]] ) -> None: if not _spaces_logs_read_cache_enabled(): return if len(logs) > _LOGS_READ_CACHE_MAX_ROWS_PER_ENTRY: return mtime_ns = _sqlite_db_invalidation_mtime_ns(db_path) if mtime_ns is None: return snapshot = [{**d} for d in logs] with _LOGS_READ_CACHE_LOCK: while len(_LOGS_READ_CACHE) >= _LOGS_READ_CACHE_MAX_KEYS: _LOGS_READ_CACHE.pop(next(iter(_LOGS_READ_CACHE))) _LOGS_READ_CACHE[key] = (mtime_ns, snapshot) _SYSTEM_LOGS_READ_CACHE: dict[tuple[Any, ...], tuple[int, list[dict[str, Any]]]] = {} def _system_logs_read_cache_key( project: str, run: str | None, run_id: str | None, max_points: int | None = None, ) -> tuple[Any, ...]: return ( "system_logs", project, run or "", run_id or "", max_points if max_points is not None else -1, ) def _system_logs_read_cache_get( db_path: Path, key: tuple[Any, ...] ) -> list[dict[str, Any]] | None: if not _spaces_logs_read_cache_enabled(): return None mtime_ns = _sqlite_db_invalidation_mtime_ns(db_path) if mtime_ns is None: return None with _LOGS_READ_CACHE_LOCK: item = _SYSTEM_LOGS_READ_CACHE.get(key) if item is None: return None cached_mtime, logs = item if cached_mtime != mtime_ns: del _SYSTEM_LOGS_READ_CACHE[key] return None return [{**d} for d in logs] def _system_logs_read_cache_put( db_path: Path, key: tuple[Any, ...], logs: list[dict[str, Any]] ) -> None: if not _spaces_logs_read_cache_enabled(): return if len(logs) > _LOGS_READ_CACHE_MAX_ROWS_PER_ENTRY: return mtime_ns = _sqlite_db_invalidation_mtime_ns(db_path) if mtime_ns is None: return snapshot = [{**d} for d in logs] with _LOGS_READ_CACHE_LOCK: while len(_SYSTEM_LOGS_READ_CACHE) >= _LOGS_READ_CACHE_MAX_KEYS: _SYSTEM_LOGS_READ_CACHE.pop(next(iter(_SYSTEM_LOGS_READ_CACHE))) _SYSTEM_LOGS_READ_CACHE[key] = (mtime_ns, snapshot) class SQLiteStorage: _dataset_import_attempted = False _dataset_import_pending = False _dataset_remote_synced = False _dataset_loading = local() _current_scheduler: CommitScheduler | DummyCommitScheduler | None = None _scheduler_lock = Lock() _ARTIFACT_PARQUET_TABLES: dict[str, list[str]] = { "artifacts": ["id", "name", "type", "description", "created_at"], "artifact_versions": [ "id", "artifact_id", "version", "manifest_digest", "manifest", "metadata", "size_bytes", "producer_run_id", "producer_run_name", "created_at", ], "artifact_aliases": ["artifact_id", "alias", "artifact_version_id"], "run_artifact_links": [ "id", "run_id", "run_name", "artifact_version_id", "direction", "created_at", ], } @staticmethod @contextmanager def _get_connection( db_path: Path, *, timeout: float = 30.0, configure_pragmas: bool = True, row_factory=sqlite3.Row, ) -> Iterator[sqlite3.Connection]: if on_spaces(): # On Spaces, all callers share a single persistent connection # that is pragma-configured at creation time. The `configure_pragmas` # flag is intentionally ignored here — the pragmas (journal mode, # synchronous, locking mode) don't affect query semantics. access_lock = _get_db_access_lock(db_path) access_lock.acquire() try: conn = _get_or_create_persistent_conn(db_path, timeout=timeout) conn.row_factory = row_factory with conn: yield conn finally: access_lock.release() else: conn = sqlite3.connect(str(db_path), timeout=timeout) try: if configure_pragmas: _configure_sqlite_pragmas(conn) if row_factory is not None: conn.row_factory = row_factory with conn: yield conn finally: conn.close() @staticmethod def _get_process_lock(project: str) -> ProcessLock: lockfile_path = TRACKIO_DIR / f"{canonical_project_name(project)}.lock" return ProcessLock(lockfile_path) @staticmethod def get_project_db_filename(project: str) -> str: """Get the database filename for a specific project.""" return f"{canonical_project_name(project)}{DB_EXT}" @staticmethod def get_project_db_path(project: str) -> Path: """Get the database path for a specific project.""" filename = SQLiteStorage.get_project_db_filename(project) return TRACKIO_DIR / filename @staticmethod def validate_project_name(project: str) -> None: """Reject project names whose canonical on-disk identity would collide with the parquet sidecar files trackio writes for another project's metrics/artifact tables (e.g. a project ``model_artifacts`` shares the file ``model_artifacts.parquet`` with project ``model``'s artifacts sidecar).""" reserved = tuple( f"_{table}" for table in ( "system", "configs", "traces", *SQLiteStorage._ARTIFACT_PARQUET_TABLES, ) ) canonical = canonical_project_name(project) for suffix in reserved: if len(canonical) > len(suffix) and canonical.endswith(suffix): sibling = canonical[: -len(suffix)] raise ValueError( f"Project name {project!r} is not allowed: its on-disk name " f"{canonical!r} ends with the reserved suffix {suffix!r}, which " f"collides with the parquet sidecar files trackio writes for a " f"project named {sibling!r}. Choose a different project name." ) @staticmethod def init_db(project: str) -> Path: """ Initialize the SQLite database with required tables. Returns the database path. """ SQLiteStorage._ensure_hub_loaded() db_path = SQLiteStorage.get_project_db_path(project) db_path.parent.mkdir(parents=True, exist_ok=True) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path, row_factory=None) as conn: cursor = conn.cursor() cursor.execute( """ CREATE TABLE IF NOT EXISTS metrics ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, timestamp TEXT NOT NULL, run_name TEXT NOT NULL, step INTEGER NOT NULL, metrics TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS configs ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, run_name TEXT NOT NULL, config TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(run_id) ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS system_metrics ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, timestamp TEXT NOT NULL, run_name TEXT NOT NULL, metrics TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS traces ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL, timestamp TEXT NOT NULL, run_name TEXT NOT NULL, step INTEGER NOT NULL, key TEXT NOT NULL, trace_index INTEGER, messages TEXT NOT NULL, metadata TEXT NOT NULL, search_text TEXT NOT NULL, log_id TEXT, space_id TEXT ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS project_metadata ( key TEXT PRIMARY KEY, value TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS pending_uploads ( id INTEGER PRIMARY KEY AUTOINCREMENT, space_id TEXT NOT NULL, run_id TEXT, run_name TEXT, step INTEGER, file_path TEXT NOT NULL, relative_path TEXT, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, timestamp TEXT NOT NULL, run_name TEXT NOT NULL, title TEXT NOT NULL, text TEXT, level TEXT NOT NULL DEFAULT 'warn', step INTEGER, alert_id TEXT ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS artifacts ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, type TEXT NOT NULL, description TEXT, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS artifact_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, artifact_id INTEGER NOT NULL REFERENCES artifacts(id), version INTEGER NOT NULL, manifest_digest TEXT NOT NULL, manifest TEXT NOT NULL, metadata TEXT, size_bytes INTEGER NOT NULL, producer_run_id TEXT, producer_run_name TEXT, created_at TEXT NOT NULL, UNIQUE(artifact_id, version), UNIQUE(artifact_id, manifest_digest) ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS artifact_aliases ( artifact_id INTEGER NOT NULL REFERENCES artifacts(id), alias TEXT NOT NULL, artifact_version_id INTEGER NOT NULL REFERENCES artifact_versions(id), PRIMARY KEY (artifact_id, alias) ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS run_artifact_links ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT, run_name TEXT, artifact_version_id INTEGER NOT NULL REFERENCES artifact_versions(id), direction TEXT NOT NULL CHECK(direction IN ('input', 'output')), created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_run_artifact_links_run ON run_artifact_links(run_id, run_name) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_run_artifact_links_version ON run_artifact_links(artifact_version_id) """ ) cursor.execute( """ CREATE UNIQUE INDEX IF NOT EXISTS idx_run_artifact_links_unique ON run_artifact_links(run_id, artifact_version_id, direction) """ ) for col in ("kind TEXT NOT NULL DEFAULT 'media'", "digest TEXT"): try: cursor.execute(f"ALTER TABLE pending_uploads ADD COLUMN {col}") except sqlite3.OperationalError: pass cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_pending_uploads_kind ON pending_uploads(kind) """ ) metrics_cols = SQLiteStorage._table_columns(conn, "metrics") metrics_run_key = "run_id" if "run_id" in metrics_cols else "run_name" cursor.execute( f""" CREATE INDEX IF NOT EXISTS idx_metrics_run_step ON metrics({metrics_run_key}, step) """ ) cursor.execute( f""" CREATE INDEX IF NOT EXISTS idx_metrics_run_timestamp ON metrics({metrics_run_key}, timestamp) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_configs_run_name ON configs(run_name) """ ) system_cols = SQLiteStorage._table_columns(conn, "system_metrics") system_run_key = "run_id" if "run_id" in system_cols else "run_name" cursor.execute( f""" CREATE INDEX IF NOT EXISTS idx_system_metrics_run_timestamp ON system_metrics({system_run_key}, timestamp) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_traces_run_step ON traces(run_id, step) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_traces_run_timestamp ON traces(run_id, timestamp) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_traces_search ON traces(search_text) """ ) alerts_cols = SQLiteStorage._table_columns(conn, "alerts") alerts_run_key = "run_id" if "run_id" in alerts_cols else "run_name" cursor.execute( f""" CREATE INDEX IF NOT EXISTS idx_alerts_run ON alerts({alerts_run_key}) """ ) cursor.execute( """ CREATE INDEX IF NOT EXISTS idx_alerts_timestamp ON alerts(timestamp) """ ) cursor.execute( """ CREATE UNIQUE INDEX IF NOT EXISTS idx_alerts_alert_id ON alerts(alert_id) WHERE alert_id IS NOT NULL """ ) for table in ("metrics", "system_metrics"): for col in ("log_id TEXT", "space_id TEXT"): try: cursor.execute(f"ALTER TABLE {table} ADD COLUMN {col}") except sqlite3.OperationalError: pass cursor.execute( f"""CREATE UNIQUE INDEX IF NOT EXISTS idx_{table}_log_id ON {table}(log_id) WHERE log_id IS NOT NULL""" ) cursor.execute( f"""CREATE INDEX IF NOT EXISTS idx_{table}_pending ON {table}(space_id) WHERE space_id IS NOT NULL""" ) conn.commit() return db_path @staticmethod def _require_pyarrow(): try: import pyarrow as pa import pyarrow.parquet as pq except ImportError as e: raise ImportError( "Parquet import/export requires `trackio[spaces]`." ) from e return pa, pq @staticmethod def _table_columns(conn: sqlite3.Connection, table: str) -> set[str]: cursor = conn.cursor() try: cursor.execute(f"PRAGMA table_info({table})") except sqlite3.OperationalError: return set() return {row[1] for row in cursor.fetchall()} @staticmethod def _supports_run_ids(conn: sqlite3.Connection, table: str = "metrics") -> bool: return "run_id" in SQLiteStorage._table_columns(conn, table) @staticmethod def _resolve_run_identity( conn: sqlite3.Connection, run_name: str | None = None, run_id: str | None = None, *, table: str = "metrics", ) -> tuple[str, str] | None: supports_run_ids = SQLiteStorage._supports_run_ids(conn, table) if supports_run_ids: if run_id is not None: return ("run_id", run_id) if run_name is None: return None source_table = ( table if "timestamp" in SQLiteStorage._table_columns(conn, table) else "metrics" ) cursor = conn.cursor() cursor.execute( f""" SELECT run_id FROM {source_table} WHERE run_name = ? GROUP BY run_id ORDER BY MIN(timestamp) DESC LIMIT 1 """, (run_name,), ) row = cursor.fetchone() if row is None: return None return ("run_id", row[0]) resolved = run_name if run_name is not None else run_id if resolved is None: return None return ("run_name", resolved) @staticmethod def get_run_records(project: str) -> list[dict[str, str | None]]: SQLiteStorage._ensure_hub_loaded() db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() if SQLiteStorage._supports_run_ids(conn): cursor.execute( """ SELECT run_id, run_name, MIN(timestamp) as created_at FROM metrics GROUP BY run_id, run_name ORDER BY created_at ASC """ ) return [ { "id": row["run_id"], "name": row["run_name"], "created_at": row["created_at"], } for row in cursor.fetchall() ] cursor.execute( """ SELECT run_name, MIN(timestamp) as created_at FROM metrics GROUP BY run_name ORDER BY created_at ASC """ ) return [ { "id": row["run_name"], "name": row["run_name"], "created_at": row["created_at"], } for row in cursor.fetchall() ] except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return [] raise @staticmethod def get_latest_run_record_by_name( project: str, run_name: str ) -> dict[str, str | None] | None: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() if SQLiteStorage._supports_run_ids(conn): cursor.execute( """ SELECT run_id, run_name, MIN(timestamp) as created_at FROM metrics WHERE run_name = ? GROUP BY run_id, run_name ORDER BY created_at DESC LIMIT 1 """, (run_name,), ) row = cursor.fetchone() if row is None: return None return { "id": row["run_id"], "name": row["run_name"], "created_at": row["created_at"], } cursor.execute( """ SELECT run_name, MIN(timestamp) as created_at FROM metrics WHERE run_name = ? GROUP BY run_name ORDER BY created_at DESC LIMIT 1 """, (run_name,), ) row = cursor.fetchone() if row is None: return None return { "id": row["run_name"], "name": row["run_name"], "created_at": row["created_at"], } @staticmethod def _normalize_trace_rows_for_parquet( rows: list[dict[str, object]], ) -> list[dict[str, object]]: normalized: list[dict[str, object]] = [] for row in rows: new_row = dict(row) for col in ("messages", "metadata"): value = new_row.get(col) if value is None: continue if isinstance(value, memoryview): value = value.tobytes() if isinstance(value, (bytes, bytearray)): new_row[col] = bytes(value).decode("utf-8") elif not isinstance(value, str): new_row[col] = orjson.dumps(value).decode("utf-8") normalized.append(new_row) return normalized @staticmethod def _read_table_rows(db_path: Path, table: str) -> list[dict[str, object]]: try: with SQLiteStorage._get_connection( db_path, timeout=5.0, configure_pragmas=False ) as conn: cursor = conn.cursor() cursor.execute(f"SELECT * FROM {table}") return [dict(row) for row in cursor.fetchall()] except Exception: return [] @staticmethod def _decode_json_blob(value: object) -> dict[str, object]: if value is None: return {} if isinstance(value, memoryview): value = value.tobytes() return deserialize_values(orjson.loads(value)) @staticmethod def _flatten_json_rows( rows: list[dict[str, object]], json_col: str ) -> list[dict[str, object]]: flattened_rows = [] for row in rows: flat_row = {key: value for key, value in row.items() if key != json_col} expanded = SQLiteStorage._decode_json_blob(row.get(json_col)) for key, value in expanded.items(): if key not in flat_row: flat_row[key] = value flattened_rows.append(flat_row) return flattened_rows @staticmethod def _write_parquet_rows(parquet_path: Path, rows: list[dict[str, object]]) -> None: if not rows: return pa, pq = SQLiteStorage._require_pyarrow() column_names: list[str] = [] seen_columns: set[str] = set() for row in rows: for key in row: if key not in seen_columns: column_names.append(key) seen_columns.add(key) normalized_rows = [{key: row.get(key) for key in column_names} for row in rows] table = pa.Table.from_pylist(normalized_rows) write_kwargs = { "write_page_index": True, "use_content_defined_chunking": True, } try: pq.write_table(table, parquet_path, **write_kwargs) except TypeError: pq.write_table(table, parquet_path) @staticmethod def _read_parquet_rows(parquet_path: Path) -> list[dict[str, object]]: _, pq = SQLiteStorage._require_pyarrow() return pq.read_table(parquet_path).to_pylist() @staticmethod def _normalize_json_column_value(value: object) -> object: if value is None: return orjson.dumps({}) if isinstance(value, memoryview): return value.tobytes() if isinstance(value, (bytes, bytearray, str)): return value return orjson.dumps(serialize_values(value)) @staticmethod def _rows_to_sql_table_rows( rows: list[dict[str, object]], *, json_col: str, structural_cols: list[str], ) -> list[dict[str, object]]: sql_rows = [] for row in rows: sql_row = {col: row.get(col) for col in structural_cols} if json_col in row: sql_row[json_col] = SQLiteStorage._normalize_json_column_value( row.get(json_col) ) else: payload = { key: value for key, value in row.items() if key not in structural_cols and key != json_col } sql_row[json_col] = orjson.dumps(serialize_values(payload)) sql_rows.append(sql_row) return sql_rows @staticmethod def _replace_table_rows( db_path: Path, table: str, rows: list[dict[str, object]], columns: list[str], ) -> None: with SQLiteStorage._get_connection( db_path, configure_pragmas=False, row_factory=None ) as conn: cursor = conn.cursor() cursor.execute(f"DELETE FROM {table}") if rows: placeholders = ", ".join(["?"] * len(columns)) cursor.executemany( f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({placeholders})", [[row.get(column) for column in columns] for row in rows], ) conn.commit() @staticmethod def _flatten_and_write_parquet( db_path: Path, table: str, json_col: str, parquet_path: Path ) -> None: db_mtime_ns = _sqlite_db_invalidation_mtime_ns(db_path) if ( parquet_path.exists() and db_mtime_ns is not None and db_mtime_ns < parquet_path.stat().st_mtime_ns ): return rows = SQLiteStorage._read_table_rows(db_path, table) if not rows: return flat_rows = SQLiteStorage._flatten_json_rows(rows, json_col) SQLiteStorage._write_parquet_rows(parquet_path, flat_rows) @staticmethod def export_to_parquet(): """ Exports all projects' DB files as Parquet under the same path but with extension ".parquet". Also exports system_metrics to separate parquet files with "_system.parquet" suffix. Also exports configs to separate parquet files with "_configs.parquet" suffix. """ if not SQLiteStorage._dataset_import_attempted: return if not TRACKIO_DIR.exists(): return all_paths = os.listdir(TRACKIO_DIR) db_names = [f for f in all_paths if f.endswith(DB_EXT)] for db_name in db_names: db_path = TRACKIO_DIR / db_name SQLiteStorage._flatten_and_write_parquet( db_path, "metrics", "metrics", db_path.with_suffix(".parquet") ) SQLiteStorage._flatten_and_write_parquet( db_path, "system_metrics", "metrics", TRACKIO_DIR / (db_path.stem + "_system.parquet"), ) SQLiteStorage._flatten_and_write_parquet( db_path, "configs", "config", TRACKIO_DIR / (db_path.stem + "_configs.parquet"), ) trace_rows = SQLiteStorage._read_table_rows(db_path, "traces") if trace_rows: SQLiteStorage._write_parquet_rows( TRACKIO_DIR / (db_path.stem + "_traces.parquet"), SQLiteStorage._normalize_trace_rows_for_parquet(trace_rows), ) db_mtime_ns = _sqlite_db_invalidation_mtime_ns(db_path) for table in SQLiteStorage._ARTIFACT_PARQUET_TABLES: parquet_path = TRACKIO_DIR / f"{db_path.stem}_{table}.parquet" if ( parquet_path.exists() and db_mtime_ns is not None and db_mtime_ns < parquet_path.stat().st_mtime_ns ): continue artifact_rows = SQLiteStorage._read_table_rows(db_path, table) if artifact_rows: SQLiteStorage._write_parquet_rows(parquet_path, artifact_rows) @staticmethod def export_for_static_space( project: str, output_dir: Path, db_path_override: Path | None = None ) -> None: """ Exports a single project's data as Parquet + JSON files for static Space deployment. Args: project: The project name. output_dir: Directory to write the exported files to. db_path_override: If provided, read from this DB file instead of the default local project path. Useful when exporting from a downloaded remote database. """ db_path = db_path_override or SQLiteStorage.get_project_db_path(project) if not db_path.exists(): raise FileNotFoundError(f"No database found for project '{project}'") output_dir.mkdir(parents=True, exist_ok=True) aux_dir = output_dir / "aux" aux_dir.mkdir(parents=True, exist_ok=True) metrics_rows = SQLiteStorage._read_table_rows(db_path, "metrics") if metrics_rows: flat = SQLiteStorage._flatten_json_rows(metrics_rows, "metrics") SQLiteStorage._write_parquet_rows(output_dir / "metrics.parquet", flat) sys_rows = SQLiteStorage._read_table_rows(db_path, "system_metrics") if sys_rows: flat = SQLiteStorage._flatten_json_rows(sys_rows, "metrics") SQLiteStorage._write_parquet_rows(aux_dir / "system_metrics.parquet", flat) trace_rows = SQLiteStorage._read_table_rows(db_path, "traces") if trace_rows: SQLiteStorage._write_parquet_rows( aux_dir / "traces.parquet", SQLiteStorage._normalize_trace_rows_for_parquet(trace_rows), ) configs_rows = SQLiteStorage._read_table_rows(db_path, "configs") if configs_rows: flat = SQLiteStorage._flatten_json_rows(configs_rows, "config") SQLiteStorage._write_parquet_rows(aux_dir / "configs.parquet", flat) try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() if SQLiteStorage._supports_run_ids(conn): cursor.execute( """SELECT run_id, run_name, MIN(timestamp) as created_at, MAX(step) as last_step, COUNT(*) as log_count FROM metrics GROUP BY run_id, run_name ORDER BY created_at ASC""" ) rows = cursor.fetchall() runs_meta = [ { "id": row["run_id"], "name": row["run_name"], "created_at": row["created_at"], "last_step": row["last_step"], "log_count": row["log_count"], } for row in rows ] else: cursor.execute( """SELECT run_name, MIN(timestamp) as created_at, MAX(step) as last_step, COUNT(*) as log_count FROM metrics GROUP BY run_name ORDER BY created_at ASC""" ) rows = cursor.fetchall() runs_meta = [ { "id": row["run_name"], "name": row["run_name"], "created_at": row["created_at"], "last_step": row["last_step"], "log_count": row["log_count"], } for row in rows ] except sqlite3.OperationalError: runs_meta = [] with open(output_dir / "runs.json", "w") as f: json_mod.dump(runs_meta, f) settings = { "color_palette": get_color_palette(), "plot_order": [ item.strip() for item in os.environ.get("TRACKIO_PLOT_ORDER", "").split(",") if item.strip() ], } with open(output_dir / "settings.json", "w") as f: json_mod.dump(settings, f) @staticmethod def _cleanup_wal_sidecars(db_path: Path) -> None: """Remove leftover -wal/-shm files for a DB basename (prevents disk I/O errors).""" for suffix in ("-wal", "-shm"): sidecar = Path(str(db_path) + suffix) try: if sidecar.exists(): sidecar.unlink() except Exception: pass @staticmethod def _project_parquet_paths(db_path: Path) -> list[Path]: """Every parquet file that export_to_parquet writes for a project: the metrics file plus the system/configs/traces and per-artifact-table sidecars. Deleting these alongside the .db removes the project's durable form, so import_from_parquet cannot resurrect it from a lingering sidecar. """ stem = db_path.stem suffixes = [ ".parquet", "_system.parquet", "_configs.parquet", "_traces.parquet", ] suffixes += [ f"_{table}.parquet" for table in SQLiteStorage._ARTIFACT_PARQUET_TABLES ] return [db_path.with_name(f"{stem}{suffix}") for suffix in suffixes] @staticmethod def import_from_parquet(): """ Imports to all DB files that have matching files under the same path but with extension ".parquet". Also imports system_metrics from "_system.parquet" files. Also imports configs from "_configs.parquet" files. """ if not TRACKIO_DIR.exists(): return all_paths = os.listdir(TRACKIO_DIR) all_paths_set = set(all_paths) def _artifact_sidecar(pq_name: str) -> tuple[str, str] | None: for table in SQLiteStorage._ARTIFACT_PARQUET_TABLES: suffix = f"_{table}.parquet" if not pq_name.endswith(suffix) or len(pq_name) <= len(suffix): continue base = pq_name[: -len(suffix)] if ( f"{base}.parquet" in all_paths_set or (TRACKIO_DIR / f"{base}{DB_EXT}").exists() or f"{base}_artifact_versions.parquet" in all_paths_set ): return base, table return None parquet_names = [ f for f in all_paths if f.endswith(".parquet") and not f.endswith("_system.parquet") and not f.endswith("_configs.parquet") and not f.endswith("_traces.parquet") and _artifact_sidecar(f) is None ] imported_projects = {Path(name).stem for name in parquet_names} for pq_name in parquet_names: parquet_path = TRACKIO_DIR / pq_name db_path = parquet_path.with_suffix(DB_EXT) SQLiteStorage._cleanup_wal_sidecars(db_path) rows = SQLiteStorage._read_parquet_rows(parquet_path) project = db_path.stem SQLiteStorage.init_db(project) metrics_rows = SQLiteStorage._rows_to_sql_table_rows( rows, json_col="metrics", structural_cols=[ "id", "run_id", "timestamp", "run_name", "step", "log_id", "space_id", ], ) SQLiteStorage._replace_table_rows( db_path, "metrics", metrics_rows, [ "id", "run_id", "timestamp", "run_name", "step", "metrics", "log_id", "space_id", ], ) system_parquet_names = [f for f in all_paths if f.endswith("_system.parquet")] for pq_name in system_parquet_names: parquet_path = TRACKIO_DIR / pq_name db_name = pq_name.replace("_system.parquet", DB_EXT) db_path = TRACKIO_DIR / db_name project_name = db_path.stem if project_name not in imported_projects and not db_path.exists(): continue rows = SQLiteStorage._read_parquet_rows(parquet_path) SQLiteStorage.init_db(project_name) system_rows = SQLiteStorage._rows_to_sql_table_rows( rows, json_col="metrics", structural_cols=[ "id", "run_id", "timestamp", "run_name", "log_id", "space_id", ], ) SQLiteStorage._replace_table_rows( db_path, "system_metrics", system_rows, [ "id", "run_id", "timestamp", "run_name", "metrics", "log_id", "space_id", ], ) configs_parquet_names = [f for f in all_paths if f.endswith("_configs.parquet")] for pq_name in configs_parquet_names: parquet_path = TRACKIO_DIR / pq_name db_name = pq_name.replace("_configs.parquet", DB_EXT) db_path = TRACKIO_DIR / db_name project_name = db_path.stem if project_name not in imported_projects and not db_path.exists(): continue rows = SQLiteStorage._read_parquet_rows(parquet_path) SQLiteStorage.init_db(project_name) config_rows = SQLiteStorage._rows_to_sql_table_rows( rows, json_col="config", structural_cols=["id", "run_id", "run_name", "created_at"], ) SQLiteStorage._replace_table_rows( db_path, "configs", config_rows, ["id", "run_id", "run_name", "config", "created_at"], ) traces_parquet_names = [f for f in all_paths if f.endswith("_traces.parquet")] for pq_name in traces_parquet_names: parquet_path = TRACKIO_DIR / pq_name db_name = pq_name.replace("_traces.parquet", DB_EXT) db_path = TRACKIO_DIR / db_name project_name = db_path.stem if project_name not in imported_projects and not db_path.exists(): continue rows = SQLiteStorage._read_parquet_rows(parquet_path) SQLiteStorage.init_db(project_name) SQLiteStorage._replace_table_rows( db_path, "traces", rows, [ "id", "run_id", "timestamp", "run_name", "step", "key", "trace_index", "messages", "metadata", "search_text", "log_id", "space_id", ], ) for pq_name in all_paths: sidecar = _artifact_sidecar(pq_name) if sidecar is None: continue project_name, table = sidecar columns = SQLiteStorage._ARTIFACT_PARQUET_TABLES[table] parquet_path = TRACKIO_DIR / pq_name db_path = TRACKIO_DIR / f"{project_name}{DB_EXT}" rows = SQLiteStorage._read_parquet_rows(parquet_path) SQLiteStorage.init_db(project_name) SQLiteStorage._replace_table_rows(db_path, table, rows, columns) @staticmethod def get_scheduler(): """ Get the scheduler for the database based on the environment variables. This applies to both local and Spaces. """ with SQLiteStorage._scheduler_lock: if SQLiteStorage._current_scheduler is not None: return SQLiteStorage._current_scheduler hf_token = os.environ.get("HF_TOKEN") dataset_id = os.environ.get("TRACKIO_DATASET_ID") space_repo_name = os.environ.get("SPACE_REPO_NAME") if dataset_id is not None and space_repo_name is not None: scheduler = CommitScheduler( repo_id=dataset_id, repo_type="dataset", folder_path=TRACKIO_DIR, private=True, allow_patterns=[ "*.parquet", "*_system.parquet", "*_configs.parquet", "*_traces.parquet", "aux/*.parquet", "media/**/*", "artifacts/**/*", ], squash_history=True, token=hf_token, on_before_commit=SQLiteStorage.export_to_parquet, ) else: scheduler = DummyCommitScheduler() SQLiteStorage._current_scheduler = scheduler return scheduler @staticmethod def log( project: str, run: str, metrics: dict, step: int | None = None, run_id: str | None = None, ): """ Safely log metrics to the database. Before logging, this method will ensure the database exists and is set up with the correct tables. It also uses a cross-process lock to prevent database locking errors when multiple processes access the same database. This method is not used in the latest versions of Trackio (replaced by bulk_log) but is kept for backwards compatibility for users who are connecting to a newer version of a Trackio Spaces dashboard with an older version of Trackio installed locally. """ db_path = SQLiteStorage.init_db(project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() supports_run_ids = SQLiteStorage._supports_run_ids(conn) resolved_run_id = run_id or run run_col = "run_id" if supports_run_ids else "run_name" cursor.execute( f""" SELECT MAX(step) FROM metrics WHERE {run_col} = ? """, (resolved_run_id if supports_run_ids else run,), ) last_step = cursor.fetchone()[0] current_step = ( 0 if step is None and last_step is None else (step if step is not None else last_step + 1) ) current_timestamp = datetime.now(timezone.utc).isoformat() clean_metrics, trace_rows = SQLiteStorage._split_trace_metrics( metrics, run=run, run_id=resolved_run_id, step=current_step, timestamp=current_timestamp, log_id=None, space_id=None, ) if supports_run_ids: cursor.execute( """ INSERT INTO metrics (timestamp, run_id, run_name, step, metrics) VALUES (?, ?, ?, ?, ?) """, ( current_timestamp, resolved_run_id, run, current_step, orjson.dumps(serialize_values(clean_metrics)), ), ) else: cursor.execute( """ INSERT INTO metrics (timestamp, run_name, step, metrics) VALUES (?, ?, ?, ?) """, ( current_timestamp, run, current_step, orjson.dumps(serialize_values(clean_metrics)), ), ) SQLiteStorage._insert_trace_rows(cursor, trace_rows) conn.commit() @staticmethod def bulk_log( project: str, run: str, metrics_list: list[dict], steps: list[int] | None = None, timestamps: list[str] | None = None, config: dict | None = None, log_ids: list[str] | None = None, space_id: str | None = None, run_id: str | None = None, ): """ Safely log bulk metrics to the database. Before logging, this method will ensure the database exists and is set up with the correct tables. It also uses a cross-process lock to prevent database locking errors when multiple processes access the same database. """ if not metrics_list: return if timestamps is None: timestamps = [datetime.now(timezone.utc).isoformat()] * len(metrics_list) db_path = SQLiteStorage.init_db(project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() supports_run_ids = SQLiteStorage._supports_run_ids(conn) resolved_run_id = run_id or run if steps is None: steps = list(range(len(metrics_list))) elif any(s is None for s in steps): run_col = "run_id" if supports_run_ids else "run_name" cursor.execute( f"SELECT MAX(step) FROM metrics WHERE {run_col} = ?", (resolved_run_id if supports_run_ids else run,), ) last_step = cursor.fetchone()[0] current_step = 0 if last_step is None else last_step + 1 processed_steps = [] for step in steps: if step is None: processed_steps.append(current_step) current_step += 1 else: processed_steps.append(step) steps = processed_steps if len(metrics_list) != len(steps) or len(metrics_list) != len( timestamps ): raise ValueError( "metrics_list, steps, and timestamps must have the same length" ) data = [] trace_rows = [] for i, metrics in enumerate(metrics_list): lid = log_ids[i] if log_ids else None clean_metrics, rows = SQLiteStorage._split_trace_metrics( metrics, run=run, run_id=resolved_run_id, step=steps[i], timestamp=timestamps[i], log_id=lid, space_id=space_id, ) trace_rows.extend(rows) if supports_run_ids: data.append( ( timestamps[i], resolved_run_id, run, steps[i], orjson.dumps(serialize_values(clean_metrics)), lid, space_id, ) ) else: data.append( ( timestamps[i], run, steps[i], orjson.dumps(serialize_values(clean_metrics)), lid, space_id, ) ) if supports_run_ids: cursor.executemany( """ INSERT OR IGNORE INTO metrics (timestamp, run_id, run_name, step, metrics, log_id, space_id) VALUES (?, ?, ?, ?, ?, ?, ?) """, data, ) else: cursor.executemany( """ INSERT OR IGNORE INTO metrics (timestamp, run_name, step, metrics, log_id, space_id) VALUES (?, ?, ?, ?, ?, ?) """, data, ) SQLiteStorage._insert_trace_rows(cursor, trace_rows) if config: current_timestamp = datetime.now(timezone.utc).isoformat() if "run_id" in SQLiteStorage._table_columns(conn, "configs"): cursor.execute( """ INSERT OR REPLACE INTO configs (run_id, run_name, config, created_at) VALUES (?, ?, ?, ?) """, ( resolved_run_id, run, orjson.dumps(serialize_values(config)), current_timestamp, ), ) else: cursor.execute( """ INSERT OR REPLACE INTO configs (run_name, config, created_at) VALUES (?, ?, ?) """, ( run, orjson.dumps(serialize_values(config)), current_timestamp, ), ) conn.commit() @staticmethod def bulk_log_system( project: str, run: str, metrics_list: list[dict], timestamps: list[str] | None = None, log_ids: list[str] | None = None, space_id: str | None = None, run_id: str | None = None, ): """ Log system metrics (GPU, etc.) to the database without step numbers. These metrics use timestamps for the x-axis instead of steps. """ if not metrics_list: return if timestamps is None: timestamps = [datetime.now(timezone.utc).isoformat()] * len(metrics_list) if len(metrics_list) != len(timestamps): raise ValueError("metrics_list and timestamps must have the same length") db_path = SQLiteStorage.init_db(project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() supports_run_ids = SQLiteStorage._supports_run_ids( conn, "system_metrics" ) resolved_run_id = run_id or run data = [] for i, metrics in enumerate(metrics_list): lid = log_ids[i] if log_ids else None if supports_run_ids: data.append( ( timestamps[i], resolved_run_id, run, orjson.dumps(serialize_values(metrics)), lid, space_id, ) ) else: data.append( ( timestamps[i], run, orjson.dumps(serialize_values(metrics)), lid, space_id, ) ) if supports_run_ids: cursor.executemany( """ INSERT OR IGNORE INTO system_metrics (timestamp, run_id, run_name, metrics, log_id, space_id) VALUES (?, ?, ?, ?, ?, ?) """, data, ) else: cursor.executemany( """ INSERT OR IGNORE INTO system_metrics (timestamp, run_name, metrics, log_id, space_id) VALUES (?, ?, ?, ?, ?) """, data, ) conn.commit() @staticmethod def bulk_alert( project: str, run: str, titles: list[str], texts: list[str | None], levels: list[str], steps: list[int | None], timestamps: list[str] | None = None, alert_ids: list[str] | None = None, run_id: str | None = None, ): if not titles: return if timestamps is None: timestamps = [datetime.now(timezone.utc).isoformat()] * len(titles) db_path = SQLiteStorage.init_db(project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() supports_run_ids = SQLiteStorage._supports_run_ids(conn, "alerts") resolved_run_id = run_id or run data = [] for i in range(len(titles)): aid = alert_ids[i] if alert_ids else None if supports_run_ids: data.append( ( resolved_run_id, timestamps[i], run, titles[i], texts[i], levels[i], steps[i], aid, ) ) else: data.append( ( timestamps[i], run, titles[i], texts[i], levels[i], steps[i], aid, ) ) if supports_run_ids: cursor.executemany( """ INSERT OR IGNORE INTO alerts (run_id, timestamp, run_name, title, text, level, step, alert_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, data, ) else: cursor.executemany( """ INSERT OR IGNORE INTO alerts (timestamp, run_name, title, text, level, step, alert_id) VALUES (?, ?, ?, ?, ?, ?, ?) """, data, ) conn.commit() @staticmethod def get_alerts( project: str, run_name: str | None = None, run_id: str | None = None, level: str | None = None, since: str | None = None, ) -> list[dict]: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: query = ( "SELECT timestamp, run_name, title, text, level, step FROM alerts" ) conditions = [] params = [] run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run_name, run_id=run_id, table="alerts" ) if run_identity is not None: conditions.append(f"{run_identity[0]} = ?") params.append(run_identity[1]) elif run_name is not None or run_id is not None: return [] if level is not None: conditions.append("level = ?") params.append(level) if since is not None: conditions.append("timestamp > ?") params.append(since) if conditions: query += " WHERE " + " AND ".join(conditions) query += " ORDER BY timestamp DESC" cursor.execute(query, params) rows = cursor.fetchall() return [ { "timestamp": row["timestamp"], "run": row["run_name"], "title": row["title"], "text": row["text"], "level": row["level"], "step": row["step"], } for row in rows ] except sqlite3.OperationalError as e: if "no such table: alerts" in str(e): return [] raise @staticmethod def get_alert_count(project: str) -> int: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return 0 with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: cursor.execute("SELECT COUNT(*) FROM alerts") return cursor.fetchone()[0] except sqlite3.OperationalError: return 0 @staticmethod def _fetch_system_logs_with_cursor( cursor: sqlite3.Cursor, run_identity: tuple[str, Any], max_points: int | None = None, ) -> list[dict[str, Any]]: cursor.execute( f""" SELECT timestamp, metrics FROM system_metrics WHERE {run_identity[0]} = ? ORDER BY timestamp """, (run_identity[1],), ) rows = cursor.fetchall() rows = SQLiteStorage._subsample_metric_rows(rows, max_points) results = [] for row in rows: metrics = orjson.loads(row["metrics"]) metrics = deserialize_values(metrics) metrics["timestamp"] = row["timestamp"] results.append(metrics) return results @staticmethod def get_system_logs( project: str, run: str | None = None, run_id: str | None = None, max_points: int | None = None, ) -> list[dict]: """Retrieve system metrics for a specific run. Returns metrics with timestamps (no steps).""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] cache_key = _system_logs_read_cache_key(project, run, run_id, max_points) cached = _system_logs_read_cache_get(db_path, cache_key) if cached is not None: return cached try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="system_metrics" ) if run_identity is None: logs: list[dict[str, Any]] = [] else: logs = SQLiteStorage._fetch_system_logs_with_cursor( cursor, run_identity, max_points ) except sqlite3.OperationalError as e: if "no such table: system_metrics" in str(e): return [] raise _system_logs_read_cache_put(db_path, cache_key, logs) return [{**d} for d in logs] @staticmethod def get_system_logs_batch( project: str, runs: list[dict[str, Any]] | None = None, max_points: int | None = None, ) -> list[dict[str, Any]]: if not runs: return [] db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [ { "run": r.get("run"), "run_id": r.get("run_id"), "logs": [], } for r in runs ] out: list[dict[str, Any]] = [] try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() for r in runs: run = r.get("run") run_id = r.get("run_id") cache_key = _system_logs_read_cache_key( project, run, run_id, max_points ) cached = _system_logs_read_cache_get(db_path, cache_key) if cached is not None: out.append( { "run": run, "run_id": run_id, "logs": cached, } ) continue run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="system_metrics" ) if run_identity is None: logs = [] else: logs = SQLiteStorage._fetch_system_logs_with_cursor( cursor, run_identity, max_points ) _system_logs_read_cache_put(db_path, cache_key, logs) out.append( { "run": run, "run_id": run_id, "logs": [{**d} for d in logs], } ) except sqlite3.OperationalError as e: if "no such table: system_metrics" in str(e): return [ { "run": r.get("run"), "run_id": r.get("run_id"), "logs": [], } for r in runs ] raise return out @staticmethod def get_all_system_metrics_for_run( project: str, run: str | None = None, run_id: str | None = None ) -> list[str]: """Get all system metric names for a specific project/run.""" return SQLiteStorage._get_metric_names( project, run, "system_metrics", exclude_keys={"timestamp"}, run_id=run_id, ) @staticmethod def has_system_metrics(project: str) -> bool: """Check if a project has any system metrics logged.""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return False with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: cursor.execute("SELECT COUNT(*) FROM system_metrics LIMIT 1") count = cursor.fetchone()[0] return count > 0 except sqlite3.OperationalError: return False @staticmethod def get_log_count( project: str, run: str | None = None, run_id: str | None = None ) -> int: SQLiteStorage._ensure_hub_loaded() db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return 0 try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id ) if run_identity is None: return 0 cursor.execute( f"SELECT COUNT(*) FROM metrics WHERE {run_identity[0]} = ?", (run_identity[1],), ) return cursor.fetchone()[0] except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return 0 raise @staticmethod def get_last_step( project: str, run: str | None = None, run_id: str | None = None ) -> int | None: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id ) if run_identity is None: return None cursor.execute( f"SELECT MAX(step) FROM metrics WHERE {run_identity[0]} = ?", (run_identity[1],), ) row = cursor.fetchone() return row[0] if row and row[0] is not None else None except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return None raise @staticmethod def get_tab_availability_flags(project: str) -> dict[str, bool]: SQLiteStorage._ensure_hub_loaded() flags = { "metrics": False, "system": False, "traces": False, "media": False, "reports": False, "alerts": False, } db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return flags def _exists(conn, sql, params=()): try: cursor = conn.cursor() cursor.execute(sql, params) return cursor.fetchone() is not None except sqlite3.OperationalError: return False with SQLiteStorage._get_connection(db_path) as conn: flags["metrics"] = _exists( conn, "SELECT 1 FROM metrics " "WHERE CAST(metrics AS TEXT) GLOB '*:[0-9]*' " "OR CAST(metrics AS TEXT) GLOB '*:-[0-9]*' " "LIMIT 1", ) flags["media"] = _exists( conn, "SELECT 1 FROM metrics WHERE " "CAST(metrics AS TEXT) GLOB ? " "OR CAST(metrics AS TEXT) GLOB ? " "OR CAST(metrics AS TEXT) GLOB ? " "OR CAST(metrics AS TEXT) GLOB ? " "LIMIT 1", ( '*"_type":"trackio.image"*', '*"_type":"trackio.video"*', '*"_type":"trackio.audio"*', '*"_type":"trackio.table"*', ), ) flags["reports"] = _exists( conn, "SELECT 1 FROM metrics WHERE CAST(metrics AS TEXT) GLOB ? LIMIT 1", ('*"_type":"trackio.markdown"*',), ) flags["system"] = _exists(conn, "SELECT 1 FROM system_metrics LIMIT 1") flags["traces"] = _exists(conn, "SELECT 1 FROM traces LIMIT 1") flags["alerts"] = _exists(conn, "SELECT 1 FROM alerts LIMIT 1") return flags @staticmethod def _subsample_metric_rows(rows: list[Any], max_points: int | None) -> list[Any]: if max_points is None or max_points < 1: return rows if len(rows) <= max_points: return rows step = len(rows) / max_points indices = {int(i * step) for i in range(max_points)} indices.add(len(rows) - 1) return [rows[i] for i in sorted(indices)] @staticmethod def _metric_rows_to_log_dicts(rows: list[Any]) -> list[dict[str, Any]]: results = [] for row in rows: metrics = orjson.loads(row["metrics"]) metrics = deserialize_values(metrics) metrics["timestamp"] = row["timestamp"] metrics["step"] = row["step"] results.append(metrics) return results @staticmethod def _fetch_metric_logs_with_cursor( cursor: sqlite3.Cursor, run_identity: tuple[str, Any], max_points: int | None, ) -> list[dict[str, Any]]: cursor.execute( f""" SELECT timestamp, step, metrics FROM metrics WHERE {run_identity[0]} = ? ORDER BY timestamp """, (run_identity[1],), ) rows = cursor.fetchall() rows = SQLiteStorage._subsample_metric_rows(rows, max_points) return SQLiteStorage._metric_rows_to_log_dicts(rows) @staticmethod def get_logs( project: str, run: str | None = None, max_points: int | None = None, run_id: str | None = None, ) -> list[dict]: """Retrieve logs for a specific run. Logs include the step count (int) and the timestamp (datetime object).""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] cache_key = _logs_read_cache_key(project, run, run_id, max_points) cached = _logs_read_cache_get(db_path, cache_key) if cached is not None: return cached try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id ) if run_identity is None: logs: list[dict[str, Any]] = [] else: logs = SQLiteStorage._fetch_metric_logs_with_cursor( cursor, run_identity, max_points ) except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return [] raise _logs_read_cache_put(db_path, cache_key, logs) return [{**d} for d in logs] @staticmethod def get_logs_batch( project: str, runs: list[dict[str, Any]] | None = None, max_points: int | None = None, ) -> list[dict[str, Any]]: if not runs: return [] db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [ { "run": r.get("run"), "run_id": r.get("run_id"), "logs": [], } for r in runs ] out: list[dict[str, Any]] = [] try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() for r in runs: run = r.get("run") run_id = r.get("run_id") cache_key = _logs_read_cache_key(project, run, run_id, max_points) cached = _logs_read_cache_get(db_path, cache_key) if cached is not None: out.append( { "run": run, "run_id": run_id, "logs": cached, } ) continue run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id ) if run_identity is None: logs = [] else: logs = SQLiteStorage._fetch_metric_logs_with_cursor( cursor, run_identity, max_points ) _logs_read_cache_put(db_path, cache_key, logs) out.append( { "run": run, "run_id": run_id, "logs": [{**d} for d in logs], } ) except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return [ { "run": r.get("run"), "run_id": r.get("run_id"), "logs": [], } for r in runs ] raise return out @staticmethod def _is_trace_payload(value: Any) -> bool: return isinstance(value, dict) and value.get("_type") == "trackio.trace" @staticmethod def _split_trace_metrics( metrics: dict, *, run: str, run_id: str, step: int, timestamp: str, log_id: str | None, space_id: str | None, ) -> tuple[dict, list[dict[str, Any]]]: clean_metrics = {} trace_rows: list[dict[str, Any]] = [] for key, value in metrics.items(): is_list = isinstance(value, list) candidates = value if is_list else [value] traces_for_key = [ (index if is_list else None, candidate) for index, candidate in enumerate(candidates) if SQLiteStorage._is_trace_payload(candidate) ] if not traces_for_key: clean_metrics[key] = value continue if is_list: non_trace_items = [ candidate for candidate in candidates if not SQLiteStorage._is_trace_payload(candidate) ] if non_trace_items: clean_metrics[key] = non_trace_items for trace_index, trace in traces_for_key: trace_id_parts = [run_id or run, log_id or uuid.uuid4().hex, key] if trace_index is not None: trace_id_parts.append(str(trace_index)) trace_record = { "id": ":".join(str(part) for part in trace_id_parts), "run_id": run_id, "timestamp": timestamp, "run_name": run, "step": step, "key": key, "trace_index": trace_index, "messages": trace.get("messages", []), "metadata": trace.get("metadata", {}), "log_id": log_id, "space_id": space_id, } trace_record["search_text"] = ( f"{trace_record['id']} {key} " f"{SQLiteStorage._flatten_trace_search_text(trace_record)}" ).lower() trace_rows.append(trace_record) return clean_metrics, trace_rows @staticmethod def _insert_trace_rows(cursor: sqlite3.Cursor, trace_rows: list[dict[str, Any]]): if not trace_rows: return cursor.executemany( """ INSERT OR IGNORE INTO traces (id, run_id, timestamp, run_name, step, key, trace_index, messages, metadata, search_text, log_id, space_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( row["id"], row["run_id"], row["timestamp"], row["run_name"], row["step"], row["key"], row["trace_index"], orjson.dumps(serialize_values(row["messages"])), orjson.dumps(serialize_values(row["metadata"])), row["search_text"], row["log_id"], row["space_id"], ) for row in trace_rows ], ) @staticmethod def _flatten_trace_search_text(trace: dict[str, Any]) -> str: parts: list[str] = [] def visit(value: Any): if value is None: return if isinstance(value, dict): for nested in value.values(): visit(nested) return if isinstance(value, list): for nested in value: visit(nested) return parts.append(str(value)) visit(trace.get("messages", [])) visit(trace.get("metadata", {})) return " ".join(parts).lower() @staticmethod def _extract_traces_from_logs( logs: list[dict[str, Any]], run: str | None, run_id: str | None, ) -> list[dict[str, Any]]: traces: list[dict[str, Any]] = [] for log in logs: step = log.get("step") timestamp = log.get("timestamp") for key, value in log.items(): if key in {"step", "timestamp"}: continue candidates = value if isinstance(value, list) else [value] for index, candidate in enumerate(candidates): if ( not isinstance(candidate, dict) or candidate.get("_type") != "trackio.trace" ): continue trace_index = index if isinstance(value, list) else None trace_id_parts = [run_id or run or "run", str(step), key] if trace_index is not None: trace_id_parts.append(str(trace_index)) trace_record = { "id": ":".join(trace_id_parts), "key": key, "index": trace_index, "run": run, "run_id": run_id, "step": step, "timestamp": timestamp, "messages": candidate.get("messages", []), "metadata": candidate.get("metadata", {}), } trace_record["_search_text"] = ( f"{trace_record['id']} {key} " f"{SQLiteStorage._flatten_trace_search_text(trace_record)}" ).lower() traces.append(trace_record) return traces @staticmethod def _sort_traces( traces: list[dict[str, Any]], sort: str | None ) -> list[dict[str, Any]]: sort_key = sort or "request_time_desc" if sort_key == "step_asc": return sorted(traces, key=lambda trace: trace.get("step") or 0) if sort_key == "step_desc": return sorted( traces, key=lambda trace: trace.get("step") or 0, reverse=True ) if sort_key == "request_time_asc": return sorted(traces, key=lambda trace: trace.get("timestamp") or "") return sorted( traces, key=lambda trace: trace.get("timestamp") or "", reverse=True ) @staticmethod def get_traces( project: str, run: str | None = None, search: str | None = None, sort: str | None = None, limit: int | None = None, offset: int = 0, run_id: str | None = None, step: int | None = None, ) -> list[dict[str, Any]]: try: offset = max(0, int(offset or 0)) except (TypeError, ValueError): offset = 0 if limit is not None: try: limit = max(0, int(limit)) except (TypeError, ValueError): limit = None db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] order_by = { "step_asc": "step ASC, timestamp ASC, id ASC", "step_desc": "step DESC, timestamp DESC, id DESC", "request_time_asc": "timestamp ASC, id ASC", "request_time_desc": "timestamp DESC, id DESC", }.get(sort or "request_time_desc", "timestamp DESC, id DESC") try: with SQLiteStorage._get_connection(db_path) as conn: run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="traces" ) if run_identity is None: return [] where = [f"{run_identity[0]} = ?"] params: list[Any] = [run_identity[1]] if step is not None: where.append("step = ?") params.append(step) if search: needle = search.strip().lower() if needle: where.append("search_text LIKE ?") params.append(f"%{needle}%") query = f""" SELECT id, key, trace_index, run_name, run_id, step, timestamp, messages, metadata FROM traces WHERE {" AND ".join(where)} ORDER BY {order_by} """ if limit is not None: query += " LIMIT ?" params.append(limit) if offset > 0: if limit is None: query += " LIMIT -1" query += " OFFSET ?" params.append(offset) cursor = conn.cursor() cursor.execute(query, params) rows = cursor.fetchall() except sqlite3.OperationalError as e: if "no such table: traces" in str(e): return [] raise return [ { "id": row["id"], "key": row["key"], "index": row["trace_index"], "run": row["run_name"], "run_id": row["run_id"], "step": row["step"], "timestamp": row["timestamp"], "messages": deserialize_values(orjson.loads(row["messages"])), "metadata": deserialize_values(orjson.loads(row["metadata"])), } for row in rows ] @staticmethod def get_trace_steps( project: str, run: str | None = None, run_id: str | None = None, ) -> dict[str, Any]: """Return per-step trace counts and total for a run. Returns: {"total": int, "steps": [{"step": int, "count": int}, ...]} Steps are returned in ascending order. """ db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return {"total": 0, "steps": []} try: with SQLiteStorage._get_connection(db_path) as conn: run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="traces" ) if run_identity is None: return {"total": 0, "steps": []} cursor = conn.cursor() cursor.execute( f""" SELECT step, COUNT(*) AS c FROM traces WHERE {run_identity[0]} = ? GROUP BY step ORDER BY step ASC """, (run_identity[1],), ) rows = cursor.fetchall() except sqlite3.OperationalError as e: if "no such table: traces" in str(e): return {"total": 0, "steps": []} raise steps = [{"step": row["step"], "count": row["c"]} for row in rows] total = sum(item["count"] for item in steps) return {"total": total, "steps": steps} @staticmethod def load_from_dataset(): """Hydrate the local trackio dir from the configured remote. Wrapped in a per-thread reentrancy guard (`_dataset_loading.active`) because the import path re-enters via `import_from_parquet` -> `init_db` -> `_ensure_hub_loaded`, and the scheduler lock held during import is not reentrant. """ if getattr(SQLiteStorage._dataset_loading, "active", False): return SQLiteStorage._dataset_loading.active = True try: SQLiteStorage._load_from_dataset_impl() finally: SQLiteStorage._dataset_loading.active = False @staticmethod def _load_from_dataset_impl(): """Download the remote files, then import them, at most once each. Three class flags coordinate this across calls: - `_dataset_remote_synced`: the remote listing + download has run; kept across calls so a retry never re-hits the Hub. - `_dataset_import_pending`: files were downloaded but not yet imported into SQLite; stays True until `import_from_parquet` succeeds, so a transient import failure is retried on the next access. - `_dataset_import_attempted`: the load has run to completion (or been short-circuited); gates `_ensure_hub_loaded` and `export_to_parquet`. """ bucket_id = os.environ.get("TRACKIO_BUCKET_ID") if bucket_id is not None: if not SQLiteStorage._dataset_import_attempted: SQLiteStorage._dataset_import_attempted = True from trackio import fragments from trackio.bucket_storage import download_bucket_to_trackio_dir try: download_bucket_to_trackio_dir(bucket_id) except Exception: pass try: fragments.import_inbox_from_bucket(bucket_id) except Exception: pass try: fragments.import_inbox_dir() except Exception: pass return dataset_id = os.environ.get("TRACKIO_DATASET_ID") space_repo_name = os.environ.get("SPACE_REPO_NAME") if dataset_id is not None and space_repo_name is not None: hfapi = hf.HfApi() if not TRACKIO_DIR.exists(): TRACKIO_DIR.mkdir(parents=True, exist_ok=True) with SQLiteStorage.get_scheduler().lock: if not SQLiteStorage._dataset_remote_synced: try: files = hfapi.list_repo_files(dataset_id, repo_type="dataset") for file in files: if not ( file.endswith(".parquet") or file.startswith("media/") or file.startswith("artifacts/") ): continue if (TRACKIO_DIR / file).exists(): continue hf.hf_hub_download( dataset_id, file, repo_type="dataset", local_dir=TRACKIO_DIR, ) SQLiteStorage._dataset_import_pending = True except hf.errors.EntryNotFoundError: pass except hf.errors.RepositoryNotFoundError: pass SQLiteStorage._dataset_remote_synced = True if SQLiteStorage._dataset_import_pending: try: SQLiteStorage.import_from_parquet() SQLiteStorage._dataset_import_pending = False except Exception as e: _emit_nonfatal_warning( f"trackio could not import downloaded dataset files; " f"will retry on the next access: {e}" ) return SQLiteStorage._dataset_import_attempted = True @staticmethod def _ensure_hub_loaded(): if not SQLiteStorage._dataset_import_attempted: SQLiteStorage.load_from_dataset() @staticmethod def get_projects() -> list[str]: """ Get list of all projects by scanning the database files in the trackio directory. """ SQLiteStorage._ensure_hub_loaded() projects: set[str] = set() if not TRACKIO_DIR.exists(): return [] for db_file in TRACKIO_DIR.glob(f"*{DB_EXT}"): project_name = db_file.stem projects.add(project_name) return sorted(projects) @staticmethod def get_runs(project: str) -> list[str]: """Get list of all runs for a project, ordered by creation time.""" return [record["name"] for record in SQLiteStorage.get_run_records(project)] @staticmethod def _validate_read_only_query(query: str) -> str: normalized = query.strip().rstrip(";").strip() if not normalized: raise ValueError("Query cannot be empty.") if not normalized.lower().startswith(_READ_ONLY_QUERY_PREFIXES): raise ValueError( "Only read-only SELECT, WITH, and safe PRAGMA queries are supported." ) return normalized @staticmethod def _query_authorizer( action_code: int, arg1: str | None, arg2: str | None, db_name: str | None, source: str | None, ) -> int: del arg2, db_name, source if action_code in { sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_FUNCTION, }: return sqlite3.SQLITE_OK pragma_code = getattr(sqlite3, "SQLITE_PRAGMA", None) if action_code == pragma_code: pragma_name = (arg1 or "").lower() if pragma_name in _READ_ONLY_PRAGMAS: return sqlite3.SQLITE_OK return sqlite3.SQLITE_DENY @staticmethod def _normalize_query_value(value: Any) -> Any: if isinstance(value, (bytes, bytearray, memoryview)): return bytes(value).hex() return value @staticmethod def query_project( project: str, query: str, max_rows: int = _QUERY_MAX_ROWS ) -> dict[str, Any]: SQLiteStorage._ensure_hub_loaded() db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): raise FileNotFoundError(f"Project '{project}' not found.") normalized_query = SQLiteStorage._validate_read_only_query(query) with SQLiteStorage._get_connection(db_path) as conn: conn.set_authorizer(SQLiteStorage._query_authorizer) try: cursor = conn.cursor() cursor.execute(normalized_query) description = cursor.description or [] columns = [column[0] for column in description] fetched = cursor.fetchmany(max_rows + 1) if len(fetched) > max_rows: raise ValueError( f"Query returned more than {max_rows} rows. " "Refine the query or add a LIMIT clause." ) rows = [ { column: SQLiteStorage._normalize_query_value(row[column]) for column in columns } for row in fetched ] except sqlite3.DatabaseError as e: raise ValueError(str(e)) from e finally: conn.set_authorizer(None) return { "project": project, "query": normalized_query, "columns": columns, "rows": rows, "row_count": len(rows), } @staticmethod def get_max_steps_for_runs(project: str) -> dict[str, int]: """Get the maximum step for each run in a project.""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return {} try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() if SQLiteStorage._supports_run_ids(conn): cursor.execute( """ SELECT run_name, run_id, MAX(step) as max_step FROM metrics GROUP BY run_id, run_name """ ) results = {} for row in cursor.fetchall(): results[row["run_id"]] = row["max_step"] return results cursor.execute( """ SELECT run_name, MAX(step) as max_step FROM metrics GROUP BY run_name """ ) results = {} for row in cursor.fetchall(): results[row["run_name"]] = row["max_step"] return results except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return {} raise @staticmethod def get_max_step_for_run( project: str, run: str | None = None, run_id: str | None = None ) -> int | None: """Get the maximum step for a specific run, or None if no logs exist.""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None try: with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id ) if run_identity is None: return None cursor.execute( f"SELECT MAX(step) FROM metrics WHERE {run_identity[0]} = ?", (run_identity[1],), ) result = cursor.fetchone()[0] return result except sqlite3.OperationalError as e: if "no such table: metrics" in str(e): return None raise @staticmethod def get_run_config( project: str, run: str | None = None, run_id: str | None = None ) -> dict | None: """Get configuration for a specific run.""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="metrics" ) if run_identity is None: return None config_col = ( "run_id" if "run_id" in SQLiteStorage._table_columns(conn, "configs") else "run_name" ) cursor.execute( f""" SELECT config FROM configs WHERE {config_col} = ? """, (run_identity[1],), ) row = cursor.fetchone() if row: config = orjson.loads(row["config"]) return deserialize_values(config) return None except sqlite3.OperationalError as e: if "no such table: configs" in str(e): return None raise @staticmethod def delete_run(project: str, run: str, run_id: str | None = None) -> bool: """Delete a run from the database (metrics, config, and system_metrics).""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return False with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id ) if run_identity is None: return False cursor.execute( f"DELETE FROM metrics WHERE {run_identity[0]} = ?", (run_identity[1],), ) config_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="configs" ) if config_identity is not None: cursor.execute( f"DELETE FROM configs WHERE {config_identity[0]} = ?", (config_identity[1],), ) try: cursor.execute( f"DELETE FROM system_metrics WHERE {run_identity[0]} = ?", (run_identity[1],), ) except sqlite3.OperationalError: pass try: cursor.execute( f"DELETE FROM alerts WHERE {run_identity[0]} = ?", (run_identity[1],), ) except sqlite3.OperationalError: pass try: trace_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="traces" ) if trace_identity is not None: cursor.execute( f"DELETE FROM traces WHERE {trace_identity[0]} = ?", (trace_identity[1],), ) except sqlite3.OperationalError: pass conn.commit() return True except sqlite3.Error: return False @staticmethod def _update_media_paths(obj, old_prefix, new_prefix): """Update media file paths in nested data structures.""" if isinstance(obj, dict): if obj.get("_type") in [ "trackio.image", "trackio.video", "trackio.audio", ]: old_path = obj.get("file_path", "") if isinstance(old_path, str): normalized_path = old_path.replace("\\", "/") if normalized_path.startswith(old_prefix): new_path = normalized_path.replace(old_prefix, new_prefix, 1) return {**obj, "file_path": new_path} return { key: SQLiteStorage._update_media_paths(value, old_prefix, new_prefix) for key, value in obj.items() } elif isinstance(obj, list): return [ SQLiteStorage._update_media_paths(item, old_prefix, new_prefix) for item in obj ] return obj @staticmethod def _rewrite_metrics_rows( metrics_rows, new_run_name, old_prefix, new_prefix, include_run_id=False ): """Deserialize metrics rows, update media paths, and reserialize.""" result = [] for row in metrics_rows: metrics_data = orjson.loads(row["metrics"]) metrics_deserialized = deserialize_values(metrics_data) updated = SQLiteStorage._update_media_paths( metrics_deserialized, old_prefix, new_prefix ) values = ( row["timestamp"], new_run_name, row["step"], orjson.dumps(serialize_values(updated)), ) if include_run_id: values = values + (row["run_id"],) result.append(values) return result @staticmethod def _rewrite_trace_rows( trace_rows, new_run_name, old_prefix, new_prefix, *, run_id: str | None = None, ): result = [] for row in trace_rows: messages = deserialize_values(orjson.loads(row["messages"])) metadata = deserialize_values(orjson.loads(row["metadata"])) messages = SQLiteStorage._update_media_paths( messages, old_prefix, new_prefix ) metadata = SQLiteStorage._update_media_paths( metadata, old_prefix, new_prefix ) result.append( ( row["id"], run_id if run_id is not None else row["run_id"], row["timestamp"], new_run_name, row["step"], row["key"], row["trace_index"], orjson.dumps(serialize_values(messages)), orjson.dumps(serialize_values(metadata)), row["search_text"], row["log_id"], row["space_id"], ) ) return result @staticmethod def _move_media_dir(source: Path, target: Path): """Move a media directory from source to target.""" if source.exists(): target.parent.mkdir(parents=True, exist_ok=True) if target.exists(): shutil.rmtree(target) shutil.move(str(source), str(target)) @staticmethod def rename_run( project: str, old_name: str, new_name: str, run_id: str | None = None ) -> None: """Rename a run within the same project. Raises: ValueError: If the new name is empty, the old run doesn't exist, or a run with the new name already exists. RuntimeError: If the database operation fails. """ if not new_name or not new_name.strip(): raise ValueError("New run name cannot be empty") new_name = new_name.strip() db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): raise ValueError(f"Project '{project}' does not exist") with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() supports_run_ids = SQLiteStorage._supports_run_ids(conn) run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=old_name, run_id=run_id ) if run_identity is None: raise ValueError( f"Run '{old_name}' does not exist in project '{project}'" ) run_col, run_value = run_identity if not supports_run_ids: cursor.execute( "SELECT COUNT(*) FROM metrics WHERE run_name = ?", (new_name,) ) if cursor.fetchone()[0] > 0: raise ValueError( f"A run named '{new_name}' already exists in project '{project}'" ) try: select_cols = ( "run_id, timestamp, step, metrics" if supports_run_ids else "timestamp, step, metrics" ) cursor.execute( f"SELECT {select_cols} FROM metrics WHERE {run_col} = ?", (run_value,), ) metrics_rows = cursor.fetchall() old_prefix = f"{project}/{old_name}/" new_prefix = f"{project}/{new_name}/" updated_rows = [] for row in metrics_rows: metrics_data = orjson.loads(row["metrics"]) metrics_deserialized = deserialize_values(metrics_data) updated = SQLiteStorage._update_media_paths( metrics_deserialized, old_prefix, new_prefix ) if supports_run_ids: updated_rows.append( ( row["run_id"], row["timestamp"], new_name, row["step"], orjson.dumps(serialize_values(updated)), ) ) else: updated_rows.append( ( row["timestamp"], new_name, row["step"], orjson.dumps(serialize_values(updated)), ) ) cursor.execute( f"DELETE FROM metrics WHERE {run_col} = ?", (run_value,) ) if supports_run_ids: cursor.executemany( "INSERT INTO metrics (run_id, timestamp, run_name, step, metrics) VALUES (?, ?, ?, ?, ?)", updated_rows, ) else: cursor.executemany( "INSERT INTO metrics (timestamp, run_name, step, metrics) VALUES (?, ?, ?, ?)", updated_rows, ) config_col = ( "run_id" if "run_id" in SQLiteStorage._table_columns(conn, "configs") else "run_name" ) cursor.execute( f"UPDATE configs SET run_name = ? WHERE {config_col} = ?", (new_name, run_value), ) try: cursor.execute( f"UPDATE system_metrics SET run_name = ? WHERE {run_col} = ?", (new_name, run_value), ) except sqlite3.OperationalError: pass try: cursor.execute( f"UPDATE alerts SET run_name = ? WHERE {run_col} = ?", (new_name, run_value), ) except sqlite3.OperationalError: pass try: cursor.execute( f""" SELECT id, run_id, timestamp, step, key, trace_index, messages, metadata, search_text, log_id, space_id FROM traces WHERE {run_col} = ? """, (run_value,), ) trace_rows = cursor.fetchall() updated_trace_rows = SQLiteStorage._rewrite_trace_rows( trace_rows, new_name, old_prefix, new_prefix, run_id=new_name if not supports_run_ids else None, ) cursor.execute( f"DELETE FROM traces WHERE {run_col} = ?", (run_value,) ) cursor.executemany( """ INSERT INTO traces (id, run_id, timestamp, run_name, step, key, trace_index, messages, metadata, search_text, log_id, space_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, updated_trace_rows, ) except sqlite3.OperationalError: pass conn.commit() SQLiteStorage._move_media_dir( project_media_dir(project) / old_name, project_media_dir(project) / new_name, ) except sqlite3.Error as e: raise RuntimeError( f"Database error while renaming run '{old_name}' to '{new_name}': {e}" ) from e @staticmethod def move_run( project: str, run: str, new_project: str, run_id: str | None = None ) -> bool: """Move a run from one project to another. When the source DB supports run_ids, ``run_id`` uniquely identifies the run being moved; only that run is touched even when other runs share the same ``run`` name. """ source_db_path = SQLiteStorage.get_project_db_path(project) if not source_db_path.exists(): return False target_db_path = SQLiteStorage.init_db(new_project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_process_lock(new_project): with SQLiteStorage._get_connection(source_db_path) as source_conn: source_cursor = source_conn.cursor() metrics_has_run_id = SQLiteStorage._supports_run_ids( source_conn, "metrics" ) configs_has_run_id = SQLiteStorage._supports_run_ids( source_conn, "configs" ) system_has_run_id = SQLiteStorage._supports_run_ids( source_conn, "system_metrics" ) alerts_has_run_id = SQLiteStorage._supports_run_ids( source_conn, "alerts" ) metrics_identity = SQLiteStorage._resolve_run_identity( source_conn, run_name=run, run_id=run_id, table="metrics" ) if metrics_identity is None: return False metrics_col, metrics_val = metrics_identity configs_identity = SQLiteStorage._resolve_run_identity( source_conn, run_name=run, run_id=run_id, table="configs" ) system_identity = SQLiteStorage._resolve_run_identity( source_conn, run_name=run, run_id=run_id, table="system_metrics", ) alerts_identity = SQLiteStorage._resolve_run_identity( source_conn, run_name=run, run_id=run_id, table="alerts" ) traces_identity = SQLiteStorage._resolve_run_identity( source_conn, run_name=run, run_id=run_id, table="traces" ) metrics_select = ( "SELECT timestamp, step, metrics" + (", run_id" if metrics_has_run_id else "") + f" FROM metrics WHERE {metrics_col} = ?" ) source_cursor.execute(metrics_select, (metrics_val,)) metrics_rows = source_cursor.fetchall() config_row = None if configs_identity is not None: configs_col, configs_val = configs_identity configs_select = ( "SELECT config, created_at" + (", run_id" if configs_has_run_id else "") + f" FROM configs WHERE {configs_col} = ?" ) source_cursor.execute(configs_select, (configs_val,)) config_row = source_cursor.fetchone() system_metrics_rows = [] if system_identity is not None: try: system_col, system_val = system_identity system_select = ( "SELECT timestamp, metrics" + (", run_id" if system_has_run_id else "") + f" FROM system_metrics WHERE {system_col} = ?" ) source_cursor.execute(system_select, (system_val,)) system_metrics_rows = source_cursor.fetchall() except sqlite3.OperationalError: system_metrics_rows = [] alert_rows = [] if alerts_identity is not None: try: alerts_col, alerts_val = alerts_identity alerts_select = ( "SELECT timestamp, title, text, level, step, alert_id" + (", run_id" if alerts_has_run_id else "") + f" FROM alerts WHERE {alerts_col} = ?" ) source_cursor.execute(alerts_select, (alerts_val,)) alert_rows = source_cursor.fetchall() except sqlite3.OperationalError: alert_rows = [] trace_rows = [] if traces_identity is not None: try: traces_col, traces_val = traces_identity source_cursor.execute( f""" SELECT id, run_id, timestamp, step, key, trace_index, messages, metadata, search_text, log_id, space_id FROM traces WHERE {traces_col} = ? """, (traces_val,), ) trace_rows = source_cursor.fetchall() except sqlite3.OperationalError: trace_rows = [] if ( not metrics_rows and not config_row and not system_metrics_rows and not trace_rows ): return False with SQLiteStorage._get_connection(target_db_path) as target_conn: target_cursor = target_conn.cursor() old_prefix = f"{project}/{run}/" new_prefix = f"{new_project}/{run}/" target_metrics_run_id = SQLiteStorage._supports_run_ids( target_conn, "metrics" ) target_configs_run_id = SQLiteStorage._supports_run_ids( target_conn, "configs" ) target_system_run_id = SQLiteStorage._supports_run_ids( target_conn, "system_metrics" ) target_alerts_run_id = SQLiteStorage._supports_run_ids( target_conn, "alerts" ) target_traces_run_id = SQLiteStorage._supports_run_ids( target_conn, "traces" ) needs_generated_run_id = ( target_metrics_run_id or target_configs_run_id or target_system_run_id or target_alerts_run_id or target_traces_run_id ) and not ( metrics_has_run_id or configs_has_run_id or system_has_run_id or alerts_has_run_id ) generated_run_id = ( uuid.uuid4().hex if needs_generated_run_id else None ) use_metrics_run_id = ( metrics_has_run_id and target_metrics_run_id ) updated_rows = SQLiteStorage._rewrite_metrics_rows( metrics_rows, run, old_prefix, new_prefix, include_run_id=use_metrics_run_id, ) if use_metrics_run_id: target_cursor.executemany( "INSERT INTO metrics (timestamp, run_name, step, metrics, run_id) VALUES (?, ?, ?, ?, ?)", updated_rows, ) elif target_metrics_run_id and generated_run_id is not None: target_cursor.executemany( "INSERT INTO metrics (timestamp, run_name, step, metrics, run_id) VALUES (?, ?, ?, ?, ?)", [row + (generated_run_id,) for row in updated_rows], ) else: target_cursor.executemany( "INSERT INTO metrics (timestamp, run_name, step, metrics) VALUES (?, ?, ?, ?)", updated_rows, ) if config_row: if ( configs_has_run_id and target_configs_run_id and "run_id" in config_row.keys() ): target_cursor.execute( """ INSERT OR REPLACE INTO configs (run_name, config, created_at, run_id) VALUES (?, ?, ?, ?) """, ( run, config_row["config"], config_row["created_at"], config_row["run_id"], ), ) elif target_configs_run_id and generated_run_id is not None: target_cursor.execute( """ INSERT OR REPLACE INTO configs (run_name, config, created_at, run_id) VALUES (?, ?, ?, ?) """, ( run, config_row["config"], config_row["created_at"], generated_run_id, ), ) else: target_cursor.execute( """ INSERT OR REPLACE INTO configs (run_name, config, created_at) VALUES (?, ?, ?) """, ( run, config_row["config"], config_row["created_at"], ), ) for row in system_metrics_rows: try: if ( system_has_run_id and target_system_run_id and "run_id" in row.keys() ): target_cursor.execute( """ INSERT INTO system_metrics (timestamp, run_name, metrics, run_id) VALUES (?, ?, ?, ?) """, ( row["timestamp"], run, row["metrics"], row["run_id"], ), ) elif ( target_system_run_id and generated_run_id is not None ): target_cursor.execute( """ INSERT INTO system_metrics (timestamp, run_name, metrics, run_id) VALUES (?, ?, ?, ?) """, ( row["timestamp"], run, row["metrics"], generated_run_id, ), ) else: target_cursor.execute( """ INSERT INTO system_metrics (timestamp, run_name, metrics) VALUES (?, ?, ?) """, (row["timestamp"], run, row["metrics"]), ) except sqlite3.OperationalError: pass for row in alert_rows: try: if ( alerts_has_run_id and target_alerts_run_id and "run_id" in row.keys() ): target_cursor.execute( """ INSERT OR IGNORE INTO alerts (timestamp, run_name, title, text, level, step, alert_id, run_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( row["timestamp"], run, row["title"], row["text"], row["level"], row["step"], row["alert_id"], row["run_id"], ), ) elif ( target_alerts_run_id and generated_run_id is not None ): target_cursor.execute( """ INSERT OR IGNORE INTO alerts (timestamp, run_name, title, text, level, step, alert_id, run_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( row["timestamp"], run, row["title"], row["text"], row["level"], row["step"], row["alert_id"], generated_run_id, ), ) else: target_cursor.execute( """ INSERT OR IGNORE INTO alerts (timestamp, run_name, title, text, level, step, alert_id) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( row["timestamp"], run, row["title"], row["text"], row["level"], row["step"], row["alert_id"], ), ) except sqlite3.OperationalError: pass if trace_rows: trace_run_id = None if target_traces_run_id: if generated_run_id is not None: trace_run_id = generated_run_id elif traces_identity is not None and trace_rows: trace_run_id = trace_rows[0]["run_id"] updated_trace_rows = SQLiteStorage._rewrite_trace_rows( trace_rows, run, old_prefix, new_prefix, run_id=trace_run_id, ) try: target_cursor.executemany( """ INSERT OR IGNORE INTO traces (id, run_id, timestamp, run_name, step, key, trace_index, messages, metadata, search_text, log_id, space_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, updated_trace_rows, ) except sqlite3.OperationalError: pass target_conn.commit() SQLiteStorage._move_media_dir( project_media_dir(project) / run, project_media_dir(new_project) / run, ) source_cursor.execute( f"DELETE FROM metrics WHERE {metrics_col} = ?", (metrics_val,), ) if configs_identity is not None: configs_col, configs_val = configs_identity source_cursor.execute( f"DELETE FROM configs WHERE {configs_col} = ?", (configs_val,), ) if system_identity is not None: try: system_col, system_val = system_identity source_cursor.execute( f"DELETE FROM system_metrics WHERE {system_col} = ?", (system_val,), ) except sqlite3.OperationalError: pass if alerts_identity is not None: try: alerts_col, alerts_val = alerts_identity source_cursor.execute( f"DELETE FROM alerts WHERE {alerts_col} = ?", (alerts_val,), ) except sqlite3.OperationalError: pass if traces_identity is not None: try: traces_col, traces_val = traces_identity source_cursor.execute( f"DELETE FROM traces WHERE {traces_col} = ?", (traces_val,), ) except sqlite3.OperationalError: pass source_conn.commit() return True @staticmethod def get_all_run_configs(project: str) -> dict[str, dict]: """Get configurations for all runs in a project.""" db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return {} with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: config_col = ( "run_id" if SQLiteStorage._supports_run_ids(conn, "configs") else "run_name" ) cursor.execute(f"SELECT {config_col}, config FROM configs") results = {} for row in cursor.fetchall(): config = orjson.loads(row["config"]) results[row[config_col]] = deserialize_values(config) return results except sqlite3.OperationalError as e: if "no such table: configs" in str(e): return {} raise @staticmethod def get_metric_values( project: str, run: str | None, metric_name: str, step: int | None = None, around_step: int | None = None, at_time: str | None = None, window: int | float | None = None, run_id: str | None = None, ) -> list[dict]: """Get values for a specific metric in a project/run with optional filtering. Filtering modes: - step: return the single row at exactly this step - around_step + window: return rows where step is in [around_step - window, around_step + window] - at_time + window: return rows within ±window seconds of the ISO timestamp - No filters: return all rows """ db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="metrics" ) if run_identity is None: return [] query = f"SELECT timestamp, step, metrics FROM metrics WHERE {run_identity[0]} = ?" params: list = [run_identity[1]] if step is not None: query += " AND step = ?" params.append(step) elif around_step is not None and window is not None: query += " AND step >= ? AND step <= ?" params.extend([around_step - int(window), around_step + int(window)]) elif at_time is not None and window is not None: query += ( " AND timestamp >= datetime(?, '-' || ? || ' seconds')" " AND timestamp <= datetime(?, '+' || ? || ' seconds')" ) params.extend([at_time, int(window), at_time, int(window)]) query += " ORDER BY timestamp" cursor.execute(query, params) rows = cursor.fetchall() results = [] for row in rows: metrics = orjson.loads(row["metrics"]) metrics = deserialize_values(metrics) if metric_name in metrics: results.append( { "timestamp": row["timestamp"], "step": row["step"], "value": metrics[metric_name], } ) return results @staticmethod def get_snapshot( project: str, run: str | None = None, step: int | None = None, around_step: int | None = None, at_time: str | None = None, window: int | float | None = None, run_id: str | None = None, ) -> dict[str, list[dict]]: """Get all metrics at/around a point in time or step. Returns a dict mapping metric names to lists of {timestamp, step, value}. """ db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return {} with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table="metrics" ) if run_identity is None: return {} query = f"SELECT timestamp, step, metrics FROM metrics WHERE {run_identity[0]} = ?" params: list = [run_identity[1]] if step is not None: query += " AND step = ?" params.append(step) elif around_step is not None and window is not None: query += " AND step >= ? AND step <= ?" params.extend([around_step - int(window), around_step + int(window)]) elif at_time is not None and window is not None: query += ( " AND timestamp >= datetime(?, '-' || ? || ' seconds')" " AND timestamp <= datetime(?, '+' || ? || ' seconds')" ) params.extend([at_time, int(window), at_time, int(window)]) query += " ORDER BY timestamp" cursor.execute(query, params) result: dict[str, list[dict]] = {} for row in cursor.fetchall(): metrics = orjson.loads(row["metrics"]) metrics = deserialize_values(metrics) for key, value in metrics.items(): if key not in result: result[key] = [] result[key].append( { "timestamp": row["timestamp"], "step": row["step"], "value": value, } ) return result @staticmethod def get_all_metrics_for_run( project: str, run: str | None = None, run_id: str | None = None ) -> list[str]: """Get all metric names for a specific project/run.""" return SQLiteStorage._get_metric_names( project, run, "metrics", exclude_keys={"timestamp", "step"}, run_id=run_id, ) @staticmethod def _get_metric_names( project: str, run: str | None, table: str, exclude_keys: set[str], run_id: str | None = None, ) -> list[str]: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: run_identity = SQLiteStorage._resolve_run_identity( conn, run_name=run, run_id=run_id, table=table ) if run_identity is None: return [] cursor.execute( f""" SELECT metrics FROM {table} WHERE {run_identity[0]} = ? ORDER BY timestamp """, (run_identity[1],), ) rows = cursor.fetchall() all_metrics = set() for row in rows: metrics = orjson.loads(row["metrics"]) metrics = deserialize_values(metrics) for key in metrics.keys(): if key not in exclude_keys: all_metrics.add(key) return sorted(list(all_metrics)) except sqlite3.OperationalError as e: if f"no such table: {table}" in str(e): return [] raise @staticmethod def set_project_metadata(project: str, key: str, value: str) -> None: db_path = SQLiteStorage.init_db(project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: conn.execute( "INSERT OR REPLACE INTO project_metadata (key, value) VALUES (?, ?)", (key, value), ) conn.commit() @staticmethod def get_project_metadata(project: str, key: str) -> str | None: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: cursor.execute( "SELECT value FROM project_metadata WHERE key = ?", (key,) ) row = cursor.fetchone() return row[0] if row else None except sqlite3.OperationalError: return None @staticmethod def get_space_id(project: str) -> str | None: return SQLiteStorage.get_project_metadata(project, "space_id") @staticmethod def has_pending_data(project: str) -> bool: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return False with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: cursor.execute( "SELECT EXISTS(SELECT 1 FROM metrics WHERE space_id IS NOT NULL LIMIT 1)" ) if cursor.fetchone()[0]: return True except sqlite3.OperationalError: pass try: cursor.execute( "SELECT EXISTS(SELECT 1 FROM system_metrics WHERE space_id IS NOT NULL LIMIT 1)" ) if cursor.fetchone()[0]: return True except sqlite3.OperationalError: pass try: cursor.execute("SELECT EXISTS(SELECT 1 FROM pending_uploads LIMIT 1)") if cursor.fetchone()[0]: return True except sqlite3.OperationalError: pass return False @staticmethod def get_pending_logs(project: str) -> dict | None: return SQLiteStorage._get_pending( project, "metrics", extra_fields=["step"], include_config=True ) @staticmethod def clear_pending_logs(project: str, metric_ids: list[int]) -> None: SQLiteStorage._clear_pending(project, "metrics", metric_ids) @staticmethod def get_pending_system_logs(project: str) -> dict | None: return SQLiteStorage._get_pending(project, "system_metrics") @staticmethod def _get_pending( project: str, table: str, extra_fields: list[str] | None = None, include_config: bool = False, ) -> dict | None: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None extra_cols = ", ".join(extra_fields) + ", " if extra_fields else "" with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: run_id_col = ( "run_id, " if SQLiteStorage._supports_run_ids(conn, table) else "" ) cursor.execute( f"""SELECT id, timestamp, {run_id_col}run_name, {extra_cols}metrics, log_id, space_id FROM {table} WHERE space_id IS NOT NULL""" ) except sqlite3.OperationalError: return None rows = cursor.fetchall() if not rows: return None logs = [] ids = [] for row in rows: metrics = deserialize_values(orjson.loads(row["metrics"])) entry = { "project": project, "run": row["run_name"], "run_id": row["run_name"], "metrics": metrics, "timestamp": row["timestamp"], "log_id": row["log_id"], } if "run_id" in row.keys(): entry["run_id"] = row["run_id"] for field in extra_fields or []: entry[field] = row[field] if include_config: entry["config"] = None logs.append(entry) ids.append(row["id"]) return {"logs": logs, "ids": ids, "space_id": rows[0]["space_id"]} @staticmethod def clear_pending_system_logs(project: str, metric_ids: list[int]) -> None: SQLiteStorage._clear_pending(project, "system_metrics", metric_ids) @staticmethod def _clear_pending(project: str, table: str, ids: list[int]) -> None: if not ids: return db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: placeholders = ",".join("?" * len(ids)) conn.execute( f"UPDATE {table} SET space_id = NULL WHERE id IN ({placeholders})", ids, ) conn.commit() @staticmethod def get_pending_uploads(project: str) -> dict | None: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: columns = SQLiteStorage._table_columns(conn, "pending_uploads") select_cols = [ "id", "space_id", "run_name", "step", "file_path", "relative_path", ] if "run_id" in columns: select_cols.insert(2, "run_id") if "kind" in columns: select_cols.append("kind") if "digest" in columns: select_cols.append("digest") cursor.execute(f"SELECT {', '.join(select_cols)} FROM pending_uploads") except sqlite3.OperationalError: return None rows = cursor.fetchall() if not rows: return None uploads = [] ids = [] for row in rows: keys = row.keys() uploads.append( { "project": project, "run": row["run_name"], "run_id": ( row["run_id"] if "run_id" in keys else row["run_name"] ) or row["run_name"], "step": row["step"], "file_path": row["file_path"], "relative_path": row["relative_path"], "kind": row["kind"] if "kind" in keys else "media", "digest": row["digest"] if "digest" in keys else None, } ) ids.append(row["id"]) return {"uploads": uploads, "ids": ids, "space_id": rows[0]["space_id"]} @staticmethod def clear_pending_uploads(project: str, upload_ids: list[int]) -> None: if not upload_ids: return db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: placeholders = ",".join("?" * len(upload_ids)) conn.execute( f"DELETE FROM pending_uploads WHERE id IN ({placeholders})", upload_ids, ) conn.commit() @staticmethod def add_pending_upload( project: str, space_id: str, run_id: str | None, run_name: str | None, step: int | None, file_path: str, relative_path: str | None, *, kind: str = "media", digest: Sha256Digest | None = None, ) -> None: db_path = SQLiteStorage.init_db(project) now = datetime.now(timezone.utc).isoformat() with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: cols, vals = SQLiteStorage._pending_upload_cols_vals( include_run_id=SQLiteStorage._supports_run_ids( conn, "pending_uploads" ), space_id=space_id, run_id=run_id, run_name=run_name, step=step, file_path=file_path, relative_path=relative_path, kind=kind, digest=digest, created_at=now, ) placeholders = ", ".join("?" for _ in cols) conn.execute( f"INSERT INTO pending_uploads ({', '.join(cols)}) VALUES ({placeholders})", vals, ) conn.commit() @staticmethod def _pending_upload_cols_vals( *, include_run_id: bool, space_id: str, run_id: str | None, run_name: str | None, step: int | None, file_path: str, relative_path: str | None, kind: str, digest: Sha256Digest | None, created_at: str, ) -> tuple[list[str], list]: """Build the (columns, values) for one pending_uploads row. `run_id` is detected at runtime rather than backfilled, so it is dropped on legacy DBs whose pending_uploads table predates the column.""" fields = [ ("space_id", space_id, True), ("run_id", run_id, include_run_id), ("run_name", run_name, True), ("step", step, True), ("file_path", file_path, True), ("relative_path", relative_path, True), ("kind", kind, True), ("digest", digest, True), ("created_at", created_at, True), ] cols = [col for col, _, include in fields if include] vals = [val for _, val, include in fields if include] return cols, vals @staticmethod def _canonical_manifest( manifest: Manifest, ) -> tuple[Manifest, Sha256Digest, int]: canonical: Manifest = [] size_bytes = 0 for entry in manifest: path = entry["path"] digest = Sha256Digest(entry["digest"]) size = int(entry["size"]) canonical.append({"path": path, "digest": digest, "size": size}) size_bytes += size canonical.sort(key=lambda e: e["path"]) payload = orjson.dumps(canonical, option=orjson.OPT_SORT_KEYS) manifest_digest = Sha256Digest(hashlib.sha256(payload).hexdigest()) return canonical, manifest_digest, size_bytes @staticmethod def _create_or_get_artifact_cursor( conn: sqlite3.Connection, name: str, type: str, description: str | None, now: str, ) -> int: """Return the id of the artifact named `name`, creating it if absent. For an existing artifact a non-None `description` is applied in place (the type is immutable and a mismatch raises).""" cursor = conn.cursor() row = cursor.execute( "SELECT id, type FROM artifacts WHERE name = ?", (name,) ).fetchone() if row is not None: if row["type"] != type: raise ValueError( f"Artifact '{name}' already exists with type " f"'{row['type']}'; cannot relog with type '{type}'." ) if description is not None: cursor.execute( "UPDATE artifacts SET description = ? WHERE id = ?", (description, int(row["id"])), ) return int(row["id"]) cursor.execute( """INSERT INTO artifacts (name, type, description, created_at) VALUES (?, ?, ?, ?)""", (name, type, description, now), ) return int(cursor.lastrowid) @staticmethod def create_or_get_artifact( project: str, name: str, type: str, description: str | None, ) -> int: db_path = SQLiteStorage.init_db(project) now = datetime.now(timezone.utc).isoformat() with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: artifact_id = SQLiteStorage._create_or_get_artifact_cursor( conn, name, type, description, now ) conn.commit() return artifact_id @staticmethod def _insert_artifact_version_cursor( conn: sqlite3.Connection, artifact_id: int, manifest: Manifest, metadata: dict | None, producer_run_id: str | None, producer_run_name: str | None, now: str, ) -> tuple[int, int, bool]: canonical, manifest_digest, size_bytes = SQLiteStorage._canonical_manifest( manifest ) manifest_json = orjson.dumps(canonical).decode("utf-8") metadata_json = orjson.dumps(metadata).decode("utf-8") if metadata else None cursor = conn.cursor() existing = cursor.execute( """SELECT id, version FROM artifact_versions WHERE artifact_id = ? AND manifest_digest = ?""", (artifact_id, manifest_digest), ).fetchone() if existing is not None: if metadata: cursor.execute( "UPDATE artifact_versions SET metadata = ? WHERE id = ?", (metadata_json, int(existing["id"])), ) return int(existing["id"]), int(existing["version"]), False row = cursor.execute( "SELECT MAX(version) AS m FROM artifact_versions WHERE artifact_id = ?", (artifact_id,), ).fetchone() next_version = 0 if row["m"] is None else int(row["m"]) + 1 cursor.execute( """INSERT INTO artifact_versions (artifact_id, version, manifest_digest, manifest, metadata, size_bytes, producer_run_id, producer_run_name, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( artifact_id, next_version, manifest_digest, manifest_json, metadata_json, size_bytes, producer_run_id, producer_run_name, now, ), ) return int(cursor.lastrowid), next_version, True @staticmethod def insert_artifact_version( project: str, artifact_id: int, manifest: Manifest, metadata: dict | None, producer_run_id: str | None, producer_run_name: str | None, ) -> tuple[int, int, bool]: """Returns `(version_id, version, created)`. `created` is False when an identical-content version already existed; its `metadata` is refreshed in place from the new call (when non-empty) before it is returned.""" db_path = SQLiteStorage.init_db(project) now = datetime.now(timezone.utc).isoformat() with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: result = SQLiteStorage._insert_artifact_version_cursor( conn, artifact_id, manifest, metadata, producer_run_id, producer_run_name, now, ) conn.commit() return result @staticmethod def _reassign_alias_cursor( conn: sqlite3.Connection, artifact_id: int, alias: str, version_id: int, ) -> None: if cas.ARTIFACT_VERSION_SPEC_RE.match(alias): raise ValueError( f"Alias '{alias}' is reserved for version pointers (vN); choose another." ) conn.execute( """INSERT INTO artifact_aliases (artifact_id, alias, artifact_version_id) VALUES (?, ?, ?) ON CONFLICT(artifact_id, alias) DO UPDATE SET artifact_version_id = excluded.artifact_version_id""", (artifact_id, alias, version_id), ) @staticmethod def _reassign_alias_forward_cursor( conn: sqlite3.Connection, artifact_id: int, alias: str, version_id: int, version_int: int, ) -> None: """Reassign `alias` only when it does not move backward.""" current = conn.execute( """SELECT av.version FROM artifact_aliases aa JOIN artifact_versions av ON av.id = aa.artifact_version_id WHERE aa.artifact_id = ? AND aa.alias = ?""", (artifact_id, alias), ).fetchone() if current is not None and int(current["version"]) > version_int: return SQLiteStorage._reassign_alias_cursor(conn, artifact_id, alias, version_id) @staticmethod def reassign_alias( project: str, artifact_id: int, alias: str, version_id: int, ) -> None: db_path = SQLiteStorage.init_db(project) with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: SQLiteStorage._reassign_alias_cursor( conn, artifact_id, alias, version_id ) conn.commit() @staticmethod def _resolve_artifact_version_cursor( conn: sqlite3.Connection, name: str, spec: str | None, ) -> dict | None: cursor = conn.cursor() art = cursor.execute( "SELECT id FROM artifacts WHERE name = ?", (name,) ).fetchone() if art is None: return None artifact_id = int(art["id"]) spec = spec if spec else "latest" m = cas.ARTIFACT_VERSION_SPEC_RE.match(spec) if m: version_int = int(m.group(1)) ver = cursor.execute( """SELECT id, version FROM artifact_versions WHERE artifact_id = ? AND version = ?""", (artifact_id, version_int), ).fetchone() else: ver = cursor.execute( """SELECT av.id, av.version FROM artifact_versions av JOIN artifact_aliases aa ON aa.artifact_version_id = av.id WHERE aa.artifact_id = ? AND aa.alias = ?""", (artifact_id, spec), ).fetchone() if ver is None: return None return { "artifact_id": artifact_id, "version_id": int(ver["id"]), "version": int(ver["version"]), } @staticmethod def resolve_artifact_version( project: str, name: str, spec: str | None, ) -> dict | None: SQLiteStorage._ensure_hub_loaded() db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None with SQLiteStorage._get_connection(db_path) as conn: return SQLiteStorage._resolve_artifact_version_cursor(conn, name, spec) @staticmethod def _insert_run_artifact_link_cursor( conn: sqlite3.Connection, run_name: str | None, run_id: str | None, version_id: int, direction: str, now: str, ) -> None: if direction not in ("input", "output"): raise ValueError( f"direction must be 'input' or 'output', got {direction!r}" ) conn.execute( """INSERT OR IGNORE INTO run_artifact_links (run_id, run_name, artifact_version_id, direction, created_at) VALUES (?, ?, ?, ?, ?)""", (run_id, run_name, version_id, direction, now), ) @staticmethod def insert_run_artifact_link( project: str, run_name: str | None, run_id: str | None, version_id: int, direction: str, ) -> None: db_path = SQLiteStorage.init_db(project) now = datetime.now(timezone.utc).isoformat() with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: SQLiteStorage._insert_run_artifact_link_cursor( conn, run_name, run_id, version_id, direction, now ) conn.commit() @staticmethod def commit_artifact_version( project: str, name: str, type: str, description: str | None, manifest: Manifest, metadata: dict | None, aliases: list[str] | None, run_name: str | None, run_id: str | None, ) -> dict: """Commit a new artifact version and return its full manifest record. `latest` advances only when a new version is created, so re-logging identical or older content never regresses `latest`. Moving aliases are reassigned the same way: a content-dedup hit to an older version never drags an existing alias backward, but a first-time or forward tag lands. """ db_path = SQLiteStorage.init_db(project) now = datetime.now(timezone.utc).isoformat() with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: artifact_id = SQLiteStorage._create_or_get_artifact_cursor( conn, name, type, description, now ) version_id, version_int, created = ( SQLiteStorage._insert_artifact_version_cursor( conn, artifact_id, manifest, metadata, run_id, run_name, now ) ) if created: SQLiteStorage._reassign_alias_cursor( conn, artifact_id, "latest", version_id ) for alias in aliases or []: SQLiteStorage._reassign_alias_forward_cursor( conn, artifact_id, alias, version_id, version_int ) SQLiteStorage._insert_run_artifact_link_cursor( conn, run_name, run_id, version_id, "output", now ) conn.commit() return SQLiteStorage._get_artifact_manifest_cursor( conn, name, f"v{version_int}" ) @staticmethod def _get_artifact_manifest_cursor( conn: sqlite3.Connection, name: str, spec: str | None, ) -> dict | None: resolved = SQLiteStorage._resolve_artifact_version_cursor(conn, name, spec) if resolved is None: return None cursor = conn.cursor() row = cursor.execute( """SELECT av.id, av.version, av.manifest, av.manifest_digest, av.metadata, av.size_bytes, av.producer_run_id, av.producer_run_name, av.created_at, a.name, a.type, a.description FROM artifact_versions av JOIN artifacts a ON a.id = av.artifact_id WHERE av.id = ?""", (resolved["version_id"],), ).fetchone() if row is None: return None alias_rows = cursor.execute( """SELECT alias FROM artifact_aliases WHERE artifact_version_id = ?""", (resolved["version_id"],), ).fetchall() return { "artifact_id": resolved["artifact_id"], "version_id": int(row["id"]), "version": int(row["version"]), "name": row["name"], "type": row["type"], "description": row["description"], "manifest": orjson.loads(row["manifest"]), "manifest_digest": row["manifest_digest"], "metadata": ( orjson.loads(row["metadata"]) if row["metadata"] is not None else None ), "size_bytes": int(row["size_bytes"]), "producer_run_id": row["producer_run_id"], "producer_run_name": row["producer_run_name"], "created_at": row["created_at"], "aliases": [r["alias"] for r in alias_rows], } @staticmethod def get_artifact_manifest( project: str, name: str, spec: str | None, ) -> dict | None: SQLiteStorage._ensure_hub_loaded() db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return None with SQLiteStorage._get_connection(db_path) as conn: return SQLiteStorage._get_artifact_manifest_cursor(conn, name, spec) @staticmethod def get_run_artifacts( project: str, run_name: str | None, run_id: str | None, ) -> dict[str, list[dict]]: SQLiteStorage._ensure_hub_loaded() empty = {"input": [], "output": []} db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return empty with SQLiteStorage._get_connection(db_path) as conn: identity = SQLiteStorage._resolve_run_identity( conn, run_name=run_name, run_id=run_id, table="run_artifact_links" ) if identity is None: col, val = ( ("run_id", run_id) if run_id is not None else ("run_name", run_name) ) else: col, val = identity if val is None: return empty cursor = conn.cursor() rows = cursor.execute( f"""SELECT ral.direction, ral.created_at, av.id AS version_id, av.version, av.size_bytes, a.name, a.type FROM run_artifact_links ral JOIN artifact_versions av ON av.id = ral.artifact_version_id JOIN artifacts a ON a.id = av.artifact_id WHERE ral.{col} = ? ORDER BY ral.created_at""", (val,), ).fetchall() result: dict[str, list[dict]] = {"input": [], "output": []} for row in rows: result[row["direction"]].append( { "version_id": int(row["version_id"]), "name": row["name"], "type": row["type"], "version": int(row["version"]), "size_bytes": int(row["size_bytes"]), "created_at": row["created_at"], } ) return result @staticmethod def enqueue_artifact_blob_uploads( project: str, space_id: str, blobs: list[tuple[Sha256Digest, str]], run_name: str | None, run_id: str | None, ) -> None: """Enqueue one pending_uploads row per `(digest, local_blob_path)` in a single connection/transaction.""" if not blobs: return db_path = SQLiteStorage.init_db(project) now = datetime.now(timezone.utc).isoformat() with SQLiteStorage._get_process_lock(project): with SQLiteStorage._get_connection(db_path) as conn: include_run_id = SQLiteStorage._supports_run_ids( conn, "pending_uploads" ) rows = [ SQLiteStorage._pending_upload_cols_vals( include_run_id=include_run_id, space_id=space_id, run_id=run_id, run_name=run_name, step=None, file_path=file_path, relative_path=None, kind="artifact_blob", digest=digest, created_at=now, ) for digest, file_path in blobs ] cols = rows[0][0] placeholders = ", ".join("?" for _ in cols) conn.executemany( f"INSERT INTO pending_uploads ({', '.join(cols)}) VALUES ({placeholders})", [vals for _, vals in rows], ) conn.commit() @staticmethod def list_artifact_blobs_present( project: str, digests: list[Sha256Digest], ) -> set[Sha256Digest]: SQLiteStorage._ensure_hub_loaded() present: set[Sha256Digest] = set() for digest in digests: if not cas.SHA256_DIGEST_RE.match(digest): continue if cas.blob_path(project, digest).is_file(): present.add(digest) return present @staticmethod def get_all_logs_for_sync(project: str) -> list[dict]: return SQLiteStorage._get_all_for_sync( project, "metrics", order_by="run_name, step", extra_fields=["step"], include_config=True, ) @staticmethod def get_all_system_logs_for_sync(project: str) -> list[dict]: return SQLiteStorage._get_all_for_sync( project, "system_metrics", order_by="run_name, timestamp" ) @staticmethod def _get_all_for_sync( project: str, table: str, order_by: str, extra_fields: list[str] | None = None, include_config: bool = False, ) -> list[dict]: db_path = SQLiteStorage.get_project_db_path(project) if not db_path.exists(): return [] extra_cols = ", ".join(extra_fields) + ", " if extra_fields else "" with SQLiteStorage._get_connection(db_path) as conn: cursor = conn.cursor() try: run_id_col = ( "run_id, " if SQLiteStorage._supports_run_ids(conn, table) else "" ) cursor.execute( f"""SELECT timestamp, {run_id_col}run_name, {extra_cols}metrics, log_id FROM {table} ORDER BY {order_by}""" ) except sqlite3.OperationalError: return [] rows = cursor.fetchall() results = [] for row in rows: metrics = deserialize_values(orjson.loads(row["metrics"])) entry = { "project": project, "run": row["run_name"], "run_id": row["run_name"], "metrics": metrics, "timestamp": row["timestamp"], "log_id": row["log_id"], } if "run_id" in row.keys(): entry["run_id"] = row["run_id"] for field in extra_fields or []: entry[field] = row[field] if include_config: entry["config"] = None results.append(entry) return results