diff --git a/AGENTS.md b/AGENTS.md index 7ca78f47..93b08f5d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -13,7 +13,7 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA - repository: persistence, reads and writes, its own domain's tables only - provider: anything outside the process (tools, forges, feeds, ledger, identity) - **Domain boundaries.** A service never touches another domain's repositories; it calls that domain's service. Several repositories of your own domain in one service are fine — the line is whose repositories, not how many. Repositories never join across domains. Callers compose across domains by typed ID; composite reads are their own query object (`RepoStateQuery`). -- **The database owns identity and time.** Every record has `id` (UUID), `created_at`, `updated_at`, generated by the database on insert (`RETURNING`). `id is None` = not saved yet; `add` returns the saved record. The app never mints IDs or timestamps. IDs are plain `UUID`, named `_id` (fields and parameters, matching the DDL's `_id`); never a `str` ID, never a path as an ID, never a path in a signature. Don't re-declare `id` or add your own created/updated timestamps on an `Entity`. +- **The database owns identity and time.** Every record has `id` (UUIDv4), `created_at`, `updated_at`, generated by the database on insert (`RETURNING`); on update the contracts `
_touch` trigger keeps `updated_at` — the app never sets it (no SQLAlchemy `onupdate`). On SQLite, `uuid()` is registered by `create_database_engine`. Append-only tables (ledger events) stop at `id` (UUIDv4) and `created_at` — rows are never updated, so there is no `updated_at`. `id is None` = not saved yet; `add` returns the saved record. The app never mints IDs or timestamps. IDs are plain `UUID`, named `_id` (fields and parameters, matching the DDL's `
_id`); never a `str` ID, never a path as an ID, never a path in a signature. Don't re-declare `id` or add your own created/updated timestamps on an `Entity` or `AppendOnly`. Order comes from `created_at`, never from the id. - **Same shapes everywhere.** Repositories: `get`, `find_*`, `list`, `add`, `update` (`add`/`update` return the saved record). Skill doors: `prepare(...) -> Input` and `record(doc, provenance)` (`@overload` per document). Rollups: `summary(filter, group_by)`; series: `timeline(filter, bucket)`. Deterministic document checks: `Lint`. - **Services return domain results.** `JobResult` exists only at the engine/CLI adapter boundary. - **One code path.** Solo (`traust …`, in-process), small group (`traust serve --with-engine`, one process) and scale (REST API, engine and runners as separate processes) run the same services; only the transport (in-process or REST) and the storage (SQLite + dir, or Postgres + object storage) differ. Never add a mode-specific branch of domain logic; the REST API holds none. CI runs service tests in-process and through the REST adapter, plus one end-to-end skill on SQLite with no server @@ -46,7 +46,7 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA - `errors` → `contracts` → `models` → `providers` → `repositories` → `rendering` → `services` → `context`; each imports only earlier ones (`tests/v1/test_layering.py`) - One question per folder: `models` what it is and the rules about its own state, `providers` how we reach outside (tools, sources, feeds, process, clock), `repositories` how it is stored, `services` what we do with it (use cases + pure policies), `context` how it is wired - Organised by role, then subject: `models/.py`, `repositories/.py`, `services/.py`; a subject becomes a package only when it needs several modules (`services/findings/`) -- `models/base.py`: `Model` (strict, internal), `Dto` (tolerant, boundary data), `Entity` (`id`, `created_at`, `updated_at` — `None` until the database saves it) +- `models/base.py`: `Model` (strict, internal), `Dto` (tolerant, boundary data), `Entity` (`id`, `created_at`, `updated_at` — `None` until the database saves it), `AppendOnly` (`id`, `created_at` — rows are never updated) - Tests mirror `src`: `tests/v1//test_.py` - Pack policy (routing rules, cadences, lanes, skills, CLIs) stays in traust - traust-core is the hub: every other traust repo depends on it, it depends on none of them. One exception: the traust-ledger client, imported only inside the ledger provider adapter (behind `LedgerClient`). traust-contracts is data, read through `traust_core.v1.contracts` diff --git a/README.md b/README.md index bea931c9..ca9936dc 100644 --- a/README.md +++ b/README.md @@ -8,12 +8,12 @@ All modules live under `traust_core.v1` (see [Versioning](#versioning)). |---|---|---|---| | **Contracts** | `contracts` | the pinned traust-contracts dependency, read as data: `validate(name, doc)` (exact JSON Schema gate), `schema()`, `enum_values()`, `normalize_enum(name, value)` (registry read view), `ddl()`, `storage_profile()` | skills remembering to run `reporting validate` | | **Errors** | `errors` | the exception hierarchy every layer raises (`ValidationError`, `NotFoundError`, …) | ad hoc exceptions | -| **Models** | `models` | what things are and the rules about their own state: `base` (`Model` strict, `Dto` tolerant, `Entity` with a UUID id), `values` (`HttpsRepoUrl`, `GitSha`, `CveId`), `integrity`, `storage`, `operations` (`JobResult`, `AssetRequest`, `Order`), and one module per subject (`findings`, `portfolio`, …) | dicts passed between scripts | +| **Models** | `models` | what things are and the rules about their own state: `base` (`Model` strict, `Dto` tolerant, `Entity` with a database-generated UUIDv4 id, `AppendOnly` for tables that are never updated), `values` (`HttpsRepoUrl`, `GitSha`, `CveId`), `integrity`, `storage`, `operations` (`JobResult`, `AssetRequest`, `Order`), and one module per subject (`findings`, `portfolio`, …) | dicts passed between scripts | | **Artifacts** | `models.artifacts` | the `Artifact` base + `SchemaView` (fields from the installed schema, `doc.findings[0].verdict`, unknown names raise) and the named documents (`TriageArtifact`, `SecurityAuditArtifact`, …, `Verdict`, `Severity`) | `json.load` + `.get("key")` | | **Artifact publishing** | `services.artifact_publishing` | `ctx.artifact_publisher(specs).publish(name, subject, raw, run_id)` (LLM) or `.publish_artifact(artifact, subject, run_id)` (code): exact gate → files (always) → index (when `database.url` is set) | writing `analysis-results` files by hand | | **Analysis results** | `repositories.analysis_results` | `ctx.analysis_results().read(TriageArtifact, subject)` / `.read_all(TriageArtifact, tree)`; raw `put/get/find(subject, kind)`; today's naming convention in one place | `json.load` + `"-security-audit.json"` path building | | **Storage records** | `repositories.storage`, `services.storage` | `record_artifact()` / `get_binding()` over contracts storage/v1 tables | contracts `Store.ingest/get_binding` | -| **Records (pattern)** | `repositories.sql` | `SqlUnitOfWork` + one `SqlRepository` per aggregate on SQLAlchemy Core; sqlite or Postgres by URL | `sqlite3.connect`, inline SQL | +| **Records (pattern)** | `repositories.sql` | `SqlUnitOfWork` + one `SqlRepository` per aggregate on SQLAlchemy Core; sqlite or Postgres by URL. `record_columns()` (mutable: UUIDv4 id + `created_at` + `updated_at`) / `append_only_columns()` (UUIDv4 id + `created_at`, no `updated_at`) + `_insert_record`/`_update_record`, database-generated via `RETURNING`; `updated_at` kept by the contracts touch trigger, SQLite `uuid()` registered by `create_database_engine` | `sqlite3.connect`, inline SQL, client-minted ids/timestamps | | **Object store** | `repositories.object_store` | `ObjectStore.put/get/list` raw bytes by key, digest-verified (local dir; S3 later) | `write_text`, `json.dump` | | **Rendering** | `rendering` | `Report` → `MarkdownRenderer` → `publish()`; `markdown_table()` | 18 copies of `render_md` | | **Jobs** | `models.operations`, `services.operations` | `Materializer.materialize(AssetRequest) -> JobResult` (succeeded · degraded · refused · failed): the seam traust-engine calls; `Router` for continuous-ops `route()` | each CLI inventing its own exit codes | diff --git a/src/traust_core/v1/models/base.py b/src/traust_core/v1/models/base.py index d29192f5..8587d0fe 100644 --- a/src/traust_core/v1/models/base.py +++ b/src/traust_core/v1/models/base.py @@ -15,7 +15,14 @@ class Dto(pydantic.BaseModel): class Entity(Model): - # None until saved: the database generates id and both timestamps on insert. + # None until saved: the database generates id and both timestamps; on update + # the contracts touch trigger keeps updated_at. id: UUID | None = None created_at: datetime | None = None updated_at: datetime | None = None + + +class AppendOnly(Model): + # None until saved; append-only rows are never updated, so there is no updated_at. + id: UUID | None = None + created_at: datetime | None = None diff --git a/src/traust_core/v1/repositories/__init__.py b/src/traust_core/v1/repositories/__init__.py index e8c2365c..fcc1cf33 100644 --- a/src/traust_core/v1/repositories/__init__.py +++ b/src/traust_core/v1/repositories/__init__.py @@ -9,8 +9,10 @@ from traust_core.v1.repositories.sql import ( SqlRepository, SqlUnitOfWork, + append_only_columns, apply_schema, create_database_engine, + record_columns, schema_drift, ) from traust_core.v1.repositories.unit_of_work import UnitOfWork @@ -25,7 +27,9 @@ "SqlRepository", "SqlUnitOfWork", "UnitOfWork", + "append_only_columns", "apply_schema", "create_database_engine", + "record_columns", "schema_drift", ] diff --git a/src/traust_core/v1/repositories/sql.py b/src/traust_core/v1/repositories/sql.py index 02154e2a..4e2627dc 100644 --- a/src/traust_core/v1/repositories/sql.py +++ b/src/traust_core/v1/repositories/sql.py @@ -2,15 +2,17 @@ import json from collections.abc import Iterable, Mapping, Sequence -from datetime import datetime +from datetime import UTC, datetime from types import TracebackType from typing import Any, Literal, Self +from uuid import UUID, uuid4 import sqlalchemy as sa from sqlalchemy.engine import Connection, CursorResult, Engine from sqlalchemy.exc import IntegrityError, SQLAlchemyError +from sqlalchemy.ext.compiler import compiles -from traust_core.v1.errors import ConflictError, RepositoryError +from traust_core.v1.errors import ConflictError, NotFoundError, RepositoryError STORAGE_SCHEMA = "traust_storage" LEDGER_SCHEMA = "traust_ledger" @@ -26,8 +28,10 @@ def create_database_engine(url: str | sa.URL) -> Engine: return engine @sa.event.listens_for(engine, "connect") - def _fk_on(dbapi_conn: Any, _: Any) -> None: + def _sqlite_functions(dbapi_conn: Any, _: Any) -> None: dbapi_conn.execute("PRAGMA foreign_keys = ON") + # Contracts' uuid(): ext/misc/uuid.c in other drivers, registered here. + dbapi_conn.create_function("uuid", 0, lambda: str(uuid4())) return engine.execution_options( schema_translate_map={STORAGE_SCHEMA: None, LEDGER_SCHEMA: None} @@ -46,10 +50,84 @@ def load_dialect_impl(self, dialect: sa.Dialect) -> sa.TypeEngine[Any]: return dialect.type_descriptor(sa.Text() if dialect.name == "sqlite" else self.impl) def process_bind_param(self, value: datetime | None, dialect: sa.Dialect) -> Any: - return value.isoformat() if value is not None and dialect.name == "sqlite" else value + if value is None or dialect.name != "sqlite": + return value + # Contracts' TEXT format (millisecond precision, Z) so ordering survives mixed writers. + if value.tzinfo is None: + value = value.replace(tzinfo=UTC) + stamp = value.astimezone(UTC) + return stamp.strftime("%Y-%m-%dT%H:%M:%S") + f".{stamp.microsecond // 1000:03d}Z" def process_result_value(self, value: Any, dialect: sa.Dialect) -> datetime | None: - return datetime.fromisoformat(value) if isinstance(value, str) else value + if isinstance(value, str): + value = datetime.fromisoformat(value) + if value is not None and value.tzinfo is None: + value = value.replace(tzinfo=UTC) # SQLite CURRENT_TIMESTAMP is UTC without offset + return value + + +class RecordUuid(sa.TypeDecorator[UUID]): + # Native UUID on Postgres; on SQLite the hyphenated text uuid() writes, + # so lookups match the stored form. + impl = sa.Uuid + cache_ok = True + + def load_dialect_impl(self, dialect: sa.Dialect) -> sa.TypeEngine[Any]: + return dialect.type_descriptor(sa.Text() if dialect.name == "sqlite" else self.impl) + + def process_bind_param(self, value: UUID | None, dialect: sa.Dialect) -> Any: + return str(value) if value is not None and dialect.name == "sqlite" else value + + def process_result_value(self, value: Any, dialect: sa.Dialect) -> UUID | None: + return UUID(value) if isinstance(value, str) else value + + +class _GeneratedUuid(sa.FunctionElement[UUID]): + name = "generated_uuid" + inherit_cache = True + + +@compiles(_GeneratedUuid, "postgresql") +def _postgres_uuid(element: Any, compiler: Any, **kw: Any) -> str: + return "gen_random_uuid()" + + +@compiles(_GeneratedUuid, "sqlite") +def _sqlite_uuid(element: Any, compiler: Any, **kw: Any) -> str: + return "uuid()" + + +class _DbNow(sa.FunctionElement[datetime]): + name = "db_now" + inherit_cache = True + + +@compiles(_DbNow, "postgresql") +def _postgres_now(element: Any, compiler: Any, **kw: Any) -> str: + return "now()" + + +@compiles(_DbNow, "sqlite") +def _sqlite_now(element: Any, compiler: Any, **kw: Any) -> str: + # Contracts' TEXT timestamp format, millisecond precision. + return "strftime('%Y-%m-%dT%H:%M:%fZ','now')" + + +def record_columns() -> tuple[sa.Column[UUID], sa.Column[datetime], sa.Column[datetime]]: + """Mutable base columns: UUIDv4 id; the touch trigger keeps updated_at.""" + return ( + sa.Column("id", RecordUuid, primary_key=True, server_default=_GeneratedUuid()), + sa.Column("created_at", UtcTimestamp, nullable=False, server_default=_DbNow()), + sa.Column("updated_at", UtcTimestamp, nullable=False, server_default=_DbNow()), + ) + + +def append_only_columns() -> tuple[sa.Column[UUID], sa.Column[datetime]]: + """Append-only base columns: UUIDv4 id and created_at, no updated_at.""" + return ( + sa.Column("id", RecordUuid, primary_key=True, server_default=_GeneratedUuid()), + sa.Column("created_at", UtcTimestamp, nullable=False, server_default=_DbNow()), + ) def apply_schema(engine: Engine, statements: Iterable[str]) -> None: @@ -111,6 +189,42 @@ class SqlRepository: def __init__(self, connection: Connection) -> None: self._connection = connection + def _insert_record( + self, operation: str, table: sa.Table, values: Mapping[str, Any] + ) -> dict[str, UUID | datetime]: + """INSERT domain columns; returns the database-generated base columns the table has.""" + returning = [table.c.id, table.c.created_at] + if "updated_at" in table.c: + returning.append(table.c.updated_at) + stmt = sa.insert(table).values(**values).returning(*returning) + return dict(self._execute(operation, stmt).mappings().one()) + + def _update_record( + self, operation: str, table: sa.Table, record_id: UUID, values: Mapping[str, Any] + ) -> dict[str, UUID | datetime]: + """UPDATE by id; returns the saved base columns (the touch trigger keeps updated_at).""" + stmt = ( + sa.update(table) + .where(table.c.id == record_id) + .values(**values) + .returning(table.c.id, table.c.created_at, table.c.updated_at) + ) + row = self._execute(operation, stmt).mappings().first() + if row is None: + raise NotFoundError(f"{type(self).__name__}: {operation}: not found") + saved = dict(row) + if self._connection.dialect.name == "sqlite": + # SQLite's AFTER UPDATE trigger fires after RETURNING is computed. + saved["updated_at"] = ( + self._execute( + f"reread {record_id}", + sa.select(table.c.updated_at).where(table.c.id == record_id), + ) + .scalars() + .one() + ) + return saved + def _execute( self, operation: str, diff --git a/tests/v1/conftest.py b/tests/v1/conftest.py index c7337748..2e45b260 100644 --- a/tests/v1/conftest.py +++ b/tests/v1/conftest.py @@ -7,9 +7,10 @@ from pathlib import Path import pytest +import sqlalchemy as sa from sqlalchemy.engine import Engine -from tests.v1.example.repository import metadata +from tests.v1.example import repository as example_repository from traust_core.v1 import contracts from traust_core.v1.providers.clock import FixedClock from traust_core.v1.repositories import apply_schema, create_database_engine @@ -31,7 +32,9 @@ def _engine(kind: str, tmp_path: Path) -> Engine: @contextmanager -def database(kind: str, tmp_path: Path) -> Generator[Engine]: +def database( + kind: str, tmp_path: Path, metadata: sa.MetaData = example_repository.metadata +) -> Generator[Engine]: eng = _engine(kind, tmp_path) metadata.create_all(eng) try: diff --git a/tests/v1/repositories/test_sql.py b/tests/v1/repositories/test_sql.py new file mode 100644 index 00000000..02271a6f --- /dev/null +++ b/tests/v1/repositories/test_sql.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path +from re import fullmatch +from time import sleep +from uuid import UUID + +import pytest +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql, sqlite + +from tests.v1.conftest import SQL_BACKENDS, database +from traust_core.v1.errors import ConflictError, NotFoundError +from traust_core.v1.models.base import AppendOnly, Entity +from traust_core.v1.repositories import ( + SqlRepository, + SqlUnitOfWork, + append_only_columns, + apply_schema, + record_columns, +) + + +class Record(Entity): + name: str + status: str + + +class Reading(AppendOnly): + sensor: str + + +metadata = sa.MetaData() + +records = sa.Table( + "base_records", + metadata, + *record_columns(), + sa.Column("name", sa.Text, nullable=False, unique=True), + sa.Column("status", sa.Text, nullable=False), +) + +readings = sa.Table( + "base_readings", + metadata, + *append_only_columns(), + sa.Column("sensor", sa.Text, nullable=False), +) + + +def _to_row(record: Record) -> dict[str, str]: + return {"name": record.name, "status": record.status} + + +class RecordRepository(SqlRepository): + def add(self, record: Record) -> Record: + saved = self._insert_record(f"add {record.name}", records, _to_row(record)) + return record.model_copy(update=saved) + + def update(self, record: Record) -> Record: + assert record.id is not None + saved = self._update_record(f"update {record.id}", records, record.id, _to_row(record)) + return record.model_copy(update=saved) + + def get(self, record_id: UUID) -> Record: + stmt = sa.select(records).where(records.c.id == record_id) + row = self._execute(f"get {record_id}", stmt).mappings().one_or_none() + if row is None: + raise NotFoundError(f"record {record_id}") + return Record.model_validate(dict(row)) + + +class ReadingRepository(SqlRepository): + def add(self, reading: Reading) -> Reading: + saved = self._insert_record(f"add {reading.sensor}", readings, {"sensor": reading.sensor}) + return reading.model_copy(update=saved) + + def get(self, reading_id: UUID) -> Reading: + stmt = sa.select(readings).where(readings.c.id == reading_id) + row = self._execute(f"get {reading_id}", stmt).mappings().one_or_none() + if row is None: + raise NotFoundError(f"reading {reading_id}") + return Reading.model_validate(dict(row)) + + +class RecordUnitOfWork(SqlUnitOfWork): + records: RecordRepository + readings: ReadingRepository + + def _open_repositories(self) -> None: + self.records = RecordRepository(self.connection) + self.readings = ReadingRepository(self.connection) + + +# The contracts touch trigger, mirrored for the test table (updated_at is the DB's). +_TOUCH_SQLITE = """ +CREATE TRIGGER IF NOT EXISTS base_records_touch +AFTER UPDATE ON base_records FOR EACH ROW +WHEN NEW.updated_at IS OLD.updated_at BEGIN + UPDATE base_records SET updated_at = (strftime('%Y-%m-%dT%H:%M:%fZ','now')) WHERE id = NEW.id; +END; +""" + +_TOUCH_POSTGRES = """ +CREATE OR REPLACE FUNCTION base_records_touch_updated_at() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + NEW.updated_at := now(); + RETURN NEW; +END +$$; +DROP TRIGGER IF EXISTS base_records_touch ON base_records; +CREATE TRIGGER base_records_touch BEFORE UPDATE ON base_records +FOR EACH ROW EXECUTE FUNCTION base_records_touch_updated_at(); +""" + + +@pytest.fixture(params=SQL_BACKENDS) +def uow(request: pytest.FixtureRequest, tmp_path: Path) -> Iterator[RecordUnitOfWork]: + with database(request.param, tmp_path, metadata) as engine: + touch = _TOUCH_POSTGRES if request.param == "postgres" else _TOUCH_SQLITE + apply_schema(engine, [touch]) + yield RecordUnitOfWork(engine) + + +def test_record_columns_declare_the_base_shape() -> None: + id_col, created_col, updated_col = record_columns() + assert id_col.name == "id" + assert id_col.primary_key + assert created_col.name == "created_at" + assert updated_col.name == "updated_at" + assert created_col.nullable is False + assert updated_col.nullable is False + assert updated_col.onupdate is None # the touch trigger owns updated_at, not the app + + +def test_append_only_columns_declare_the_base_shape() -> None: + id_col, created_col = append_only_columns() + assert id_col.name == "id" + assert id_col.primary_key + assert created_col.name == "created_at" + assert created_col.nullable is False + assert "updated_at" not in readings.c + assert "updated_at" not in Reading.model_fields + + +def test_uuid_default_compiles_per_dialect() -> None: + create = sa.schema.CreateTable(records) + sqlite_ddl = str(create.compile(dialect=sqlite.dialect())) + postgres_ddl = str(create.compile(dialect=postgresql.dialect())) + assert "uuid()" in sqlite_ddl + assert "gen_random_uuid()" in postgres_ddl + + +def test_add_returns_the_database_generated_record(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.records.add(Record(name="a", status="open")) + u.commit() + assert isinstance(saved.id, UUID) + assert saved.id.version == 4 + assert saved.created_at is not None + assert saved.created_at.tzinfo is not None + assert saved.updated_at is not None + assert saved.updated_at >= saved.created_at + + +def test_append_only_add_returns_the_database_generated_reading(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.readings.add(Reading(sensor="temp")) + u.commit() + assert isinstance(saved.id, UUID) + assert saved.id.version == 4 + assert saved.created_at is not None + assert saved.created_at.tzinfo is not None + + +def test_append_only_reading_round_trips_by_id(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.readings.add(Reading(sensor="temp")) + u.commit() + assert saved.id is not None + with uow as u: + reread = u.readings.get(saved.id) + assert reread == saved + + +def test_each_row_gets_its_own_id(uow: RecordUnitOfWork) -> None: + with uow as u: + first = u.records.add(Record(name="a", status="open")) + second = u.records.add(Record(name="b", status="open")) + u.commit() + assert first.id != second.id + + +def test_duplicate_natural_key_raises_conflict(uow: RecordUnitOfWork) -> None: + with uow as u: + u.records.add(Record(name="a", status="open")) + u.commit() + with uow as u: + with pytest.raises(ConflictError): + u.records.add(Record(name="a", status="open")) + u.rollback() + + +def test_update_returns_the_saved_record_and_maintains_updated_at(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.records.add(Record(name="a", status="open")) + u.commit() + assert saved.id is not None + sleep(0.01) # the touch trigger has millisecond precision + with uow as u: + done = u.records.update(saved.model_copy(update={"status": "done"})) + u.commit() + assert done.status == "done" + assert done.updated_at is not None + assert done.updated_at > done.created_at # type: ignore[operator] + + +def test_raw_sql_update_outside_the_app_still_touches_updated_at(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.records.add(Record(name="a", status="open")) + u.commit() + assert saved.id is not None + sleep(0.01) + with uow as u: + u.records._execute( + f"raw update {saved.id}", + sa.text("UPDATE base_records SET status = 'done' WHERE id = :id"), + {"id": str(saved.id)}, # raw SQL carries no type information + ) + u.commit() + reread = u.records.get(saved.id) + assert reread.status == "done" + assert reread.updated_at > saved.updated_at # type: ignore[operator] + + +def test_sqlite_columns_match_the_contracts_text_formats(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.readings.add(Reading(sensor="temp")) + u.commit() + assert saved.id is not None + with uow as u: + row = u.readings._execute( + f"raw get {saved.id}", + sa.text("SELECT id, created_at FROM base_readings WHERE id = :id"), + {"id": str(saved.id)}, + ).one() + if not isinstance(row[0], str): # Postgres returns native types + return + assert fullmatch(r"[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}", row[0]) + assert fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", row[1]) + + +def test_update_missing_row_raises_not_found(uow: RecordUnitOfWork) -> None: + with uow as u: + with pytest.raises(NotFoundError): + u.records.update(Record(id=UUID(int=1), name="a", status="done")) + u.rollback() + + +def test_saved_record_round_trips_by_id(uow: RecordUnitOfWork) -> None: + with uow as u: + saved = u.records.add(Record(name="a", status="open")) + u.commit() + assert saved.id is not None + with uow as u: + reread = u.records.get(saved.id) + assert reread == saved