diff --git a/docs/README.md b/docs/README.md index 60cc8843..af235f5c 100644 --- a/docs/README.md +++ b/docs/README.md @@ -3,3 +3,7 @@ Product and architecture docs land here. Start with the repo-root [README](../README.md) and [CONTRIBUTING](../CONTRIBUTING.md); the architecture contracts (kernel purity, headless annotator) are described there and enforced in CI. + +| Doc | Covers | +| --- | --- | +| [persistence.md](persistence.md) | The metadata store: repositories, unit of work, table layout, migrations and `format_version` | diff --git a/docs/persistence.md b/docs/persistence.md new file mode 100644 index 00000000..7ca962be --- /dev/null +++ b/docs/persistence.md @@ -0,0 +1,95 @@ +# Persistence + +One workspace is one SQLite file. `SqliteMetadataStore` is the default +`MetadataStore` adapter; everything above it — services, the REST surface, the CLI — +talks to the **port**, never to SQLAlchemy. + +``` +kernel/ports/metadata_store.py Repository[T], UnitOfWork, MetadataStore — no SQL +kernel/adapters/_tables.py SQLAlchemy tables (private) +kernel/adapters/_mappers.py row <-> pydantic translation (private) +kernel/adapters/migrations.py the ordered migration list +kernel/adapters/sqlite_metadata_store.py +``` + +`kernel/adapters/__init__.py` exports only `SqliteMetadataStore`, so no SQLAlchemy type +appears in a domain or port signature. If you find yourself wanting one there, the +mapping layer is the thing to extend. + +## The repository contract + +Every persisted entity has a UUID primary key and **at most one parent** — a Project +belongs to a Workspace, an Annotation to an Asset. That regularity is why a single +generic repository serves all fourteen entity types: + +```python +with store.unit_of_work() as uow: + project = uow.projects.add(Project(workspace_id=workspace.id, name="road-signs")) + assets = uow.assets.list(project.id) # scoped by the one parent FK +``` + +- `add` raises `EntityAlreadyExists` on a primary-key collision; `update` raises + `EntityNotFound`. They are deliberately not one upsert — a service that inserts a + duplicate or updates something deleted has a bug and should hear about it. +- `delete` returns `False` rather than raising when there was nothing to remove. +- `list(parent_id)` on `workspaces` — the one root entity — raises `ValueError`. +- Ordering is insertion order (SQLite's implicit `rowid`). + +Uniqueness beyond the primary key (project names, schema versions) is a **service** +concern. The store persists shapes; it does not know the rules. + +## Unit of work + +The transaction boundary is one operation on a Project aggregate: + +```python +with store.unit_of_work() as uow: + ... # everything here commits together +``` + +Clean exit commits; any exception rolls the whole block back. A batch approval that +partitions into jobs must never leave half its jobs behind, which is why the scope is +the operation and not the individual write. + +## What is a column, what is a table, what is JSON + +| Kind | Storage | Why | +| --- | --- | --- | +| Relations that get mutated element by element | Child table — `batch_asset`, `annotation_job_asset` | Membership and per-asset progress are edited one row at a time and queried from the asset side. `batch_asset.position` preserves order. | +| Immutable nested values | JSON column — `annotation_schema.classes`, `annotation.geometry`, `release.manifest` | A schema version must rehydrate byte-identical, and nothing queries a single `LabelClass` in SQL. Child tables would only add ordering columns. | +| Timestamps | TEXT holding ISO-8601 **with offset** | SQLite's `DATETIME` storage drops the timezone. Domain timestamps are timezone-aware UTC and a naive value is rejected at construction. | + +Foreign keys are declared `ON DELETE CASCADE` — and the store issues +`PRAGMA foreign_keys = ON` for every connection, because SQLite ships with foreign keys +**off**. Without that pragma every constraint here would be decorative. + +## Migrations and `format_version` + +There is no alembic. A local-first, single-file, single-writer store does not need a +migration framework, and adding one would mean keeping `format_version` in sync with a +second ledger by hand. Instead: + +```python +MIGRATIONS: list[Migration] = [ + Migration(version=1, name="initial_schema", upgrade=_create_initial_schema), +] +FORMAT_VERSION: int = MIGRATIONS[-1].version +``` + +`initialize()` reads the version stamped in `_visionset_meta` and runs whatever is +missing: + +| Stored | What happens | +| --- | --- | +| absent | every migration runs; the file is stamped at `FORMAT_VERSION` | +| equal to `FORMAT_VERSION` | nothing — `initialize()` is idempotent | +| lower | the pending migrations run; the file is restamped | +| higher | `WorkspaceFormatTooNew` — migrations only run forward | + +**Adding migration 002:** append a `Migration` with the next version and an `upgrade` +taking a live `Connection`. Never edit an existing migration — a workspace already +stamped at that version will never run it again. `FORMAT_VERSION` is derived from the +list, so it cannot drift. + +`format_version` here is the *database* generation. Validating the on-disk workspace +layout around it (directories, blob-store root) belongs to `WorkspaceService`. diff --git a/src/visionset/kernel/__init__.py b/src/visionset/kernel/__init__.py index c109a51d..f0aac064 100644 --- a/src/visionset/kernel/__init__.py +++ b/src/visionset/kernel/__init__.py @@ -5,3 +5,17 @@ The boundary is machine-enforced by import-linter contracts (see pyproject.toml) and by the architecture tests in ``tests/architecture/``. """ + +from visionset.kernel.errors import ( + EntityAlreadyExists, + EntityNotFound, + VisionSetError, + WorkspaceFormatTooNew, +) + +__all__ = [ + "EntityAlreadyExists", + "EntityNotFound", + "VisionSetError", + "WorkspaceFormatTooNew", +] diff --git a/src/visionset/kernel/adapters/_mappers.py b/src/visionset/kernel/adapters/_mappers.py new file mode 100644 index 00000000..23141712 --- /dev/null +++ b/src/visionset/kernel/adapters/_mappers.py @@ -0,0 +1,296 @@ +"""Translation between SQLAlchemy rows and pydantic domain models. + +This module is the reason the domain never sees a SQLAlchemy type. Each entity +gets one ``EntityMapping`` describing its table, its parent column, and the two +directions of the conversion; the repository in +``sqlite_metadata_store`` is written once against that description rather than +fourteen times against fourteen tables. + +Most entities are flat — every field is a column — and share +``_flat_mapping``. The six that are not say so explicitly: + +- ``AnnotationSchema``, ``Annotation`` and ``Release`` hold immutable nested + values, encoded as JSON. +- ``Batch`` and ``AnnotationJob`` own child tables, so their mappings carry a + ``sync_children`` hook and rebuild their collections on read. +- ``DatasetChange`` encodes its UUID list and its timezone-aware timestamp. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Protocol, cast +from uuid import UUID + +from pydantic import BaseModel, TypeAdapter +from sqlalchemy import delete, select +from sqlalchemy.orm import Session + +from visionset.kernel.adapters import _tables as t +from visionset.kernel.domain import ( + Annotation, + AnnotationJob, + AnnotationSchema, + Asset, + AssetProgress, + Batch, + Dataset, + DatasetChange, + DatasetMember, + Geometry, + IngestJob, + LabelClass, + Manifest, + Project, + Release, + Source, + TaskGroup, + Workspace, +) + +_geometry_adapter: TypeAdapter[Geometry] = TypeAdapter(Geometry) + + +class Entity(Protocol): + """Anything the store persists: a pydantic model addressed by a UUID.""" + + id: UUID + + +@dataclass(frozen=True) +class EntityMapping[T: Entity]: + """How one domain model is stored, read back, and scoped to its parent. + + ``parent_column`` is ``None`` only for ``Workspace``, the single root + entity; for everything else it names the one foreign key that + ``Repository.list`` filters on. + """ + + row: type[t.Base] + parent_column: str | None + to_row: Callable[[T], t.Base] + to_domain: Callable[[Session, Any], T] + sync_children: Callable[[Session, T], None] | None = None + + +def _columns(row: Any) -> dict[str, Any]: + return {column.name: getattr(row, column.name) for column in row.__table__.columns} + + +def _flat_mapping[M: Entity]( + domain: type[M], row: type[t.Base], parent_column: str | None +) -> EntityMapping[M]: + """Mapping for an entity whose fields are exactly its table's columns. + + The casts bridge ``Entity`` (which promises only a ``id``) and pydantic's + API. Every call site below passes a real ``BaseModel`` subclass; expressing + that as a bound would need an intersection type, which Python has no syntax + for. + """ + model = cast(type[BaseModel], domain) + + def to_row(entity: M) -> t.Base: + return row(**cast(BaseModel, entity).model_dump()) + + def to_domain(_: Session, stored: Any) -> M: + return cast(M, model.model_validate(_columns(stored))) + + return EntityMapping(row=row, parent_column=parent_column, to_row=to_row, to_domain=to_domain) + + +# --- Entities with nested immutable values, stored as JSON ------------------ + + +def _schema_to_row(entity: AnnotationSchema) -> t.Base: + return t.AnnotationSchemaRow( + id=entity.id, + project_id=entity.project_id, + version=entity.version, + classes=[c.model_dump(mode="json") for c in entity.classes], + ) + + +def _schema_to_domain(_: Session, row: Any) -> AnnotationSchema: + return AnnotationSchema( + id=row.id, + project_id=row.project_id, + version=row.version, + classes=[LabelClass.model_validate(c) for c in row.classes], + ) + + +def _annotation_to_row(entity: Annotation) -> t.Base: + return t.AnnotationRow( + id=entity.id, + asset_id=entity.asset_id, + label_class=entity.label_class, + schema_version=entity.schema_version, + geometry=entity.geometry.model_dump(mode="json"), + provenance=entity.provenance, + model_ref=entity.model_ref, + confidence=entity.confidence, + ) + + +def _annotation_to_domain(_: Session, row: Any) -> Annotation: + return Annotation( + id=row.id, + asset_id=row.asset_id, + label_class=row.label_class, + schema_version=row.schema_version, + geometry=_geometry_adapter.validate_python(row.geometry), + provenance=row.provenance, + model_ref=row.model_ref, + confidence=row.confidence, + ) + + +def _release_to_row(entity: Release) -> t.Base: + return t.ReleaseRow( + id=entity.id, + dataset_id=entity.dataset_id, + tag=entity.tag, + manifest=entity.manifest.model_dump(mode="json"), + ) + + +def _release_to_domain(_: Session, row: Any) -> Release: + return Release( + id=row.id, + dataset_id=row.dataset_id, + tag=row.tag, + manifest=Manifest.model_validate(row.manifest), + ) + + +def _change_to_row(entity: DatasetChange) -> t.Base: + return t.DatasetChangeRow( + id=entity.id, + dataset_id=entity.dataset_id, + operation=entity.operation, + subject_ids=[str(s) for s in entity.subject_ids], + actor=entity.actor, + occurred_at=entity.occurred_at.isoformat(), + ) + + +def _change_to_domain(_: Session, row: Any) -> DatasetChange: + return DatasetChange( + id=row.id, + dataset_id=row.dataset_id, + operation=row.operation, + subject_ids=[UUID(s) for s in row.subject_ids], + actor=row.actor, + occurred_at=datetime.fromisoformat(row.occurred_at), + ) + + +# --- Entities owning a child table ----------------------------------------- + + +def _batch_to_row(entity: Batch) -> t.Base: + return t.BatchRow( + id=entity.id, project_id=entity.project_id, name=entity.name, state=entity.state + ) + + +def _batch_to_domain(session: Session, row: Any) -> Batch: + members = session.scalars( + select(t.BatchAssetRow.asset_id) + .where(t.BatchAssetRow.batch_id == row.id) + .order_by(t.BatchAssetRow.position) + ).all() + return Batch( + id=row.id, + project_id=row.project_id, + name=row.name, + state=row.state, + asset_ids=list(members), + ) + + +def _batch_sync_children(session: Session, entity: Batch) -> None: + session.execute(delete(t.BatchAssetRow).where(t.BatchAssetRow.batch_id == entity.id)) + session.add_all( + t.BatchAssetRow(batch_id=entity.id, asset_id=asset_id, position=position) + for position, asset_id in enumerate(entity.asset_ids) + ) + + +def _job_to_row(entity: AnnotationJob) -> t.Base: + return t.AnnotationJobRow(id=entity.id, task_group_id=entity.task_group_id, state=entity.state) + + +def _job_to_domain(session: Session, row: Any) -> AnnotationJob: + rows = session.execute( + select(t.AnnotationJobAssetRow.asset_id, t.AnnotationJobAssetRow.progress).where( + t.AnnotationJobAssetRow.job_id == row.id + ) + ).all() + return AnnotationJob( + id=row.id, + task_group_id=row.task_group_id, + state=row.state, + progress={asset_id: AssetProgress(progress) for asset_id, progress in rows}, + ) + + +def _job_sync_children(session: Session, entity: AnnotationJob) -> None: + session.execute( + delete(t.AnnotationJobAssetRow).where(t.AnnotationJobAssetRow.job_id == entity.id) + ) + session.add_all( + t.AnnotationJobAssetRow(job_id=entity.id, asset_id=asset_id, progress=progress) + for asset_id, progress in entity.progress.items() + ) + + +WORKSPACES = _flat_mapping(Workspace, t.WorkspaceRow, None) +PROJECTS = _flat_mapping(Project, t.ProjectRow, "workspace_id") +SOURCES = _flat_mapping(Source, t.SourceRow, "project_id") +INGEST_JOBS = _flat_mapping(IngestJob, t.IngestJobRow, "source_id") +ASSETS = _flat_mapping(Asset, t.AssetRow, "project_id") +TASK_GROUPS = _flat_mapping(TaskGroup, t.TaskGroupRow, "batch_id") +DATASETS = _flat_mapping(Dataset, t.DatasetRow, "project_id") +DATASET_MEMBERS = _flat_mapping(DatasetMember, t.DatasetMemberRow, "dataset_id") + +SCHEMAS: EntityMapping[AnnotationSchema] = EntityMapping( + row=t.AnnotationSchemaRow, + parent_column="project_id", + to_row=_schema_to_row, + to_domain=_schema_to_domain, +) +ANNOTATIONS: EntityMapping[Annotation] = EntityMapping( + row=t.AnnotationRow, + parent_column="asset_id", + to_row=_annotation_to_row, + to_domain=_annotation_to_domain, +) +RELEASES: EntityMapping[Release] = EntityMapping( + row=t.ReleaseRow, + parent_column="dataset_id", + to_row=_release_to_row, + to_domain=_release_to_domain, +) +DATASET_CHANGES: EntityMapping[DatasetChange] = EntityMapping( + row=t.DatasetChangeRow, + parent_column="dataset_id", + to_row=_change_to_row, + to_domain=_change_to_domain, +) +BATCHES: EntityMapping[Batch] = EntityMapping( + row=t.BatchRow, + parent_column="project_id", + to_row=_batch_to_row, + to_domain=_batch_to_domain, + sync_children=_batch_sync_children, +) +ANNOTATION_JOBS: EntityMapping[AnnotationJob] = EntityMapping( + row=t.AnnotationJobRow, + parent_column="task_group_id", + to_row=_job_to_row, + to_domain=_job_to_domain, + sync_children=_job_sync_children, +) diff --git a/src/visionset/kernel/adapters/_tables.py b/src/visionset/kernel/adapters/_tables.py new file mode 100644 index 00000000..e643e8d6 --- /dev/null +++ b/src/visionset/kernel/adapters/_tables.py @@ -0,0 +1,235 @@ +"""SQLAlchemy table definitions for the SQLite metadata store. + +Private on purpose: ``visionset.kernel.adapters`` exports only +``SqliteMetadataStore``, so no SQLAlchemy type ever reaches a domain or port +signature. Rows are translated to and from domain models in ``_mappers``. + +Storage decisions, and why (see ``docs/persistence.md`` for the long form): + +- Collections that are *relations* get child tables — ``batch_asset`` and + ``annotation_job_asset``. They are mutated one element at a time and queried + from the asset side, which a JSON blob cannot serve. +- Collections that are *immutable value objects* get JSON columns — + ``annotation_schema.classes``, ``annotation.geometry``, ``release.manifest``. + A schema version must rehydrate byte-identical, and nothing ever queries a + single ``LabelClass`` by name in SQL. +- Timestamps are TEXT holding an ISO-8601 string WITH its offset. SQLite's + DATETIME storage format drops the timezone, and a timestamp that silently + loses its offset is worse than no timestamp at all. + +``list()`` ordering is SQLite's implicit ``rowid``, i.e. insertion order. +""" + +from __future__ import annotations + +from typing import Any +from uuid import UUID + +from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint +from sqlalchemy import Uuid as SaUuid +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +from sqlalchemy.types import JSON, Float + +META_TABLE = "_visionset_meta" + + +class Base(DeclarativeBase): + """Declarative base for every VisionSet table.""" + + +class MetaRow(Base): + """The one-row schema ledger: which migration generation this file is at.""" + + __tablename__ = META_TABLE + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + format_version: Mapped[int] = mapped_column(Integer, nullable=False) + + +class WorkspaceRow(Base): + __tablename__ = "workspace" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + name: Mapped[str] = mapped_column(String, nullable=False) + root_dir: Mapped[str | None] = mapped_column(String, nullable=True) + + +class ProjectRow(Base): + __tablename__ = "project" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + workspace_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("workspace.id", ondelete="CASCADE"), index=True, nullable=False + ) + name: Mapped[str] = mapped_column(String, nullable=False) + description: Mapped[str | None] = mapped_column(String, nullable=True) + + +class AnnotationSchemaRow(Base): + __tablename__ = "annotation_schema" + __table_args__ = (UniqueConstraint("project_id", "version", name="uq_schema_project_version"),) + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + project_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("project.id", ondelete="CASCADE"), index=True, nullable=False + ) + version: Mapped[int] = mapped_column(Integer, nullable=False) + classes: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False) + + +class SourceRow(Base): + __tablename__ = "source" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + project_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("project.id", ondelete="CASCADE"), index=True, nullable=False + ) + kind: Mapped[str] = mapped_column(String, nullable=False) + uri: Mapped[str] = mapped_column(String, nullable=False) + + +class IngestJobRow(Base): + __tablename__ = "ingest_job" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + source_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("source.id", ondelete="CASCADE"), index=True, nullable=False + ) + state: Mapped[str] = mapped_column(String, nullable=False) + error: Mapped[str | None] = mapped_column(String, nullable=True) + + +class AssetRow(Base): + __tablename__ = "asset" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + project_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("project.id", ondelete="CASCADE"), index=True, nullable=False + ) + modality: Mapped[str] = mapped_column(String, nullable=False) + content_hash: Mapped[str] = mapped_column(String, index=True, nullable=False) + uri: Mapped[str] = mapped_column(String, nullable=False) + width: Mapped[int | None] = mapped_column(Integer, nullable=True) + height: Mapped[int | None] = mapped_column(Integer, nullable=True) + + +class BatchRow(Base): + __tablename__ = "batch" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + project_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("project.id", ondelete="CASCADE"), index=True, nullable=False + ) + name: Mapped[str] = mapped_column(String, nullable=False) + state: Mapped[str] = mapped_column(String, nullable=False) + + +class BatchAssetRow(Base): + """Batch membership. ``position`` preserves the order assets were added in.""" + + __tablename__ = "batch_asset" + + batch_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("batch.id", ondelete="CASCADE"), primary_key=True + ) + asset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("asset.id", ondelete="CASCADE"), primary_key=True + ) + position: Mapped[int] = mapped_column(Integer, nullable=False) + + +class TaskGroupRow(Base): + __tablename__ = "task_group" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + batch_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("batch.id", ondelete="CASCADE"), index=True, nullable=False + ) + name: Mapped[str] = mapped_column(String, nullable=False) + + +class AnnotationJobRow(Base): + __tablename__ = "annotation_job" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + task_group_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("task_group.id", ondelete="CASCADE"), index=True, nullable=False + ) + state: Mapped[str] = mapped_column(String, nullable=False) + + +class AnnotationJobAssetRow(Base): + """Per-asset annotation progress inside a job.""" + + __tablename__ = "annotation_job_asset" + + job_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("annotation_job.id", ondelete="CASCADE"), primary_key=True + ) + asset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("asset.id", ondelete="CASCADE"), primary_key=True + ) + progress: Mapped[str] = mapped_column(String, nullable=False) + + +class AnnotationRow(Base): + __tablename__ = "annotation" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + asset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("asset.id", ondelete="CASCADE"), index=True, nullable=False + ) + label_class: Mapped[str] = mapped_column(String, index=True, nullable=False) + schema_version: Mapped[int] = mapped_column(Integer, nullable=False) + geometry: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + provenance: Mapped[str] = mapped_column(String, nullable=False) + model_ref: Mapped[str | None] = mapped_column(String, nullable=True) + confidence: Mapped[float | None] = mapped_column(Float, nullable=True) + + +class DatasetRow(Base): + __tablename__ = "dataset" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + project_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("project.id", ondelete="CASCADE"), index=True, nullable=False + ) + name: Mapped[str] = mapped_column(String, nullable=False) + description: Mapped[str | None] = mapped_column(String, nullable=True) + + +class DatasetMemberRow(Base): + __tablename__ = "dataset_member" + __table_args__ = (UniqueConstraint("dataset_id", "asset_id", name="uq_member_dataset_asset"),) + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + dataset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("dataset.id", ondelete="CASCADE"), index=True, nullable=False + ) + asset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("asset.id", ondelete="CASCADE"), nullable=False + ) + + +class DatasetChangeRow(Base): + __tablename__ = "dataset_change" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + dataset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("dataset.id", ondelete="CASCADE"), index=True, nullable=False + ) + operation: Mapped[str] = mapped_column(String, nullable=False) + subject_ids: Mapped[list[str]] = mapped_column(JSON, nullable=False) + actor: Mapped[str | None] = mapped_column(String, nullable=True) + occurred_at: Mapped[str] = mapped_column(String, nullable=False) + + +class ReleaseRow(Base): + __tablename__ = "release" + + id: Mapped[UUID] = mapped_column(SaUuid, primary_key=True) + dataset_id: Mapped[UUID] = mapped_column( + SaUuid, ForeignKey("dataset.id", ondelete="CASCADE"), index=True, nullable=False + ) + tag: Mapped[str] = mapped_column(String, nullable=False) + manifest: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) diff --git a/src/visionset/kernel/adapters/migrations.py b/src/visionset/kernel/adapters/migrations.py new file mode 100644 index 00000000..422386a4 --- /dev/null +++ b/src/visionset/kernel/adapters/migrations.py @@ -0,0 +1,46 @@ +"""Forward-only schema migrations for the SQLite metadata store. + +The mechanism is deliberately small: an ordered list of migrations, and a +``format_version`` stored in the workspace itself saying how far that file has +been taken. Opening a workspace runs whatever is missing. There is no alembic +here — a local-first, single-file, single-writer store does not need a +migration framework, and ``format_version`` would then have to be kept in sync +with a second ledger by hand. + +**Adding a migration.** Append a ``Migration`` with the next version number and +an ``upgrade`` that takes a live connection. Do NOT edit an existing one — a +workspace already stamped at that version will never run it again. +``FORMAT_VERSION`` is derived from the list, so it cannot drift from reality. + +Migrations only run forward. A workspace stamped ahead of this build is +rejected (``WorkspaceFormatTooNew``) rather than silently downgraded. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass + +from sqlalchemy import Connection + +from visionset.kernel.adapters._tables import Base + + +@dataclass(frozen=True) +class Migration: + """One schema generation: what it is called and how to get there.""" + + version: int + name: str + upgrade: Callable[[Connection], None] + + +def _create_initial_schema(connection: Connection) -> None: + Base.metadata.create_all(connection) + + +MIGRATIONS: list[Migration] = [ + Migration(version=1, name="initial_schema", upgrade=_create_initial_schema), +] + +FORMAT_VERSION: int = MIGRATIONS[-1].version diff --git a/src/visionset/kernel/adapters/sqlite_metadata_store.py b/src/visionset/kernel/adapters/sqlite_metadata_store.py index 9014e340..8db0ad9c 100644 --- a/src/visionset/kernel/adapters/sqlite_metadata_store.py +++ b/src/visionset/kernel/adapters/sqlite_metadata_store.py @@ -1,29 +1,190 @@ """Default MetadataStore adapter: SQLite via SQLAlchemy. -Only engine setup and (empty) schema creation exist today; table models land -in a later session together with the entity-level operations on the port. +One workspace is one SQLite file. The store owns the engine, the schema +generation (``format_version``), and a repository per entity type; the +row/model translation lives in ``_mappers`` so that nothing SQLAlchemy-shaped +escapes this package. """ from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path +from typing import Any +from uuid import UUID -from sqlalchemy import Engine, MetaData, create_engine +from sqlalchemy import ( + Connection, + Engine, + create_engine, + delete, + event, + insert, + inspect, + select, + text, +) +from sqlalchemy.orm import Session + +from visionset.kernel.adapters import _mappers as m +from visionset.kernel.adapters._tables import META_TABLE, MetaRow +from visionset.kernel.adapters.migrations import FORMAT_VERSION, MIGRATIONS +from visionset.kernel.errors import EntityAlreadyExists, EntityNotFound, WorkspaceFormatTooNew +from visionset.kernel.ports.metadata_store import UnitOfWork + +#: ``format_version`` reported by a database whose schema has not been created yet. +UNINITIALIZED = 0 + + +def _enable_foreign_keys(dbapi_connection: Any, _: Any) -> None: + """SQLite ships with foreign keys OFF, per connection. + + Without this every ``ForeignKey`` in ``_tables`` is decorative: orphan rows + would insert happily and cascades would never fire. + """ + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA foreign_keys = ON") + cursor.close() + + +def _stored_format_version(connection: Connection) -> int | None: + """The version stamped in this file, or None if it was never initialized.""" + if not inspect(connection).has_table(META_TABLE): + return None + return connection.execute(select(MetaRow.format_version)).scalar_one_or_none() + + +def _stamp(connection: Connection, version: int) -> None: + connection.execute(delete(MetaRow)) + connection.execute(insert(MetaRow).values(id=1, format_version=version)) + + +class SqlRepository[T: m.Entity]: + """Generic repository driven by one ``EntityMapping``. + + Fourteen entity types share this implementation because they share a shape: + a UUID primary key and at most one parent. Anything that needs more than + that is a query a service should express, not a method the port should grow. + """ + + def __init__(self, session: Session, mapping: m.EntityMapping[T]) -> None: + self._session = session + self._mapping = mapping + + def _row(self, entity_id: UUID) -> Any: + return self._session.get(self._mapping.row, entity_id) + + def _sync_children(self, entity: T) -> None: + if self._mapping.sync_children is not None: + self._mapping.sync_children(self._session, entity) + + def add(self, entity: T) -> T: + if self._row(entity.id) is not None: + raise EntityAlreadyExists( + f"{self._mapping.row.__tablename__} {entity.id} already exists" + ) + self._session.add(self._mapping.to_row(entity)) + self._sync_children(entity) + self._session.flush() + return entity + + def update(self, entity: T) -> T: + if self._row(entity.id) is None: + raise EntityNotFound(f"no {self._mapping.row.__tablename__} with id {entity.id}") + self._session.merge(self._mapping.to_row(entity)) + self._sync_children(entity) + self._session.flush() + return entity + + def get(self, entity_id: UUID) -> T | None: + row = self._row(entity_id) + return None if row is None else self._mapping.to_domain(self._session, row) + + def list(self, parent_id: UUID | None = None) -> list[T]: + parent_column = self._mapping.parent_column + if parent_id is not None and parent_column is None: + raise ValueError( + f"{self._mapping.row.__tablename__} is a root entity: it has no parent" + ) + statement = select(self._mapping.row) + if parent_id is not None and parent_column is not None: + statement = statement.where(getattr(self._mapping.row, parent_column) == parent_id) + rows = list(self._session.scalars(statement.order_by(text("rowid")))) + return [self._mapping.to_domain(self._session, row) for row in rows] + + def delete(self, entity_id: UUID) -> bool: + row = self._row(entity_id) + if row is None: + return False + self._session.delete(row) + self._session.flush() + return True + + +class SqlUnitOfWork: + """The repositories of one transaction, all sharing a single session.""" + + def __init__(self, session: Session) -> None: + self.workspaces = SqlRepository(session, m.WORKSPACES) + self.projects = SqlRepository(session, m.PROJECTS) + self.schemas = SqlRepository(session, m.SCHEMAS) + self.sources = SqlRepository(session, m.SOURCES) + self.ingest_jobs = SqlRepository(session, m.INGEST_JOBS) + self.assets = SqlRepository(session, m.ASSETS) + self.batches = SqlRepository(session, m.BATCHES) + self.task_groups = SqlRepository(session, m.TASK_GROUPS) + self.annotation_jobs = SqlRepository(session, m.ANNOTATION_JOBS) + self.annotations = SqlRepository(session, m.ANNOTATIONS) + self.datasets = SqlRepository(session, m.DATASETS) + self.dataset_members = SqlRepository(session, m.DATASET_MEMBERS) + self.dataset_changes = SqlRepository(session, m.DATASET_CHANGES) + self.releases = SqlRepository(session, m.RELEASES) class SqliteMetadataStore: def __init__(self, db_path: Path) -> None: db_path.parent.mkdir(parents=True, exist_ok=True) self._engine: Engine = create_engine(f"sqlite:///{db_path}") - self._metadata = MetaData() + event.listen(self._engine, "connect", _enable_foreign_keys) @property def engine(self) -> Engine: return self._engine + @property + def format_version(self) -> int: + """The stamped schema generation, or ``UNINITIALIZED`` before creation.""" + with self._engine.connect() as connection: + stored = _stored_format_version(connection) + return UNINITIALIZED if stored is None else stored + def initialize(self) -> None: - """Create the storage schema if it does not exist. Idempotent.""" - self._metadata.create_all(self._engine) + """Create or migrate the storage schema. Idempotent. + + A fresh file gets every migration and is stamped at ``FORMAT_VERSION``; + an existing one gets only what it is missing. A file stamped ahead of + this build raises rather than being opened on a guess. + """ + with self._engine.begin() as connection: + stored = _stored_format_version(connection) + if stored is not None and stored > FORMAT_VERSION: + raise WorkspaceFormatTooNew( + f"workspace format_version {stored} is newer than this VisionSet " + f"understands (max {FORMAT_VERSION}); upgrade VisionSet to open it" + ) + pending = [mig for mig in MIGRATIONS if stored is None or mig.version > stored] + if not pending: + return + for migration in pending: + migration.upgrade(connection) + _stamp(connection, FORMAT_VERSION) + + @contextmanager + def unit_of_work(self) -> Iterator[UnitOfWork]: + """One transaction: commits on clean exit, rolls back on any exception.""" + with Session(self._engine) as session, session.begin(): + yield SqlUnitOfWork(session) def close(self) -> None: self._engine.dispose() diff --git a/src/visionset/kernel/domain/__init__.py b/src/visionset/kernel/domain/__init__.py index 6d2c039c..a912bdb2 100644 --- a/src/visionset/kernel/domain/__init__.py +++ b/src/visionset/kernel/domain/__init__.py @@ -10,7 +10,7 @@ from visionset.kernel.domain.annotation import Annotation, Provenance from visionset.kernel.domain.asset import Asset from visionset.kernel.domain.batch import Batch, BatchState -from visionset.kernel.domain.dataset import Dataset +from visionset.kernel.domain.dataset import Dataset, DatasetChange, DatasetMember from visionset.kernel.domain.geometry import ( BboxGeometry, ClassificationGeometry, @@ -38,6 +38,8 @@ "BboxGeometry", "ClassificationGeometry", "Dataset", + "DatasetChange", + "DatasetMember", "Geometry", "GeometryType", "IngestJob", diff --git a/src/visionset/kernel/domain/dataset.py b/src/visionset/kernel/domain/dataset.py index 70139c82..2b5754b7 100644 --- a/src/visionset/kernel/domain/dataset.py +++ b/src/visionset/kernel/domain/dataset.py @@ -1,9 +1,10 @@ -# usage: from visionset.kernel.domain import Dataset +# usage: from visionset.kernel.domain import Dataset, DatasetMember, DatasetChange from __future__ import annotations +from datetime import UTC, datetime from uuid import UUID, uuid4 -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator class Dataset(BaseModel): @@ -13,3 +14,46 @@ class Dataset(BaseModel): project_id: UUID name: str description: str | None = None + + +class DatasetMember(BaseModel): + """One asset's membership in a Dataset. + + Membership is a row rather than a list on ``Dataset`` because it is the thing + promotion adds to and curation removes from, one asset at a time. It carries + its own ``id`` so it addresses like every other entity, and + ``(dataset_id, asset_id)`` is unique. The rules that govern promotion and + removal belong to DatasetService, not here. + """ + + id: UUID = Field(default_factory=uuid4) + dataset_id: UUID + asset_id: UUID + + +class DatasetChange(BaseModel): + """One append-only entry in a Dataset's mutation log. + + Every mutation of the curated trunk is recorded — this is the base of + reproducibility, and later of enterprise auditing. Entries are NEVER updated + or deleted; the append-only discipline is DatasetService's to enforce. + + ``occurred_at`` is timezone-aware UTC, and that is the convention for every + timestamp in the domain: a naive datetime is rejected outright rather than + read as local time and silently misfiled once it crosses a machine boundary. + ``actor`` is a placeholder until identities exist. + """ + + id: UUID = Field(default_factory=uuid4) + dataset_id: UUID + operation: str + subject_ids: list[UUID] = Field(default_factory=list) + actor: str | None = None + occurred_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + + @field_validator("occurred_at") + @classmethod + def _occurred_at_is_timezone_aware(cls, value: datetime) -> datetime: + if value.tzinfo is None: + raise ValueError("occurred_at must be timezone-aware (UTC)") + return value.astimezone(UTC) diff --git a/src/visionset/kernel/errors.py b/src/visionset/kernel/errors.py new file mode 100644 index 00000000..2e66fefb --- /dev/null +++ b/src/visionset/kernel/errors.py @@ -0,0 +1,33 @@ +# usage: from visionset.kernel import EntityNotFound, VisionSetError +"""Domain errors raised by the kernel. + +Every error the kernel raises derives from ``VisionSetError``, so a delivery +surface can translate the whole family to an HTTP status or an exit code with a +single ``except`` clause. The kernel NEVER raises a framework exception — +``HTTPException`` and friends belong to the boundary, not here. + +Only the errors the persistence layer needs live here today; services add their +own (``ProjectNotFound``, ``InvalidTransition``, ...) as they land. +""" + +from __future__ import annotations + + +class VisionSetError(Exception): + """Base class for every error the kernel raises.""" + + +class EntityNotFound(VisionSetError): + """An operation addressed an entity id that is not in the store.""" + + +class EntityAlreadyExists(VisionSetError): + """An insert collided with an existing primary key.""" + + +class WorkspaceFormatTooNew(VisionSetError): + """The stored ``format_version`` is newer than this VisionSet understands. + + Migrations only ever run forward, so a workspace written by a later version + is unreadable rather than silently downgraded. + """ diff --git a/src/visionset/kernel/ports/__init__.py b/src/visionset/kernel/ports/__init__.py index 9c5e1cd3..afd1e09d 100644 --- a/src/visionset/kernel/ports/__init__.py +++ b/src/visionset/kernel/ports/__init__.py @@ -10,7 +10,7 @@ from visionset.kernel.ports.exporter import Exporter from visionset.kernel.ports.importer import Importer from visionset.kernel.ports.media_processor import MediaProcessor -from visionset.kernel.ports.metadata_store import MetadataStore +from visionset.kernel.ports.metadata_store import MetadataStore, Repository, UnitOfWork from visionset.kernel.ports.model_provider import ModelProvider __all__ = [ @@ -23,4 +23,6 @@ "MediaProcessor", "MetadataStore", "ModelProvider", + "Repository", + "UnitOfWork", ] diff --git a/src/visionset/kernel/ports/metadata_store.py b/src/visionset/kernel/ports/metadata_store.py index da136617..5097b854 100644 --- a/src/visionset/kernel/ports/metadata_store.py +++ b/src/visionset/kernel/ports/metadata_store.py @@ -1,16 +1,149 @@ +"""Persistence port: repositories, a unit of work, and the store that owns them. + +Nothing here knows about SQL. The port speaks in domain models and UUIDs only — +translating those into rows is the adapter's job, which is exactly why a service +written against this file cannot accidentally depend on SQLite. +""" + +from __future__ import annotations + +from contextlib import AbstractContextManager from typing import Protocol, runtime_checkable +from uuid import UUID + +from visionset.kernel.domain import ( + Annotation, + AnnotationJob, + AnnotationSchema, + Asset, + Batch, + Dataset, + DatasetChange, + DatasetMember, + IngestJob, + Project, + Release, + Source, + TaskGroup, + Workspace, +) + + +class Repository[T](Protocol): + """Storage for one entity type, addressed by UUID. + + Every entity in the domain has at most one parent — a Project belongs to a + Workspace, an Annotation to an Asset — so a single ``parent_id`` filter + covers every scoped read the services need, and no query language leaks into + the port. + + ``add`` and ``update`` are deliberately separate rather than one upsert: a + service that inserts a duplicate, or updates something that was deleted, has + a bug and should hear about it instead of silently overwriting. + """ + + def add(self, entity: T) -> T: + """Insert. Raises ``EntityAlreadyExists`` if the id is already stored.""" + ... + + def update(self, entity: T) -> T: + """Replace by id. Raises ``EntityNotFound`` if the id is not stored.""" + ... + + def get(self, entity_id: UUID) -> T | None: ... + + def list(self, parent_id: UUID | None = None) -> list[T]: + """All entities, or only those under ``parent_id``. + + Ordering is insertion order. Passing a ``parent_id`` for a root entity + that has no parent raises ``ValueError``. + """ + ... + + def delete(self, entity_id: UUID) -> bool: + """Remove by id; returns False if it was not there.""" + ... + + +@runtime_checkable +class UnitOfWork(Protocol): + """One transaction, with a repository per entity type. + + The scope of a unit of work is one operation on a Project aggregate: open + it, do the whole operation, let it close. Everything inside commits together + or not at all — a batch approval that partitions into jobs must never leave + half its jobs behind. + + The repositories are read-only properties rather than attributes because a + mutable protocol attribute is invariant: an adapter would then have to hand + back exactly ``Repository[Project]`` and never its own implementation of it. + """ + + @property + def workspaces(self) -> Repository[Workspace]: ... + + @property + def projects(self) -> Repository[Project]: ... + + @property + def schemas(self) -> Repository[AnnotationSchema]: ... + + @property + def sources(self) -> Repository[Source]: ... + + @property + def ingest_jobs(self) -> Repository[IngestJob]: ... + + @property + def assets(self) -> Repository[Asset]: ... + + @property + def batches(self) -> Repository[Batch]: ... + + @property + def task_groups(self) -> Repository[TaskGroup]: ... + + @property + def annotation_jobs(self) -> Repository[AnnotationJob]: ... + + @property + def annotations(self) -> Repository[Annotation]: ... + + @property + def datasets(self) -> Repository[Dataset]: ... + + @property + def dataset_members(self) -> Repository[DatasetMember]: ... + + @property + def dataset_changes(self) -> Repository[DatasetChange]: ... + + @property + def releases(self) -> Repository[Release]: ... @runtime_checkable class MetadataStore(Protocol): """Persistence for domain entities (projects, assets, annotations, ...). - Only lifecycle management is defined today; entity-level operations land - together with the table models in a later session. + ``format_version`` is the stored schema generation. It is what makes a + workspace readable — or knowably unreadable — by a different VisionSet + build, and it is checked on every open, not only at creation. """ + @property + def format_version(self) -> int: ... + def initialize(self) -> None: - """Create the storage schema if it does not exist. Idempotent.""" + """Create or migrate the storage schema. Idempotent. + + Raises ``WorkspaceFormatTooNew`` if the stored ``format_version`` is + ahead of this build: migrations only ever run forward. + """ + ... + + def unit_of_work(self) -> AbstractContextManager[UnitOfWork]: + """Open a transaction; commit on clean exit, roll back on any exception.""" ... def close(self) -> None: ... diff --git a/tests/kernel/test_metadata_store.py b/tests/kernel/test_metadata_store.py index 2d2372be..4a31ee6c 100644 --- a/tests/kernel/test_metadata_store.py +++ b/tests/kernel/test_metadata_store.py @@ -1,9 +1,121 @@ +from datetime import UTC, datetime from pathlib import Path +from uuid import UUID, uuid4 +import pytest +from pydantic import ValidationError from sqlalchemy import text +from sqlalchemy.exc import IntegrityError +from visionset.kernel import EntityAlreadyExists, EntityNotFound, WorkspaceFormatTooNew from visionset.kernel.adapters import SqliteMetadataStore -from visionset.kernel.ports import MetadataStore +from visionset.kernel.domain import ( + Annotation, + AnnotationJob, + AnnotationSchema, + Asset, + AssetProgress, + Attribute, + Batch, + BboxGeometry, + ClassificationGeometry, + Dataset, + DatasetChange, + DatasetMember, + GeometryType, + IngestJob, + LabelClass, + Manifest, + PolygonGeometry, + Project, + Release, + Source, + TaskGroup, + Workspace, +) +from visionset.kernel.ports import MetadataStore, UnitOfWork + + +def _store(tmp_path: Path, name: str = "visionset.db") -> SqliteMetadataStore: + store = SqliteMetadataStore(tmp_path / name) + store.initialize() + return store + + +def _seed(uow: UnitOfWork) -> list[tuple[str, UUID]]: + """Persist one of every entity, wired into a valid parent chain.""" + workspace = uow.workspaces.add(Workspace(name="w", root_dir="/tmp/w")) + project = uow.projects.add(Project(workspace_id=workspace.id, name="p", description="d")) + source = uow.sources.add(Source(project_id=project.id, uri="file:///images")) + ingest = uow.ingest_jobs.add(IngestJob(source_id=source.id)) + first = uow.assets.add( + Asset(project_id=project.id, content_hash="a" * 64, uri="file:///1.png", width=8, height=6) + ) + second = uow.assets.add( + Asset(project_id=project.id, content_hash="b" * 64, uri="file:///2.png") + ) + schema = uow.schemas.add( + AnnotationSchema( + project_id=project.id, + version=1, + classes=[ + LabelClass( + name="car", + geometry=GeometryType.BBOX, + color="#ff0000", + attributes=[Attribute(name="occluded", kind="boolean", required=True)], + ) + ], + ) + ) + batch = uow.batches.add(Batch(project_id=project.id, name="b", asset_ids=[second.id, first.id])) + group = uow.task_groups.add(TaskGroup(batch_id=batch.id, name="tg")) + job = uow.annotation_jobs.add( + AnnotationJob( + task_group_id=group.id, + progress={first.id: AssetProgress.ANNOTATED, second.id: AssetProgress.SKIPPED}, + ) + ) + annotation = uow.annotations.add( + Annotation( + asset_id=first.id, + label_class="car", + schema_version=1, + geometry=BboxGeometry(x=1.0, y=2.0, width=10.0, height=20.0), + provenance="model", + model_ref="yolo:v8", + confidence=0.5, + ) + ) + dataset = uow.datasets.add(Dataset(project_id=project.id, name="ds")) + member = uow.dataset_members.add(DatasetMember(dataset_id=dataset.id, asset_id=first.id)) + change = uow.dataset_changes.add( + DatasetChange(dataset_id=dataset.id, operation="promote", subject_ids=[first.id]) + ) + release = uow.releases.add( + Release( + dataset_id=dataset.id, + tag="v1", + manifest=Manifest(schema_version=1, asset_count=1, content_hashes=["a" * 64]), + ) + ) + return [ + ("workspaces", workspace.id), + ("projects", project.id), + ("sources", source.id), + ("ingest_jobs", ingest.id), + ("assets", first.id), + ("assets", second.id), + ("schemas", schema.id), + ("batches", batch.id), + ("task_groups", group.id), + ("annotation_jobs", job.id), + ("annotations", annotation.id), + ("datasets", dataset.id), + ("dataset_members", member.id), + ("dataset_changes", change.id), + ("releases", release.id), + ] def test_initialize_creates_database_file(tmp_path: Path) -> None: @@ -25,3 +137,268 @@ def test_initialize_is_idempotent(tmp_path: Path) -> None: def test_satisfies_metadata_store_port(tmp_path: Path) -> None: assert isinstance(SqliteMetadataStore(tmp_path / "visionset.db"), MetadataStore) + + +def test_unit_of_work_satisfies_the_port(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + assert isinstance(uow, UnitOfWork) + store.close() + + +def test_format_version_is_zero_until_initialized(tmp_path: Path) -> None: + store = SqliteMetadataStore(tmp_path / "visionset.db") + assert store.format_version == 0 + store.initialize() + assert store.format_version == 1 + store.close() + + +def test_a_reopened_workspace_reports_the_same_format_version(tmp_path: Path) -> None: + _store(tmp_path).close() + reopened = SqliteMetadataStore(tmp_path / "visionset.db") + reopened.initialize() + assert reopened.format_version == 1 + reopened.close() + + +def test_a_workspace_from_the_future_is_refused(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.engine.begin() as conn: + conn.execute(text("update _visionset_meta set format_version = 99")) + store.close() + + reopened = SqliteMetadataStore(tmp_path / "visionset.db") + with pytest.raises(WorkspaceFormatTooNew, match="99"): + reopened.initialize() + reopened.close() + + +def test_every_entity_round_trips_through_a_reopened_store(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + seeded = _seed(uow) + written = {key: getattr(uow, name).get(key) for name, key in seeded} + store.close() + + reopened = SqliteMetadataStore(tmp_path / "visionset.db") + reopened.initialize() + with reopened.unit_of_work() as uow: + for name, key in seeded: + assert getattr(uow, name).get(key) == written[key] + reopened.close() + + +def test_annotation_round_trips_every_geometry_variant(tmp_path: Path) -> None: + store = _store(tmp_path) + geometries = [ + BboxGeometry(x=1.0, y=2.0, width=10.0, height=20.0), + PolygonGeometry(points=[(0.0, 0.0), (10.0, 0.0), (10.0, 10.0)]), + ClassificationGeometry(), + ] + with store.unit_of_work() as uow: + asset_id = _seed(uow)[4][1] + for geometry in geometries: + written = uow.annotations.add( + Annotation( + asset_id=asset_id, + label_class="car", + schema_version=1, + geometry=geometry, + provenance="human", + ) + ) + read = uow.annotations.get(written.id) + assert read is not None + assert read == written + assert type(read.geometry) is type(geometry) + assert isinstance(read.geometry.type, GeometryType) + store.close() + + +def test_batch_membership_keeps_the_order_it_was_written_in(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + seeded = _seed(uow) + first, second, batch_id = seeded[4][1], seeded[5][1], seeded[7][1] + stored = uow.batches.get(batch_id) + assert stored is not None + assert stored.asset_ids == [second, first] + + stored.asset_ids = [first, second] + uow.batches.update(stored) + reread = uow.batches.get(batch_id) + assert reread is not None + assert reread.asset_ids == [first, second] + store.close() + + +def test_annotation_job_progress_round_trips_keyed_by_asset(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + seeded = _seed(uow) + first, second = seeded[4][1], seeded[5][1] + job = uow.annotation_jobs.get(seeded[9][1]) + assert job is not None + assert job.progress == {first: AssetProgress.ANNOTATED, second: AssetProgress.SKIPPED} + assert all(isinstance(key, UUID) for key in job.progress) + store.close() + + +def test_schema_classes_and_attributes_round_trip(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + schema = uow.schemas.get(_seed(uow)[6][1]) + assert schema is not None + label_class = schema.classes[0] + assert label_class.geometry is GeometryType.BBOX + assert label_class.attributes[0] == Attribute( + name="occluded", kind="boolean", required=True + ) + store.close() + + +def test_release_manifest_round_trips(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + release = uow.releases.get(_seed(uow)[14][1]) + assert release is not None + assert release.manifest == Manifest( + schema_version=1, asset_count=1, content_hashes=["a" * 64] + ) + store.close() + + +def test_dataset_change_timestamp_stays_utc_aware(tmp_path: Path) -> None: + store = _store(tmp_path) + written_at = datetime(2026, 7, 26, 12, 30, tzinfo=UTC) + with store.unit_of_work() as uow: + change = uow.dataset_changes.add( + DatasetChange( + dataset_id=_seed(uow)[11][1], + operation="remove_asset", + actor="anaya", + occurred_at=written_at, + ) + ) + store.close() + + reopened = SqliteMetadataStore(tmp_path / "visionset.db") + reopened.initialize() + with reopened.unit_of_work() as uow: + stored = uow.dataset_changes.get(change.id) + assert stored is not None + assert stored.occurred_at == written_at + assert stored.occurred_at.tzinfo is not None + reopened.close() + + +def test_a_naive_timestamp_is_rejected() -> None: + with pytest.raises(ValidationError, match="timezone-aware"): + DatasetChange(dataset_id=uuid4(), operation="promote", occurred_at=datetime(2026, 7, 26)) + + +def test_unit_of_work_rolls_back_on_error(tmp_path: Path) -> None: + store = _store(tmp_path) + workspace = Workspace(name="doomed") + with pytest.raises(RuntimeError, match="boom"), store.unit_of_work() as uow: + uow.workspaces.add(workspace) + raise RuntimeError("boom") + + with store.unit_of_work() as uow: + assert uow.workspaces.get(workspace.id) is None + store.close() + + +def test_unit_of_work_commits_on_clean_exit(tmp_path: Path) -> None: + store = _store(tmp_path) + workspace = Workspace(name="kept") + with store.unit_of_work() as uow: + uow.workspaces.add(workspace) + store.close() + + reopened = SqliteMetadataStore(tmp_path / "visionset.db") + reopened.initialize() + with reopened.unit_of_work() as uow: + assert uow.workspaces.get(workspace.id) == workspace + reopened.close() + + +def test_add_rejects_a_duplicate_id(tmp_path: Path) -> None: + store = _store(tmp_path) + workspace = Workspace(name="w") + with store.unit_of_work() as uow: + uow.workspaces.add(workspace) + with pytest.raises(EntityAlreadyExists, match=str(workspace.id)): + uow.workspaces.add(workspace) + store.close() + + +def test_update_requires_an_existing_entity(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow, pytest.raises(EntityNotFound, match="workspace"): + uow.workspaces.update(Workspace(name="ghost")) + store.close() + + +def test_update_replaces_the_stored_fields(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + workspace = uow.workspaces.add(Workspace(name="before")) + workspace.name = "after" + uow.workspaces.update(workspace) + stored = uow.workspaces.get(workspace.id) + assert stored is not None + assert stored.name == "after" + store.close() + + +def test_delete_reports_whether_anything_was_removed(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + workspace = uow.workspaces.add(Workspace(name="w")) + assert uow.workspaces.delete(workspace.id) is True + assert uow.workspaces.get(workspace.id) is None + assert uow.workspaces.delete(workspace.id) is False + assert uow.workspaces.delete(uuid4()) is False + store.close() + + +def test_list_is_scoped_to_the_parent(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + workspace = uow.workspaces.add(Workspace(name="w")) + other = uow.workspaces.add(Workspace(name="other")) + mine = uow.projects.add(Project(workspace_id=workspace.id, name="mine")) + uow.projects.add(Project(workspace_id=other.id, name="theirs")) + + assert [project.id for project in uow.projects.list(workspace.id)] == [mine.id] + assert len(uow.projects.list()) == 2 + store.close() + + +def test_list_rejects_a_parent_id_for_a_root_entity(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + uow.workspaces.list() + with pytest.raises(ValueError, match="root entity"): + uow.workspaces.list(uuid4()) + store.close() + + +def test_foreign_keys_are_enforced(tmp_path: Path) -> None: + store = _store(tmp_path) + with pytest.raises(IntegrityError, match="FOREIGN KEY"), store.unit_of_work() as uow: + uow.projects.add(Project(workspace_id=uuid4(), name="orphan")) + store.close() + + +def test_deleting_a_parent_cascades_to_its_children(tmp_path: Path) -> None: + store = _store(tmp_path) + with store.unit_of_work() as uow: + project_id = _seed(uow)[1][1] + uow.projects.delete(project_id) + assert uow.assets.list(project_id) == [] + assert uow.batches.list(project_id) == [] + assert uow.datasets.list(project_id) == [] + store.close() diff --git a/tests/kernel/test_migrations.py b/tests/kernel/test_migrations.py new file mode 100644 index 00000000..d3f5ddb6 --- /dev/null +++ b/tests/kernel/test_migrations.py @@ -0,0 +1,15 @@ +from visionset.kernel.adapters.migrations import FORMAT_VERSION, MIGRATIONS + + +def test_format_version_is_derived_from_the_last_migration() -> None: + assert MIGRATIONS[-1].version == FORMAT_VERSION + + +def test_migration_versions_are_unique_and_start_at_one() -> None: + versions = [migration.version for migration in MIGRATIONS] + assert versions == list(range(1, len(versions) + 1)) + + +def test_every_migration_is_named() -> None: + for migration in MIGRATIONS: + assert migration.name