From 26b3e0f721b9bc85bb2a4cbc070585a4252ebc6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Patryk=20W=C3=B3jtowicz?= Date: Wed, 11 Nov 2026 23:45:49 +0100 Subject: [PATCH] feat(adapters): add the SQLite metadata store --- CHANGELOG.md | 1 + src/miniops/adapters/__init__.py | 0 src/miniops/adapters/local/__init__.py | 0 src/miniops/adapters/local/metadata.py | 362 +++++++++++++++++++++ src/miniops/adapters/local/schema.py | 110 +++++++ src/miniops/core/errors.py | 4 + src/miniops/ports/metadata.py | 34 +- tests/adapters/__init__.py | 0 tests/adapters/local/__init__.py | 0 tests/adapters/local/test_metadata.py | 415 +++++++++++++++++++++++++ tests/adapters/local/test_schema.py | 93 ++++++ 11 files changed, 1015 insertions(+), 4 deletions(-) create mode 100644 src/miniops/adapters/__init__.py create mode 100644 src/miniops/adapters/local/__init__.py create mode 100644 src/miniops/adapters/local/metadata.py create mode 100644 src/miniops/adapters/local/schema.py create mode 100644 tests/adapters/__init__.py create mode 100644 tests/adapters/local/__init__.py create mode 100644 tests/adapters/local/test_metadata.py create mode 100644 tests/adapters/local/test_schema.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c3f44f1..3645b0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - Core domain model: models, versions with semantic version numbers, runs and aliases, with the rules for releasing, renumbering and deleting a version. - Interfaces (ports) for the metadata store and the artifact store, and provisional ones for the executor, the serving backend and the image registry. +- Local metadata store in SQLite, the first implementation of the metadata port. Its schema is versioned and upgrades itself when a store is opened, after making a backup. ## [0.1.0a1] - 2026-11-11 diff --git a/src/miniops/adapters/__init__.py b/src/miniops/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/miniops/adapters/local/__init__.py b/src/miniops/adapters/local/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/miniops/adapters/local/metadata.py b/src/miniops/adapters/local/metadata.py new file mode 100644 index 0000000..3138827 --- /dev/null +++ b/src/miniops/adapters/local/metadata.py @@ -0,0 +1,362 @@ +import json +import sqlite3 +from collections.abc import Callable, Generator, Mapping, Sequence +from contextlib import contextmanager +from dataclasses import asdict +from datetime import UTC, datetime +from pathlib import Path +from typing import Self, override +from uuid import UUID + +from miniops.adapters.local.schema import migrate +from miniops.core.entities import ( + Alias, + ArtifactRef, + Column, + DatasetRef, + Model, + ModelVersion, + Run, + RunStatus, + Signature, + VersionRef, + VersionStatus, +) +from miniops.core.errors import AlreadyExistsError, InvalidStateError, NotFoundError +from miniops.core.versions import SemVer +from miniops.ports.metadata import MetadataStore + + +class SqliteMetadataStore(MetadataStore): + """The metadata store of the ``local`` and ``docker`` targets: one SQLite file.""" + + def __init__(self, path: Path) -> None: + """Open the store at ``path``. A missing store is created and an older one is upgraded. + + Raises: + IncompatibleStoreError: If the store was written by a newer miniOPs. + """ + path.parent.mkdir(parents=True, exist_ok=True) + self._connection = sqlite3.connect(path, autocommit=True) + self._connection.row_factory = sqlite3.Row + try: + self._connection.execute("PRAGMA foreign_keys = ON") + migrate(self._connection, path) + except BaseException: + self._connection.close() + raise + + def close(self) -> None: + """Close the database file.""" + self._connection.close() + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_: object) -> None: + self.close() + + @override + def add_model(self, model: Model) -> None: + with _constraints(taken=f"Model {model.name!r} is already in the store."): + self._connection.execute(*_insert("models", _model_row(model))) + + @override + def get_model(self, name: str) -> Model | None: + row = self._connection.execute("SELECT * FROM models WHERE name = ?", (name,)).fetchone() + return _maybe(row, _model) + + @override + def list_models(self) -> Sequence[Model]: + rows = self._connection.execute("SELECT * FROM models ORDER BY name") + return [_model(row) for row in rows] + + @override + def add_version(self, version: ModelVersion) -> None: + if version.status is not VersionStatus.DRAFT: + raise InvalidStateError( + f"Version {version.number} is {version.status}. A version enters the store " + "as a draft and is released there." + ) + with _constraints( + taken=f"The model already has a version {version.number}.", + missing=f"The store has no model with id {version.model_id}.", + ): + self._connection.execute(*_insert("versions", _version_row(version))) + + @override + def update_version(self, version: ModelVersion) -> None: + with _constraints( + taken=f"The model already has a version {version.number}.", + missing=f"Run {version.released_run_id} is not a run of version {version.number}.", + ): + updated = self._connection.execute(*_update("versions", _version_row(version))) + if updated.rowcount == 0: + raise NotFoundError(f"The store has no version with id {version.id}.") + + @override + def delete_version(self, version_id: UUID) -> None: + deleted = self._connection.execute( + "DELETE FROM versions WHERE id = ? AND status = 'draft'", (str(version_id),) + ) + if deleted.rowcount: + return + row = self._connection.execute( + "SELECT number, status FROM versions WHERE id = ?", (str(version_id),) + ).fetchone() + if row is None: + raise NotFoundError(f"The store has no version with id {version_id}.") + raise InvalidStateError( + f"Version {row['number']} is {row['status']}, only a draft can be removed." + ) + + @override + def get_version(self, model_id: UUID, number: SemVer) -> ModelVersion | None: + row = self._connection.execute( + "SELECT * FROM versions WHERE model_id = ? AND number = ?", (str(model_id), str(number)) + ).fetchone() + return _maybe(row, _version) + + @override + def list_versions(self, model_id: UUID) -> Sequence[ModelVersion]: + rows = self._connection.execute( + "SELECT * FROM versions WHERE model_id = ?", (str(model_id),) + ) + # Numbers are stored as text, and "1.10.0" sorts before "1.2.0" there. + versions = [_version(row) for row in rows] + return sorted(versions, key=lambda version: version.number, reverse=True) + + @override + def add_run(self, run: Run) -> None: + with _constraints( + taken=f"Run {run.id} is already in the store.", + missing=f"Run {run.id} belongs to a version or a parent run that is not in the store.", + ): + self._connection.execute(*_insert("runs", _run_row(run))) + + @override + def update_run(self, run: Run) -> None: + with _constraints( + missing=f"Run {run.id} belongs to a version or a parent run that is not in the store." + ): + updated = self._connection.execute(*_update("runs", _run_row(run))) + if updated.rowcount == 0: + raise NotFoundError(f"The store has no run with id {run.id}.") + + @override + def get_run(self, run_id: UUID) -> Run | None: + row = self._connection.execute("SELECT * FROM runs WHERE id = ?", (str(run_id),)).fetchone() + return _maybe(row, _run) + + @override + def list_runs(self, version_id: UUID) -> Sequence[Run]: + rows = self._connection.execute( + "SELECT * FROM runs WHERE version_id = ? ORDER BY started_at, rowid", (str(version_id),) + ) + return [_run(row) for row in rows] + + @override + def set_alias(self, alias: Alias) -> None: + sql, row = _insert("aliases", _alias_row(alias)) + with _constraints(missing=f"The model has no version with id {alias.version_id}."): + self._connection.execute( + f"{sql} ON CONFLICT (model_id, name) DO UPDATE SET " + "version_id = excluded.version_id, updated_at = excluded.updated_at", + row, + ) + + @override + def delete_alias(self, model_id: UUID, name: str) -> None: + deleted = self._connection.execute( + "DELETE FROM aliases WHERE model_id = ? AND name = ?", (str(model_id), name) + ) + if deleted.rowcount == 0: + raise NotFoundError(f"The model has no alias {name!r}.") + + @override + def get_alias(self, model_id: UUID, name: str) -> Alias | None: + row = self._connection.execute( + "SELECT * FROM aliases WHERE model_id = ? AND name = ?", (str(model_id), name) + ).fetchone() + return _maybe(row, _alias) + + @override + def list_aliases(self, model_id: UUID) -> Sequence[Alias]: + rows = self._connection.execute( + "SELECT * FROM aliases WHERE model_id = ? ORDER BY name", (str(model_id),) + ) + return [_alias(row) for row in rows] + + +type _Row = Mapping[str, object] + + +@contextmanager +def _constraints(*, taken: str | None = None, missing: str | None = None) -> Generator[None]: + """Turn a broken database constraint into the error that the port promises.""" + try: + yield + except sqlite3.IntegrityError as error: + kind = error.sqlite_errorname + if taken and kind in ("SQLITE_CONSTRAINT_UNIQUE", "SQLITE_CONSTRAINT_PRIMARYKEY"): + raise AlreadyExistsError(taken) from error + if missing and kind == "SQLITE_CONSTRAINT_FOREIGNKEY": + raise NotFoundError(missing) from error + raise + + +def _insert(table: str, row: _Row) -> tuple[str, _Row]: + columns = ", ".join(row) + values = ", ".join(f":{column}" for column in row) + return f"INSERT INTO {table} ({columns}) VALUES ({values})", row + + +def _update(table: str, row: _Row) -> tuple[str, _Row]: + changes = ", ".join(f"{column} = :{column}" for column in row if column != "id") + return f"UPDATE {table} SET {changes} WHERE id = :id", row + + +def _maybe[T, R](value: T | None, convert: Callable[[T], R]) -> R | None: + return None if value is None else convert(value) + + +def _text(moment: datetime) -> str: + # Always UTC and always the same width, so sorting the text sorts by time. + return moment.astimezone(UTC).isoformat(timespec="microseconds") + + +def _model_row(model: Model) -> _Row: + return { + "id": str(model.id), + "name": model.name, + "flavor": model.flavor, + "task": model.task, + "created_at": _text(model.created_at), + } + + +def _model(row: sqlite3.Row) -> Model: + return Model( + id=UUID(row["id"]), + name=row["name"], + flavor=row["flavor"], + task=row["task"], + created_at=datetime.fromisoformat(row["created_at"]), + ) + + +def _version_row(version: ModelVersion) -> _Row: + return { + "id": str(version.id), + "model_id": str(version.model_id), + "number": str(version.number), + "status": str(version.status), + "base_number": _maybe(version.base, lambda base: str(base.number)), + "base_digest": _maybe(version.base, lambda base: base.digest), + "released_run_id": _maybe(version.released_run_id, str), + "content_digest": version.content_digest, + "released_at": _maybe(version.released_at, _text), + "created_at": _text(version.created_at), + } + + +def _version(row: sqlite3.Row) -> ModelVersion: + base = None + if row["base_number"] is not None: + base = VersionRef(number=SemVer.parse(row["base_number"]), digest=row["base_digest"]) + return ModelVersion( + id=UUID(row["id"]), + model_id=UUID(row["model_id"]), + number=SemVer.parse(row["number"]), + created_at=datetime.fromisoformat(row["created_at"]), + status=VersionStatus(row["status"]), + base=base, + released_run_id=_maybe(row["released_run_id"], UUID), + content_digest=row["content_digest"], + released_at=_maybe(row["released_at"], datetime.fromisoformat), + ) + + +def _run_row(run: Run) -> _Row: + return { + "id": str(run.id), + "version_id": str(run.version_id), + "parent_run_id": _maybe(run.parent_run_id, str), + "status": str(run.status), + "started_at": _text(run.started_at), + "finished_at": _maybe(run.finished_at, _text), + "miniops_version": run.miniops_version, + "params": json.dumps(dict(run.params)), + "metrics": json.dumps(dict(run.metrics)), + "dataset_uri": _maybe(run.dataset, lambda dataset: dataset.uri), + "dataset_fingerprint": _maybe(run.dataset, lambda dataset: dataset.fingerprint), + **_artifact_row("snapshot", run.snapshot), + "lock_digest": run.lock_digest, + "code_commit": run.code_commit, + **_artifact_row("artifact", run.artifact), + "signature": _maybe(run.signature, lambda signature: json.dumps(asdict(signature))), + "error": run.error, + } + + +def _run(row: sqlite3.Row) -> Run: + dataset = None + if row["dataset_uri"] is not None: + dataset = DatasetRef(uri=row["dataset_uri"], fingerprint=row["dataset_fingerprint"]) + return Run( + id=UUID(row["id"]), + version_id=UUID(row["version_id"]), + parent_run_id=_maybe(row["parent_run_id"], UUID), + status=RunStatus(row["status"]), + started_at=datetime.fromisoformat(row["started_at"]), + finished_at=_maybe(row["finished_at"], datetime.fromisoformat), + miniops_version=row["miniops_version"], + params=json.loads(row["params"]), + metrics=json.loads(row["metrics"]), + dataset=dataset, + snapshot=_artifact(row, "snapshot"), + lock_digest=row["lock_digest"], + code_commit=row["code_commit"], + artifact=_artifact(row, "artifact"), + signature=_maybe(row["signature"], _signature), + error=row["error"], + ) + + +def _artifact_row(prefix: str, ref: ArtifactRef | None) -> _Row: + key, digest, size = (None, None, None) if ref is None else (ref.key, ref.digest, ref.size) + return {f"{prefix}_key": key, f"{prefix}_digest": digest, f"{prefix}_size": size} + + +def _artifact(row: sqlite3.Row, prefix: str) -> ArtifactRef | None: + if row[f"{prefix}_key"] is None: + return None + return ArtifactRef( + key=row[f"{prefix}_key"], digest=row[f"{prefix}_digest"], size=row[f"{prefix}_size"] + ) + + +def _signature(text: str) -> Signature: + columns = json.loads(text) + return Signature( + inputs=tuple(Column(**column) for column in columns["inputs"]), + outputs=tuple(Column(**column) for column in columns["outputs"]), + ) + + +def _alias_row(alias: Alias) -> _Row: + return { + "model_id": str(alias.model_id), + "name": alias.name, + "version_id": str(alias.version_id), + "updated_at": _text(alias.updated_at), + } + + +def _alias(row: sqlite3.Row) -> Alias: + return Alias( + model_id=UUID(row["model_id"]), + name=row["name"], + version_id=UUID(row["version_id"]), + updated_at=datetime.fromisoformat(row["updated_at"]), + ) diff --git a/src/miniops/adapters/local/schema.py b/src/miniops/adapters/local/schema.py new file mode 100644 index 0000000..7585fcc --- /dev/null +++ b/src/miniops/adapters/local/schema.py @@ -0,0 +1,110 @@ +import sqlite3 +from contextlib import closing +from pathlib import Path + +from miniops.core.errors import IncompatibleStoreError + +# A released script never changes: a change to the schema is a new script at the end. +MIGRATIONS: tuple[str, ...] = ( + """ + CREATE TABLE models ( + id TEXT NOT NULL PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + flavor TEXT NOT NULL, + task TEXT NOT NULL, + created_at TEXT NOT NULL + ); + + CREATE TABLE versions ( + id TEXT NOT NULL PRIMARY KEY, + model_id TEXT NOT NULL REFERENCES models (id), + number TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('draft', 'released', 'deleted')), + base_number TEXT, + base_digest TEXT, + released_run_id TEXT, + content_digest TEXT, + released_at TEXT, + created_at TEXT NOT NULL, + UNIQUE (model_id, number), + UNIQUE (model_id, id), + FOREIGN KEY (id, released_run_id) REFERENCES runs (version_id, id) + ); + + CREATE TABLE runs ( + id TEXT NOT NULL PRIMARY KEY, + version_id TEXT NOT NULL REFERENCES versions (id) ON DELETE CASCADE, + parent_run_id TEXT REFERENCES runs (id), + status TEXT NOT NULL CHECK (status IN ('running', 'succeeded', 'failed', 'interrupted')), + started_at TEXT NOT NULL, + finished_at TEXT, + miniops_version TEXT NOT NULL, + params TEXT NOT NULL, + metrics TEXT NOT NULL, + dataset_uri TEXT, + dataset_fingerprint TEXT, + snapshot_key TEXT, + snapshot_digest TEXT, + snapshot_size INTEGER, + lock_digest TEXT, + code_commit TEXT, + artifact_key TEXT, + artifact_digest TEXT, + artifact_size INTEGER, + signature TEXT, + error TEXT, + UNIQUE (version_id, id) + ); + + CREATE TABLE aliases ( + model_id TEXT NOT NULL, + name TEXT NOT NULL, + version_id TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (model_id, name), + FOREIGN KEY (model_id, version_id) REFERENCES versions (model_id, id) + ); + """, +) + + +def schema_version(connection: sqlite3.Connection) -> int: + """Return the schema version of a store. An empty database has version 0.""" + version: int = connection.execute("PRAGMA user_version").fetchone()[0] + return version + + +def migrate(connection: sqlite3.Connection, path: Path) -> None: + """Bring a store up to the newest schema, after saving a copy of an older one. + + Raises: + IncompatibleStoreError: If the store was written by a newer miniOPs. + """ + version = schema_version(connection) + newest = len(MIGRATIONS) + if version > newest: + raise IncompatibleStoreError( + f"The store at {path} has schema version {version}, but this miniOPs reads " + f"up to version {newest}. Upgrade miniOPs to use this store." + ) + if version == newest: + return + if version > 0: + _back_up(connection, path, version) + + connection.execute("BEGIN IMMEDIATE") + try: + # Read the version again: another process may have upgraded the store in the meantime. + for number in range(schema_version(connection), newest): + connection.executescript(MIGRATIONS[number]) + connection.execute(f"PRAGMA user_version = {number + 1}") + connection.execute("COMMIT") + except BaseException: + connection.execute("ROLLBACK") + raise + + +def _back_up(connection: sqlite3.Connection, path: Path, version: int) -> None: + backup = path.with_name(f"{path.name}.v{version}.bak") + with closing(sqlite3.connect(backup)) as copy: + connection.backup(copy) diff --git a/src/miniops/core/errors.py b/src/miniops/core/errors.py index 10e3578..5b42117 100644 --- a/src/miniops/core/errors.py +++ b/src/miniops/core/errors.py @@ -16,3 +16,7 @@ class NotFoundError(MiniopsError): class IntegrityError(MiniopsError): """Stored content does not match the digest recorded for it.""" + + +class IncompatibleStoreError(MiniopsError): + """The store was written by a newer miniOPs than the one that opened it.""" diff --git a/src/miniops/ports/metadata.py b/src/miniops/ports/metadata.py index cec2ac9..99a7a21 100644 --- a/src/miniops/ports/metadata.py +++ b/src/miniops/ports/metadata.py @@ -27,9 +27,11 @@ def list_models(self) -> Sequence[Model]: """Return all models, ordered by name.""" def add_version(self, version: ModelVersion) -> None: - """Add a version to its model. + """Add a draft to its model. A version is released later, with ``update_version``. Raises: + InvalidStateError: If the version is not a draft. + NotFoundError: If the store does not have its model. AlreadyExistsError: If the model has this number already, even as a deleted version. """ @@ -37,10 +39,21 @@ def update_version(self, version: ModelVersion) -> None: """Replace the stored version that has the same id. Raises: - NotFoundError: If the store has no version with this id. + NotFoundError: If the store has no version with this id, or the released run + is not one of its runs. AlreadyExistsError: If another version of the model has the new number. """ + def delete_version(self, version_id: UUID) -> None: + """Remove a draft with its runs, which frees its number. + + A released version is never removed: it is deleted by storing its tombstone. + + Raises: + NotFoundError: If the store has no version with this id. + InvalidStateError: If the version is not a draft. + """ + def get_version(self, model_id: UUID, number: SemVer) -> ModelVersion | None: """Return the version of a model with this number, or None.""" @@ -51,6 +64,7 @@ def add_run(self, run: Run) -> None: """Add a run to its version. Raises: + NotFoundError: If the store does not have its version or its parent run. AlreadyExistsError: If the store has a run with this id. """ @@ -58,7 +72,8 @@ def update_run(self, run: Run) -> None: """Replace the stored run that has the same id. Raises: - NotFoundError: If the store has no run with this id. + NotFoundError: If the store has no run with this id, or does not have its version + or its parent run. """ def get_run(self, run_id: UUID) -> Run | None: @@ -68,7 +83,18 @@ def list_runs(self, version_id: UUID) -> Sequence[Run]: """Return the runs of a version, oldest first.""" def set_alias(self, alias: Alias) -> None: - """Create the alias, or move it when the model already has one with this name.""" + """Create the alias, or move it when the model already has one with this name. + + Raises: + NotFoundError: If the model has no version with this id. + """ + + def delete_alias(self, model_id: UUID, name: str) -> None: + """Remove an alias of a model. + + Raises: + NotFoundError: If the model has no alias with this name. + """ def get_alias(self, model_id: UUID, name: str) -> Alias | None: """Return the alias of a model with this name, or None.""" diff --git a/tests/adapters/__init__.py b/tests/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/adapters/local/__init__.py b/tests/adapters/local/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/adapters/local/test_metadata.py b/tests/adapters/local/test_metadata.py new file mode 100644 index 0000000..f0dc818 --- /dev/null +++ b/tests/adapters/local/test_metadata.py @@ -0,0 +1,415 @@ +import math +import sqlite3 +from collections.abc import Iterator +from dataclasses import replace +from datetime import UTC, datetime, timedelta, timezone +from pathlib import Path +from typing import cast +from uuid import uuid4 + +import pytest + +from miniops.adapters.local.metadata import SqliteMetadataStore +from miniops.core.entities import ( + Alias, + ArtifactRef, + Column, + DatasetRef, + Model, + ModelVersion, + Run, + RunStatus, + Signature, + VersionRef, + VersionStatus, +) +from miniops.core.errors import AlreadyExistsError, InvalidStateError, NotFoundError +from miniops.core.versions import SemVer + +NOW = datetime(2026, 11, 11, 12, 0, tzinfo=UTC) +LATER = NOW + timedelta(minutes=5) +DIGEST = "a" * 64 +ARTIFACT = ArtifactRef(key="models/churn/model.skops", digest=DIGEST, size=1024) + + +@pytest.fixture +def store(tmp_path: Path) -> Iterator[SqliteMetadataStore]: + with SqliteMetadataStore(tmp_path / "metadata.db") as opened: + yield opened + + +def add_model(store: SqliteMetadataStore, name: str = "churn") -> Model: + model = Model( + id=uuid4(), name=name, flavor="sklearn", task="binary_classification", created_at=NOW + ) + store.add_model(model) + return model + + +def add_draft(store: SqliteMetadataStore, model: Model, number: str = "1.0.0") -> ModelVersion: + version = ModelVersion( + id=uuid4(), model_id=model.id, number=SemVer.parse(number), created_at=NOW + ) + store.add_version(version) + return version + + +def add_run(store: SqliteMetadataStore, version: ModelVersion, started_at: datetime = NOW) -> Run: + run = Run(id=uuid4(), version_id=version.id, started_at=started_at, miniops_version="0.1.0a1") + store.add_run(run) + return run + + +def release(store: SqliteMetadataStore, version: ModelVersion) -> ModelVersion: + run = add_run(store, version).finish(RunStatus.SUCCEEDED, LATER, artifact=ARTIFACT) + store.update_run(run) + released = version.release(run, DIGEST, LATER) + store.update_version(released) + return released + + +def test_a_model_comes_back_as_it_was_stored(store: SqliteMetadataStore) -> None: + model = add_model(store) + + assert store.get_model("churn") == model + + +def test_an_unknown_model_is_none(store: SqliteMetadataStore) -> None: + assert store.get_model("churn") is None + + +def test_models_are_listed_by_name(store: SqliteMetadataStore) -> None: + for name in ("fraud", "churn", "demand"): + add_model(store, name) + + assert [model.name for model in store.list_models()] == ["churn", "demand", "fraud"] + + +def test_a_model_name_is_used_once(store: SqliteMetadataStore) -> None: + add_model(store) + + with pytest.raises(AlreadyExistsError, match="'churn' is already in the store"): + add_model(store) + + +def test_a_draft_comes_back_as_it_was_stored(store: SqliteMetadataStore) -> None: + model = add_model(store) + version = ModelVersion( + id=uuid4(), + model_id=model.id, + number=SemVer(1, 1, 0), + created_at=NOW, + base=VersionRef(number=SemVer(1, 0, 0), digest=DIGEST), + ) + + store.add_version(version) + + assert store.get_version(model.id, SemVer(1, 1, 0)) == version + assert store.get_version(model.id, SemVer(2, 0, 0)) is None + + +def test_a_version_needs_its_model(store: SqliteMetadataStore) -> None: + stranger = Model(id=uuid4(), name="churn", flavor="sklearn", task="regression", created_at=NOW) + version = ModelVersion(id=uuid4(), model_id=stranger.id, number=SemVer(1, 0, 0), created_at=NOW) + + with pytest.raises(NotFoundError, match="no model"): + store.add_version(version) + + +def test_a_version_enters_the_store_as_a_draft(store: SqliteMetadataStore) -> None: + model = add_model(store) + released = release(store, add_draft(store, model)) + + with pytest.raises(InvalidStateError, match="enters the store as a draft"): + store.add_version(replace(released, id=uuid4(), number=SemVer(2, 0, 0))) + + +def test_versions_are_listed_highest_first(store: SqliteMetadataStore) -> None: + model = add_model(store) + for number in ("1.2.0", "1.10.0", "2.0.0", "1.2.1"): + add_draft(store, model, number) + + numbers = [str(version.number) for version in store.list_versions(model.id)] + + assert numbers == ["2.0.0", "1.10.0", "1.2.1", "1.2.0"] + + +def test_a_number_is_used_once_per_model(store: SqliteMetadataStore) -> None: + churn, fraud = add_model(store, "churn"), add_model(store, "fraud") + add_draft(store, churn, "1.0.0") + + add_draft(store, fraud, "1.0.0") + with pytest.raises(AlreadyExistsError, match=r"already has a version 1\.0\.0"): + add_draft(store, churn, "1.0.0") + + +def test_a_renumbered_draft_is_found_under_its_new_number(store: SqliteMetadataStore) -> None: + model = add_model(store) + version = add_draft(store, model, "1.1.0").renumber(SemVer(2, 0, 0)) + + store.update_version(version) + + assert store.get_version(model.id, SemVer(2, 0, 0)) == version + assert store.get_version(model.id, SemVer(1, 1, 0)) is None + + +def test_a_draft_cannot_take_the_number_of_another_version(store: SqliteMetadataStore) -> None: + model = add_model(store) + add_draft(store, model, "1.0.0") + version = add_draft(store, model, "1.1.0") + + with pytest.raises(AlreadyExistsError, match=r"already has a version 1\.0\.0"): + store.update_version(version.renumber(SemVer(1, 0, 0))) + + +def test_only_a_stored_version_can_be_updated(store: SqliteMetadataStore) -> None: + model = add_model(store) + version = ModelVersion(id=uuid4(), model_id=model.id, number=SemVer(1, 0, 0), created_at=NOW) + + with pytest.raises(NotFoundError, match="no version"): + store.update_version(version) + + +def test_a_released_version_comes_back_with_its_run_digest_and_date( + store: SqliteMetadataStore, +) -> None: + model = add_model(store) + released = release(store, add_draft(store, model)) + + stored = store.get_version(model.id, SemVer(1, 0, 0)) + + assert stored == released + assert stored.status is VersionStatus.RELEASED + + +def test_a_version_is_released_only_with_its_own_run(store: SqliteMetadataStore) -> None: + model = add_model(store) + version, other = add_draft(store, model, "1.0.0"), add_draft(store, model, "2.0.0") + run = add_run(store, other).finish(RunStatus.SUCCEEDED, LATER, artifact=ARTIFACT) + store.update_run(run) + forged = replace( + version, + status=VersionStatus.RELEASED, + released_run_id=run.id, + content_digest=DIGEST, + released_at=LATER, + ) + + with pytest.raises(NotFoundError, match=r"is not a run of version 1\.0\.0"): + store.update_version(forged) + + +def test_a_deleted_version_keeps_its_number(store: SqliteMetadataStore) -> None: + model = add_model(store) + tombstone = release(store, add_draft(store, model)).delete() + + store.update_version(tombstone) + + assert store.get_version(model.id, SemVer(1, 0, 0)) == tombstone + with pytest.raises(AlreadyExistsError, match=r"already has a version 1\.0\.0"): + add_draft(store, model, "1.0.0") + + +def test_removing_a_draft_takes_its_runs_and_frees_its_number(store: SqliteMetadataStore) -> None: + model = add_model(store) + version = add_draft(store, model) + tuning = add_run(store, version) + trial = replace(add_run(store, version), parent_run_id=tuning.id) + store.update_run(trial) + + store.delete_version(version.id) + + assert store.list_versions(model.id) == [] + assert store.get_run(tuning.id) is None + assert store.get_run(trial.id) is None + add_draft(store, model, "1.0.0") + + +def test_only_a_draft_can_be_removed(store: SqliteMetadataStore) -> None: + model = add_model(store) + released = release(store, add_draft(store, model)) + + with pytest.raises(InvalidStateError, match=r"1\.0\.0 is released, only a draft"): + store.delete_version(released.id) + with pytest.raises(NotFoundError, match="no version"): + store.delete_version(uuid4()) + + +def test_a_run_comes_back_as_it_was_stored(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + tuning = add_run(store, version) + run = Run( + id=uuid4(), + version_id=version.id, + started_at=NOW, + miniops_version="0.1.0a1", + status=RunStatus.SUCCEEDED, + parent_run_id=tuning.id, + params={"max_depth": 6, "class_weight": {"0": 1.0, "1": 2.5}, "features": ["age", "plan"]}, + metrics={"roc_auc": 0.87, "log_loss": 0.31}, + dataset=DatasetRef(uri="data/train.parquet", fingerprint=DIGEST), + snapshot=ArtifactRef(key="snapshots/ab/cdef.zip", digest="b" * 64, size=2048), + lock_digest="c" * 64, + code_commit="5d73e7c", + artifact=ARTIFACT, + signature=Signature( + inputs=(Column("age", "Int64"), Column("plan", "String", nullable=True)), + outputs=(Column("churn", "Float64"),), + ), + finished_at=LATER, + ) + + store.add_run(run) + + assert store.get_run(run.id) == run + + +def test_a_failed_run_keeps_its_error(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + run = add_run(store, version).finish(RunStatus.FAILED, LATER, error="out of memory") + + store.update_run(run) + + assert store.get_run(run.id) == run + + +def test_an_unknown_run_is_none(store: SqliteMetadataStore) -> None: + assert store.get_run(uuid4()) is None + + +def test_a_run_needs_its_version_and_its_parent(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + orphan = Run(id=uuid4(), version_id=uuid4(), started_at=NOW, miniops_version="0.1.0a1") + trial = replace(orphan, version_id=version.id, parent_run_id=uuid4()) + + with pytest.raises(NotFoundError, match="not in the store"): + store.add_run(orphan) + with pytest.raises(NotFoundError, match="not in the store"): + store.add_run(trial) + + +def test_a_run_id_is_used_once(store: SqliteMetadataStore) -> None: + run = add_run(store, add_draft(store, add_model(store))) + + with pytest.raises(AlreadyExistsError, match="already in the store"): + store.add_run(run) + + +def test_only_a_stored_run_can_be_updated(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + run = Run(id=uuid4(), version_id=version.id, started_at=NOW, miniops_version="0.1.0a1") + + with pytest.raises(NotFoundError, match="no run"): + store.update_run(run) + + +def test_runs_are_listed_oldest_first(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + third = add_run(store, version, NOW + timedelta(seconds=2)) + first = add_run(store, version, NOW) + second = add_run(store, version, NOW + timedelta(seconds=1, microseconds=5)) + + assert store.list_runs(version.id) == [first, second, third] + + +def test_a_metric_that_is_not_a_number_survives(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + run = add_run(store, version).finish( + RunStatus.FAILED, LATER, metrics={"loss": math.nan}, error="the loss diverged" + ) + + store.update_run(run) + + stored = store.get_run(run.id) + assert stored is not None + assert math.isnan(stored.metrics["loss"]) + + +def test_a_moment_comes_back_in_utc(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + warsaw = datetime(2026, 11, 11, 13, 0, tzinfo=timezone(timedelta(hours=1))) + + run = add_run(store, version, warsaw) + + stored = store.get_run(run.id) + assert stored is not None + assert stored.started_at == warsaw + assert stored.started_at.utcoffset() == timedelta(0) + + +def test_a_status_the_schema_does_not_know_is_refused(store: SqliteMetadataStore) -> None: + version = add_draft(store, add_model(store)) + run = Run( + id=uuid4(), + version_id=version.id, + started_at=NOW, + miniops_version="0.1.0a1", + status=cast(RunStatus, "paused"), + finished_at=LATER, + ) + + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + store.add_run(run) + + +def test_an_alias_is_created_and_then_moved(store: SqliteMetadataStore) -> None: + model = add_model(store) + first = release(store, add_draft(store, model, "1.0.0")) + second = release(store, add_draft(store, model, "1.1.0")) + alias = Alias(model_id=model.id, name="champion", version_id=first.id, updated_at=NOW) + + store.set_alias(alias) + assert store.get_alias(model.id, "champion") == alias + + moved = replace(alias, version_id=second.id, updated_at=LATER) + store.set_alias(moved) + assert store.list_aliases(model.id) == [moved] + + +def test_aliases_are_listed_by_name(store: SqliteMetadataStore) -> None: + model = add_model(store) + version = release(store, add_draft(store, model)) + for name in ("champion", "baseline", "challenger"): + store.set_alias(Alias(model_id=model.id, name=name, version_id=version.id, updated_at=NOW)) + + names = [alias.name for alias in store.list_aliases(model.id)] + + assert names == ["baseline", "challenger", "champion"] + + +def test_an_alias_points_at_a_version_of_its_own_model(store: SqliteMetadataStore) -> None: + churn, fraud = add_model(store, "churn"), add_model(store, "fraud") + version = release(store, add_draft(store, fraud)) + alias = Alias(model_id=churn.id, name="champion", version_id=version.id, updated_at=NOW) + + with pytest.raises(NotFoundError, match="no version"): + store.set_alias(alias) + with pytest.raises(NotFoundError, match="no version"): + store.set_alias(replace(alias, version_id=uuid4())) + + +def test_an_alias_can_be_removed(store: SqliteMetadataStore) -> None: + model = add_model(store) + version = release(store, add_draft(store, model)) + store.set_alias( + Alias(model_id=model.id, name="champion", version_id=version.id, updated_at=NOW) + ) + + store.delete_alias(model.id, "champion") + + assert store.get_alias(model.id, "champion") is None + with pytest.raises(NotFoundError, match="no alias 'champion'"): + store.delete_alias(model.id, "champion") + + +def test_a_store_keeps_its_content_between_sessions(tmp_path: Path) -> None: + path = tmp_path / "project" / ".miniops" / "metadata.db" + with SqliteMetadataStore(path) as store: + model = add_model(store) + version = add_draft(store, model) + run = add_run(store, version) + + with SqliteMetadataStore(path) as store: + assert store.list_models() == [model] + assert store.list_versions(model.id) == [version] + assert store.list_runs(version.id) == [run] diff --git a/tests/adapters/local/test_schema.py b/tests/adapters/local/test_schema.py new file mode 100644 index 0000000..ff9118a --- /dev/null +++ b/tests/adapters/local/test_schema.py @@ -0,0 +1,93 @@ +import sqlite3 +from contextlib import closing +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +import pytest + +from miniops.adapters.local import schema +from miniops.adapters.local.metadata import SqliteMetadataStore +from miniops.core.entities import Model +from miniops.core.errors import IncompatibleStoreError + +MODEL = Model( + id=uuid4(), + name="churn", + flavor="sklearn", + task="binary_classification", + created_at=datetime(2026, 11, 11, 12, 0, tzinfo=UTC), +) + + +def version_of(path: Path) -> int: + with closing(sqlite3.connect(path)) as connection: + return schema.schema_version(connection) + + +def tables_of(path: Path) -> set[str]: + with closing(sqlite3.connect(path)) as connection: + rows = connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'") + return {name for (name,) in rows} + + +def test_a_new_store_gets_the_newest_schema(tmp_path: Path) -> None: + path = tmp_path / "metadata.db" + + SqliteMetadataStore(path).close() + + assert version_of(path) == len(schema.MIGRATIONS) + assert tables_of(path) == {"models", "versions", "runs", "aliases"} + assert list(tmp_path.iterdir()) == [path] + + +def test_an_older_store_is_upgraded_after_a_backup( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "metadata.db" + with SqliteMetadataStore(path) as store: + store.add_model(MODEL) + older = len(schema.MIGRATIONS) + monkeypatch.setattr( + schema, "MIGRATIONS", (*schema.MIGRATIONS, "CREATE TABLE deployments (id TEXT);") + ) + + with SqliteMetadataStore(path) as store: + assert store.list_models() == [MODEL] + + backup = tmp_path / f"metadata.db.v{older}.bak" + assert version_of(path) == older + 1 + assert "deployments" in tables_of(path) + assert version_of(backup) == older + assert "deployments" not in tables_of(backup) + with SqliteMetadataStore(backup) as store: + assert store.list_models() == [MODEL] + + +def test_a_failed_upgrade_changes_nothing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + path = tmp_path / "metadata.db" + with SqliteMetadataStore(path) as store: + store.add_model(MODEL) + older = len(schema.MIGRATIONS) + broken = "CREATE TABLE deployments (id TEXT); CREATE TABLE models (id TEXT);" + monkeypatch.setattr(schema, "MIGRATIONS", (*schema.MIGRATIONS, broken)) + + with pytest.raises(sqlite3.OperationalError, match="already exists"): + SqliteMetadataStore(path) + + assert version_of(path) == older + assert "deployments" not in tables_of(path) + + +def test_a_store_from_a_newer_miniops_is_refused(tmp_path: Path) -> None: + path = tmp_path / "metadata.db" + SqliteMetadataStore(path).close() + newer = len(schema.MIGRATIONS) + 1 + with closing(sqlite3.connect(path)) as connection: + connection.execute(f"PRAGMA user_version = {newer}") + + with pytest.raises(IncompatibleStoreError, match="Upgrade miniOPs"): + SqliteMetadataStore(path) + + # Windows refuses to delete a file that is still open, so this also proves it was closed. + path.unlink()