From 7426db20cb85d721e2b8d0cc5d06bcec793459d9 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:03:17 +0000 Subject: [PATCH 01/12] feat(v2): add V2RecordStatus enum (pending/completed/discarded) --- extralit-server/src/extralit_server/enums.py | 9 +++++++++ extralit-server/tests/integration/test_enums_v2.py | 8 +++++++- 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/extralit-server/src/extralit_server/enums.py b/extralit-server/src/extralit_server/enums.py index 9a82f6508..a8c12fc7b 100644 --- a/extralit-server/src/extralit_server/enums.py +++ b/extralit-server/src/extralit_server/enums.py @@ -105,3 +105,12 @@ class SchemaKind(StrEnum): class SchemaStatus(StrEnum): draft = "draft" published = "published" + + +class V2RecordStatus(StrEnum): + """v2 record status. Distinct from v1 RecordStatus: adds `discarded` and maps to its + own PG enum type (v2_record_status_enum) so v1's record_status_enum is untouched.""" + + pending = "pending" + completed = "completed" + discarded = "discarded" diff --git a/extralit-server/tests/integration/test_enums_v2.py b/extralit-server/tests/integration/test_enums_v2.py index 529bb2901..3538770ec 100644 --- a/extralit-server/tests/integration/test_enums_v2.py +++ b/extralit-server/tests/integration/test_enums_v2.py @@ -1,4 +1,4 @@ -from extralit_server.enums import SchemaKind, SchemaStatus +from extralit_server.enums import SchemaKind, SchemaStatus, V2RecordStatus def test_schema_kind_values(): @@ -11,3 +11,9 @@ def test_schema_status_values(): assert SchemaStatus.draft == "draft" assert SchemaStatus.published == "published" assert {s.value for s in SchemaStatus} == {"draft", "published"} + + +def test_v2_record_status_values(): + assert V2RecordStatus.pending == "pending" + assert V2RecordStatus.discarded == "discarded" + assert {s.value for s in V2RecordStatus} == {"pending", "completed", "discarded"} From 58e66a0e70d0055f82548abe8d78af4d835858d5 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:06:44 +0000 Subject: [PATCH 02/12] feat(v2): V2Record ORM model (v2_records) pinned to schema versions --- .../src/extralit_server/models/__init__.py | 1 + .../src/extralit_server/models/v2/__init__.py | 4 +- .../src/extralit_server/models/v2/records.py | 53 ++++++++++++++++ .../models/v2/test_record_models.py | 62 +++++++++++++++++++ 4 files changed, 119 insertions(+), 1 deletion(-) create mode 100644 extralit-server/src/extralit_server/models/v2/records.py create mode 100644 extralit-server/tests/integration/models/v2/test_record_models.py diff --git a/extralit-server/src/extralit_server/models/__init__.py b/extralit-server/src/extralit_server/models/__init__.py index 5d092989b..6f882d7c8 100644 --- a/extralit-server/src/extralit_server/models/__init__.py +++ b/extralit-server/src/extralit_server/models/__init__.py @@ -6,3 +6,4 @@ from .database import * from .metadata_properties import * from .v2 import Schema, SchemaVersion # register v2 tables on DatabaseModel.metadata +from .v2 import Record as V2Record # aliased: v1 Record is star-exported above diff --git a/extralit-server/src/extralit_server/models/v2/__init__.py b/extralit-server/src/extralit_server/models/v2/__init__.py index 62fca1211..caae0054b 100644 --- a/extralit-server/src/extralit_server/models/v2/__init__.py +++ b/extralit-server/src/extralit_server/models/v2/__init__.py @@ -1,3 +1,5 @@ +from extralit_server.models.v2.records import V2Record +from extralit_server.models.v2.records import V2Record as Record # v2-namespace alias from extralit_server.models.v2.schemas import Schema, SchemaVersion -__all__ = ["Schema", "SchemaVersion"] +__all__ = ["Record", "Schema", "SchemaVersion", "V2Record"] diff --git a/extralit-server/src/extralit_server/models/v2/records.py b/extralit-server/src/extralit_server/models/v2/records.py new file mode 100644 index 000000000..fc4644422 --- /dev/null +++ b/extralit-server/src/extralit_server/models/v2/records.py @@ -0,0 +1,53 @@ +from typing import TYPE_CHECKING +from uuid import UUID + +from sqlalchemy import JSON, ForeignKey, Index, String, UniqueConstraint +from sqlalchemy import Enum as SAEnum +from sqlalchemy.ext.mutable import MutableDict +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from extralit_server.enums import V2RecordStatus +from extralit_server.models.base import DatabaseModel + +if TYPE_CHECKING: + from extralit_server.models.v2.schemas import Schema, SchemaVersion + +V2RecordStatusEnum = SAEnum(V2RecordStatus, name="v2_record_status_enum") + + +class V2Record(DatabaseModel): + """v2 record: one typed row pinned to a schema version (spec §5). + + Table is `v2_records` because v1 owns `records`; the class is `V2Record` because a second + declarative class named `Record` breaks v1's string-based `relationship("Record")` lookups. + Both are renamed to the canonical names on v1 retirement (Phase 6). + """ + + __tablename__ = "v2_records" + + schema_id: Mapped[UUID] = mapped_column(ForeignKey("schemas.id", ondelete="CASCADE"), index=True) + # CASCADE (not RESTRICT): versions are immutable and only deleted via the schema + # cascade; RESTRICT could break the schemas-delete cascade on FK-check ordering. + schema_version_id: Mapped[UUID] = mapped_column(ForeignKey("schema_versions.id", ondelete="CASCADE")) + reference: Mapped[str] = mapped_column(String, index=True) + external_id: Mapped[str | None] = mapped_column(String, nullable=True) + fields: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON), default=dict) + metadata_: Mapped[dict | None] = mapped_column("metadata", MutableDict.as_mutable(JSON), nullable=True) + status: Mapped[V2RecordStatus] = mapped_column( + V2RecordStatusEnum, default=V2RecordStatus.pending, server_default=V2RecordStatus.pending, index=True + ) + + # One-directional (Phase 1 convention): no Schema.records collection; DB-level CASCADE owns deletes. + schema: Mapped["Schema"] = relationship("Schema") + version: Mapped["SchemaVersion"] = relationship("SchemaVersion") + + __table_args__ = ( + UniqueConstraint("schema_id", "external_id", name="v2_record_schema_id_external_id_uq"), + Index("ix_v2_records_schema_id_reference", "schema_id", "reference"), + ) + + def __repr__(self) -> str: + return ( + f"Record(id={self.id!s}, schema_id={self.schema_id!s}, " + f"reference={self.reference!r}, status={self.status!r})" + ) diff --git a/extralit-server/tests/integration/models/v2/test_record_models.py b/extralit-server/tests/integration/models/v2/test_record_models.py new file mode 100644 index 000000000..d1a41e824 --- /dev/null +++ b/extralit-server/tests/integration/models/v2/test_record_models.py @@ -0,0 +1,62 @@ +import pytest +from sqlalchemy.exc import IntegrityError + +from extralit_server.enums import V2RecordStatus +from extralit_server.models.v2 import Record +from tests.factories import SchemaVersionFactory + +pytestmark = pytest.mark.asyncio + + +async def test_create_record_with_defaults(db): + version = await SchemaVersionFactory.create() + record = await Record.create( + db, + schema_id=version.schema_id, + schema_version_id=version.id, + reference="pmid:12345", + fields={"name": "Anopheles", "age": 3}, + ) + assert record.id is not None + assert record.status == V2RecordStatus.pending + assert record.external_id is None + assert record.metadata_ is None + + loaded = await Record.get(db, record.id) + assert loaded.reference == "pmid:12345" + assert loaded.fields == {"name": "Anopheles", "age": 3} + + +async def test_duplicate_external_id_in_schema_raises(db): + version = await SchemaVersionFactory.create() + db.add_all( + [ + Record( + schema_id=version.schema_id, + schema_version_id=version.id, + reference="r", + external_id="x", + fields={}, + ), + Record( + schema_id=version.schema_id, + schema_version_id=version.id, + reference="r", + external_id="x", + fields={}, + ), + ] + ) + with pytest.raises(IntegrityError, match="v2_record_schema_id_external_id_uq|UNIQUE"): + await db.commit() + + +async def test_null_external_ids_do_not_collide(db): + version = await SchemaVersionFactory.create() + db.add_all( + [ + Record(schema_id=version.schema_id, schema_version_id=version.id, reference="r", fields={}), + Record(schema_id=version.schema_id, schema_version_id=version.id, reference="r", fields={}), + ] + ) + await db.commit() # nullable uniq: multiple NULLs allowed From 50cb7fc5abcdd7fa9ce5b2210189ac11c4be4a25 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:10:21 +0000 Subject: [PATCH 03/12] feat(v2): alembic migration for v2_records table --- .../8136bc88ee3a_create_v2_records_table.py | 54 +++++++++++++++++++ .../src/extralit_server/models/v2/records.py | 2 +- 2 files changed, 55 insertions(+), 1 deletion(-) create mode 100644 extralit-server/src/extralit_server/alembic/versions/8136bc88ee3a_create_v2_records_table.py diff --git a/extralit-server/src/extralit_server/alembic/versions/8136bc88ee3a_create_v2_records_table.py b/extralit-server/src/extralit_server/alembic/versions/8136bc88ee3a_create_v2_records_table.py new file mode 100644 index 000000000..db095d0d7 --- /dev/null +++ b/extralit-server/src/extralit_server/alembic/versions/8136bc88ee3a_create_v2_records_table.py @@ -0,0 +1,54 @@ +"""create v2_records table + +Revision ID: 8136bc88ee3a +Revises: 9f3010c649c8 +Create Date: 2026-07-03 18:07:04.507576 + +""" + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = "8136bc88ee3a" +down_revision = "9f3010c649c8" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "v2_records", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("schema_id", sa.Uuid(), nullable=False), + sa.Column("schema_version_id", sa.Uuid(), nullable=False), + sa.Column("reference", sa.String(), nullable=False), + sa.Column("external_id", sa.String(), nullable=True), + sa.Column("fields", sa.JSON(), nullable=False), + sa.Column("metadata", sa.JSON(), nullable=True), + sa.Column( + "status", + sa.Enum("pending", "completed", "discarded", name="v2_record_status_enum"), + server_default="pending", + nullable=False, + ), + sa.Column("inserted_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(["schema_id"], ["schemas.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["schema_version_id"], ["schema_versions.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("schema_id", "external_id", name="v2_record_schema_id_external_id_uq"), + ) + op.create_index(op.f("ix_v2_records_schema_id"), "v2_records", ["schema_id"], unique=False) + op.create_index(op.f("ix_v2_records_reference"), "v2_records", ["reference"], unique=False) + op.create_index(op.f("ix_v2_records_status"), "v2_records", ["status"], unique=False) + op.create_index("ix_v2_records_schema_id_reference", "v2_records", ["schema_id", "reference"], unique=False) + + +def downgrade() -> None: + op.drop_index("ix_v2_records_schema_id_reference", table_name="v2_records") + op.drop_index(op.f("ix_v2_records_status"), table_name="v2_records") + op.drop_index(op.f("ix_v2_records_reference"), table_name="v2_records") + op.drop_index(op.f("ix_v2_records_schema_id"), table_name="v2_records") + op.drop_table("v2_records") + sa.Enum(name="v2_record_status_enum").drop(op.get_bind(), checkfirst=True) diff --git a/extralit-server/src/extralit_server/models/v2/records.py b/extralit-server/src/extralit_server/models/v2/records.py index fc4644422..2b534e551 100644 --- a/extralit-server/src/extralit_server/models/v2/records.py +++ b/extralit-server/src/extralit_server/models/v2/records.py @@ -48,6 +48,6 @@ class V2Record(DatabaseModel): def __repr__(self) -> str: return ( - f"Record(id={self.id!s}, schema_id={self.schema_id!s}, " + f"V2Record(id={self.id!s}, schema_id={self.schema_id!s}, " f"reference={self.reference!r}, status={self.status!r})" ) From 92805b204bb817267485c178a686ffefbf955ead Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:12:20 +0000 Subject: [PATCH 04/12] test(v2): V2RecordFactory --- extralit-server/tests/factories.py | 25 +++++++++++++++++++ .../models/v2/test_record_models.py | 13 ++++++++++ 2 files changed, 38 insertions(+) diff --git a/extralit-server/tests/factories.py b/extralit-server/tests/factories.py index d8452802c..15b9e8ca2 100644 --- a/extralit-server/tests/factories.py +++ b/extralit-server/tests/factories.py @@ -40,6 +40,7 @@ from extralit_server.models.base import DatabaseModel from extralit_server.models.v2 import Schema as SchemaModel from extralit_server.models.v2 import SchemaVersion as SchemaVersionModel +from extralit_server.models.v2 import V2Record as V2RecordModel from extralit_server.webhooks.v1.enums import WebhookEvent from tests.database import SyncTestSession, TestSession @@ -671,3 +672,27 @@ class Meta: etag = factory.Sequence(lambda n: f"etag-{n}") checksum = factory.Sequence(lambda n: f"checksum-{n}") columns_cache = factory.LazyFunction(list) # fresh list per row, not a shared mutable default + + +class V2RecordFactory(BaseFactory): + class Meta: + model = V2RecordModel + + version = factory.SubFactory(SchemaVersionFactory) + reference = factory.Sequence(lambda n: f"ref-{n}") + external_id = factory.Sequence(lambda n: f"v2-external-{n}") + fields = factory.LazyFunction(dict) # fresh dict per row, not a shared mutable default + + @classmethod + async def _create(cls, model_class, *args, **kwargs): + # LazyAttribute cannot derive the FK columns because the SubFactory result is still a + # coroutine during attribute evaluation (see SchemaVersionFactory.object_key); await it + # here and wire schema_id/schema_version_id to the version's schema. + version = kwargs.get("version") + if inspect.isawaitable(version): + version = await version + kwargs["version"] = version + if version is not None: + kwargs.setdefault("schema_version_id", version.id) + kwargs.setdefault("schema_id", version.schema_id) + return await super()._create(model_class, *args, **kwargs) diff --git a/extralit-server/tests/integration/models/v2/test_record_models.py b/extralit-server/tests/integration/models/v2/test_record_models.py index d1a41e824..b47f004ad 100644 --- a/extralit-server/tests/integration/models/v2/test_record_models.py +++ b/extralit-server/tests/integration/models/v2/test_record_models.py @@ -51,6 +51,19 @@ async def test_duplicate_external_id_in_schema_raises(db): await db.commit() +async def test_v2_record_factory_builds_valid_row(db): + from tests.factories import V2RecordFactory + + record = await V2RecordFactory.create() + assert record.id is not None + assert record.schema_version_id is not None + assert record.schema_id is not None + assert record.status == V2RecordStatus.pending + + version = await record.awaitable_attrs.version + assert record.schema_id == version.schema_id # factory wires the record to the version's schema + + async def test_null_external_ids_do_not_collide(db): version = await SchemaVersionFactory.create() db.add_all( From 1d9dbd6cfe50677da8998c2a736d1ad12fa378c7 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:12:57 +0000 Subject: [PATCH 05/12] feat(v2): pydantic request/response schemas for v2 records and references --- .../extralit_server/api/schemas/v2/records.py | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 extralit-server/src/extralit_server/api/schemas/v2/records.py diff --git a/extralit-server/src/extralit_server/api/schemas/v2/records.py b/extralit-server/src/extralit_server/api/schemas/v2/records.py new file mode 100644 index 000000000..1a022bee5 --- /dev/null +++ b/extralit-server/src/extralit_server/api/schemas/v2/records.py @@ -0,0 +1,67 @@ +from datetime import datetime +from typing import Any +from uuid import UUID + +from pydantic import AliasChoices, BaseModel, ConfigDict, Field, constr + +from extralit_server.enums import V2RecordStatus + +RECORDS_BULK_UPSERT_MIN_ITEMS = 1 +RECORDS_BULK_UPSERT_MAX_ITEMS = 500 # mirrors v1 RECORDS_BULK_CREATE_MAX_ITEMS +LIST_RECORDS_LIMIT_DEFAULT = 50 # mirrors v1 LIST_DATASET_RECORDS_LIMIT_DEFAULT +LIST_RECORDS_LIMIT_LE = 1000 # mirrors v1 LIST_DATASET_RECORDS_LIMIT_LE +DELETE_RECORDS_LIMIT = 100 # mirrors v1 DELETE_DATASET_RECORDS_LIMIT + +Reference = constr(min_length=1, max_length=500) + + +class RecordUpsert(BaseModel): + fields: dict[str, Any] + reference: Reference + external_id: str | None = None + metadata: dict[str, Any] | None = None + status: V2RecordStatus | None = None + schema_version_id: UUID | None = Field( + default=None, description="Pin to a specific version; defaults to the schema's current_version_id" + ) + + +class RecordsBulkUpsert(BaseModel): + items: list[RecordUpsert] = Field( + ..., min_length=RECORDS_BULK_UPSERT_MIN_ITEMS, max_length=RECORDS_BULK_UPSERT_MAX_ITEMS + ) + + +class RecordRead(BaseModel): + model_config = ConfigDict(from_attributes=True, populate_by_name=True) + + id: UUID + schema_id: UUID + schema_version_id: UUID + reference: str + external_id: str | None + fields: dict[str, Any] + # ORM attr is `metadata_` (column "metadata"); accept either name, serialize as `metadata`. + metadata: dict[str, Any] | None = Field(default=None, validation_alias=AliasChoices("metadata_", "metadata")) + status: V2RecordStatus + inserted_at: datetime + updated_at: datetime + + +class Records(BaseModel): + items: list[RecordRead] + total: int + + +class ReferenceGroup(BaseModel): + # Flattened schema_id/schema_name (not a nested `schema:` field) to avoid pydantic's + # BaseModel.schema() attribute shadowing. + schema_id: UUID + schema_name: str + records: list[RecordRead] + + +class ReferenceView(BaseModel): + reference: str + groups: list[ReferenceGroup] + total_records: int From b01b11a7aa59781866d8b994e933c1c744722b52 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:15:22 +0000 Subject: [PATCH 06/12] feat(v2): records context with validated bulk-upsert, list, delete, reference query --- .../extralit_server/contexts/v2/records.py | 151 ++++++++++++ .../contexts/v2/test_records_context.py | 227 ++++++++++++++++++ 2 files changed, 378 insertions(+) create mode 100644 extralit-server/src/extralit_server/contexts/v2/records.py create mode 100644 extralit-server/tests/integration/contexts/v2/test_records_context.py diff --git a/extralit-server/src/extralit_server/contexts/v2/records.py b/extralit-server/src/extralit_server/contexts/v2/records.py new file mode 100644 index 000000000..1ee56f2cc --- /dev/null +++ b/extralit-server/src/extralit_server/contexts/v2/records.py @@ -0,0 +1,151 @@ +"""Business logic for v2 Records: validated bulk-upsert, listing, deletion, reference view.""" + +from typing import TYPE_CHECKING +from uuid import UUID + +from sqlalchemy import delete as sql_delete +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from extralit_server.api.schemas.v2.records import RecordUpsert +from extralit_server.contexts import files as files_ctx +from extralit_server.contexts.v2.schema_bodies import SchemaValidationError, validate_record_fields +from extralit_server.enums import V2RecordStatus +from extralit_server.errors.future import UnprocessableEntityError +from extralit_server.models.v2 import Schema, SchemaVersion, V2Record + +if TYPE_CHECKING: + from types_aiobotocore_s3.client import S3Client + + +async def _fetch_body_json(s3_client: "S3Client", bucket: str, version: SchemaVersion) -> str: + """Fetch a schema version's Pandera body from the object store, pinned to its S3 version.""" + obj = await files_ctx.get_object(s3_client, bucket, version.object_key, version_id=version.object_version_id) + data = await obj.response.read() + return data.decode("utf-8") if isinstance(data, bytes) else data + + +async def bulk_upsert_records( + db: AsyncSession, + s3_client: "S3Client", + schema: Schema, + *, + items: list[RecordUpsert], + bucket: str, +) -> list[V2Record]: + """Validate every item against its (pinned or current) schema version, then upsert. + + All-or-nothing: every item is validated against the Pandera body before any row is + written, so a single invalid item fails the whole request without partial writes. + Identity for updates is (schema_id, external_id); items without an external_id always + insert. Each distinct schema version's body is fetched from the object store once per + request, never per record. + """ + default_version_id = schema.current_version_id + if default_version_id is None: + raise UnprocessableEntityError( + f"Schema `{schema.id}` has no published version; publish a version before writing records" + ) + + provided = [item.external_id for item in items if item.external_id is not None] + if len(provided) != len(set(provided)): + raise UnprocessableEntityError("Duplicate `external_id` values in the same bulk-upsert payload") + + # Resolve distinct pinned versions; every pin must belong to this schema. + version_ids = {item.schema_version_id or default_version_id for item in items} + versions: dict[UUID, SchemaVersion] = {} + for version_id in version_ids: + version = await SchemaVersion.get(db, version_id) + if version is None or version.schema_id != schema.id: + raise UnprocessableEntityError(f"Schema version `{version_id}` does not belong to schema `{schema.id}`") + versions[version_id] = version + + bodies: dict[UUID, str] = {} + for version_id, version in versions.items(): + bodies[version_id] = await _fetch_body_json(s3_client, bucket, version) + + errors: list[str] = [] + validated_fields: list[dict] = [] + for idx, item in enumerate(items): + version_id = item.schema_version_id or default_version_id + try: + validated_fields.append(validate_record_fields(bodies[version_id], item.fields)) + except SchemaValidationError as exc: + validated_fields.append({}) + errors.extend( + f"items[{idx}]: column={e['column']!r} check={e['check']!r} error={e['error']}" for e in exc.errors + ) + if errors: + raise UnprocessableEntityError("Record fields failed schema validation: " + "; ".join(errors)) + + # v1-style merge (contexts/records_bulk.py): external_id is nullable, which rules out a + # single ON CONFLICT statement, and we must return ORM rows in input order. + existing: dict[str, V2Record] = {} + if provided: + stmt = select(V2Record).where(V2Record.schema_id == schema.id, V2Record.external_id.in_(provided)) + existing = {r.external_id: r for r in (await db.execute(stmt)).scalars().all()} + + result: list[V2Record] = [] + for item, fields in zip(items, validated_fields, strict=False): + version_id = item.schema_version_id or default_version_id + record = existing.get(item.external_id) if item.external_id is not None else None + if record is None: + record = V2Record( + schema_id=schema.id, + schema_version_id=version_id, + reference=item.reference, + external_id=item.external_id, + fields=fields, + metadata_=item.metadata, + status=item.status or V2RecordStatus.pending, + ) + db.add(record) + else: + record.schema_version_id = version_id + record.reference = item.reference + record.fields = fields + if item.metadata is not None: + record.metadata_ = item.metadata + if item.status is not None: + record.status = item.status + result.append(record) + + await db.flush() + await db.commit() + return result + + +async def list_records( + db: AsyncSession, + schema: Schema, + *, + offset: int, + limit: int, + status: V2RecordStatus | None = None, + reference: str | None = None, +) -> tuple[list[V2Record], int]: + filters = [V2Record.schema_id == schema.id] + if status is not None: + filters.append(V2Record.status == status) + if reference is not None: + filters.append(V2Record.reference == reference) + + total = (await db.execute(select(func.count(V2Record.id)).where(*filters))).scalar_one() + stmt = select(V2Record).where(*filters).order_by(V2Record.inserted_at.asc()).offset(offset).limit(limit) + return (await db.execute(stmt)).scalars().all(), total + + +async def delete_records(db: AsyncSession, schema: Schema, record_ids: list[UUID]) -> int: + result = await db.execute(sql_delete(V2Record).where(V2Record.id.in_(record_ids), V2Record.schema_id == schema.id)) + await db.commit() + return result.rowcount + + +async def list_records_by_reference(db: AsyncSession, *, workspace_id: UUID, reference: str) -> list[V2Record]: + stmt = ( + select(V2Record) + .join(Schema, V2Record.schema_id == Schema.id) + .where(Schema.workspace_id == workspace_id, V2Record.reference == reference) + .order_by(V2Record.schema_id, V2Record.inserted_at.asc()) + ) + return (await db.execute(stmt)).scalars().all() diff --git a/extralit-server/tests/integration/contexts/v2/test_records_context.py b/extralit-server/tests/integration/contexts/v2/test_records_context.py new file mode 100644 index 000000000..87c4e77f1 --- /dev/null +++ b/extralit-server/tests/integration/contexts/v2/test_records_context.py @@ -0,0 +1,227 @@ +from unittest.mock import AsyncMock + +import pandera.pandas as pa +import pytest + +from extralit_server.contexts.v2 import records as records_ctx +from extralit_server.enums import SchemaKind, V2RecordStatus +from extralit_server.errors.future import UnprocessableEntityError +from extralit_server.models.v2 import Schema, SchemaVersion, V2Record +from tests.factories import SchemaFactory, SchemaVersionFactory, V2RecordFactory, WorkspaceFactory + +pytestmark = pytest.mark.asyncio + +BODY = pa.DataFrameSchema( + columns={ + "name": pa.Column(pa.String, nullable=False), + "age": pa.Column(pa.Int, nullable=True), + } +).to_json() + + +def _patch_fetch(monkeypatch, body: str = BODY) -> AsyncMock: + fetch = AsyncMock(return_value=body) + monkeypatch.setattr("extralit_server.contexts.v2.records._fetch_body_json", fetch) + return fetch + + +async def _published_schema(db) -> tuple[Schema, SchemaVersion]: + schema = await SchemaFactory.create(kind=SchemaKind.table) + version = await SchemaVersionFactory.create(schema=schema, version=1) + await schema.update(db, current_version_id=version.id) + return schema, version + + +def _item(**overrides) -> dict: + from extralit_server.api.schemas.v2.records import RecordUpsert + + payload = {"fields": {"name": "Ada", "age": 36}, "reference": "pmid:1"} + payload.update(overrides) + return RecordUpsert(**payload) + + +async def test_bulk_upsert_creates_records_in_input_order(db, monkeypatch): + fetch = _patch_fetch(monkeypatch) + schema, version = await _published_schema(db) + + s3 = AsyncMock() + records = await records_ctx.bulk_upsert_records( + db, + s3, + schema, + items=[ + _item(external_id="x-1", fields={"name": "Ada", "age": 36}), + _item(fields={"name": "Grace", "age": None}, reference="pmid:2"), + ], + bucket="ws", + ) + + assert [r.external_id for r in records] == ["x-1", None] + assert records[0].schema_version_id == version.id + assert records[0].fields == {"name": "Ada", "age": 36} + assert records[1].fields == {"name": "Grace", "age": None} + assert records[0].status == V2RecordStatus.pending + assert fetch.await_count == 1 + + +async def test_bulk_upsert_updates_existing_by_external_id(db, monkeypatch): + _patch_fetch(monkeypatch) + schema, _ = await _published_schema(db) + s3 = AsyncMock() + + first = await records_ctx.bulk_upsert_records(db, s3, schema, items=[_item(external_id="x-1")], bucket="ws") + updated = await records_ctx.bulk_upsert_records( + db, + s3, + schema, + items=[_item(external_id="x-1", fields={"name": "Ada L.", "age": 37}, status=V2RecordStatus.completed)], + bucket="ws", + ) + + assert updated[0].id == first[0].id + assert updated[0].fields == {"name": "Ada L.", "age": 37} + assert updated[0].status == V2RecordStatus.completed + + _, total = await records_ctx.list_records(db, schema, offset=0, limit=10) + assert total == 1 + + +async def test_bulk_upsert_fetches_body_once_per_distinct_version(db, monkeypatch): + fetch = _patch_fetch(monkeypatch) + schema, version1 = await _published_schema(db) + version2 = await SchemaVersionFactory.create(schema=schema, version=2) + await schema.update(db, current_version_id=version2.id) + + s3 = AsyncMock() + records = await records_ctx.bulk_upsert_records( + db, + s3, + schema, + items=[ + _item(schema_version_id=version1.id), + _item(reference="pmid:2"), + _item(schema_version_id=version1.id, reference="pmid:3"), + _item(reference="pmid:4"), + ], + bucket="ws", + ) + + assert fetch.await_count == 2 + assert records[0].schema_version_id == version1.id + assert records[1].schema_version_id == version2.id + + +async def test_bulk_upsert_validation_failure_is_all_or_nothing(db, monkeypatch): + _patch_fetch(monkeypatch) + schema, _ = await _published_schema(db) + s3 = AsyncMock() + + with pytest.raises(UnprocessableEntityError) as exc: + await records_ctx.bulk_upsert_records( + db, + s3, + schema, + items=[_item(), _item(fields={"name": "Bob", "age": "not-a-number"}, reference="pmid:2")], + bucket="ws", + ) + assert "items[1]" in exc.value.message + assert "age" in exc.value.message + + _, total = await records_ctx.list_records(db, schema, offset=0, limit=10) + assert total == 0 + + +async def test_bulk_upsert_requires_published_version(db, monkeypatch): + _patch_fetch(monkeypatch) + schema = await SchemaFactory.create() + + with pytest.raises(UnprocessableEntityError, match="no published version"): + await records_ctx.bulk_upsert_records(db, AsyncMock(), schema, items=[_item()], bucket="ws") + + +async def test_bulk_upsert_rejects_foreign_schema_version_pin(db, monkeypatch): + _patch_fetch(monkeypatch) + schema, _ = await _published_schema(db) + other_schema, other_version = await _published_schema(db) + + with pytest.raises(UnprocessableEntityError, match="does not belong"): + await records_ctx.bulk_upsert_records( + db, AsyncMock(), schema, items=[_item(schema_version_id=other_version.id)], bucket="ws" + ) + + +async def test_bulk_upsert_rejects_duplicate_external_ids_in_payload(db, monkeypatch): + _patch_fetch(monkeypatch) + schema, _ = await _published_schema(db) + + with pytest.raises(UnprocessableEntityError, match="[Dd]uplicate"): + await records_ctx.bulk_upsert_records( + db, + AsyncMock(), + schema, + items=[_item(external_id="x-1"), _item(external_id="x-1", reference="pmid:2")], + bucket="ws", + ) + + +async def test_list_records_paginates_and_filters(db): + schema, version = await _published_schema(db) + other_schema, other_version = await _published_schema(db) + + r1 = await V2RecordFactory.create(version=version, reference="pmid:1") + r2 = await V2RecordFactory.create(version=version, reference="pmid:1", status=V2RecordStatus.completed) + r3 = await V2RecordFactory.create(version=version, reference="pmid:2") + await V2RecordFactory.create(version=other_version) + + items, total = await records_ctx.list_records(db, schema, offset=0, limit=10) + assert total == 3 + assert [r.id for r in items] == [r1.id, r2.id, r3.id] + + items, total = await records_ctx.list_records(db, schema, offset=1, limit=1) + assert total == 3 + assert len(items) == 1 + + items, total = await records_ctx.list_records(db, schema, offset=0, limit=10, reference="pmid:1") + assert total == 2 + + items, total = await records_ctx.list_records(db, schema, offset=0, limit=10, status=V2RecordStatus.completed) + assert total == 1 + assert items[0].id == r2.id + + +async def test_delete_records_is_schema_scoped(db): + schema, version = await _published_schema(db) + _, other_version = await _published_schema(db) + + r1 = await V2RecordFactory.create(version=version) + r2 = await V2RecordFactory.create(version=version) + r3 = await V2RecordFactory.create(version=version) + foreign = await V2RecordFactory.create(version=other_version) + + deleted = await records_ctx.delete_records(db, schema, [r1.id, r2.id, foreign.id]) + assert deleted == 2 + + _, total = await records_ctx.list_records(db, schema, offset=0, limit=10) + assert total == 1 + assert (await V2Record.get(db, r3.id)) is not None + assert (await V2Record.get(db, foreign.id)) is not None + + +async def test_list_records_by_reference_is_workspace_scoped(db): + workspace_a = await WorkspaceFactory.create() + workspace_b = await WorkspaceFactory.create() + + schema1 = await SchemaFactory.create(workspace=workspace_a) + schema2 = await SchemaFactory.create(workspace=workspace_a) + schema3 = await SchemaFactory.create(workspace=workspace_b) + version1 = await SchemaVersionFactory.create(schema=schema1, version=1) + version2 = await SchemaVersionFactory.create(schema=schema2, version=1) + version3 = await SchemaVersionFactory.create(schema=schema3, version=1) + + r1 = await V2RecordFactory.create(version=version1, reference="pmid:99") + r2 = await V2RecordFactory.create(version=version2, reference="pmid:99") + await V2RecordFactory.create(version=version3, reference="pmid:99") # workspace B + await V2RecordFactory.create(version=version1, reference="pmid:other") + + records = await records_ctx.list_records_by_reference(db, workspace_id=workspace_a.id, reference="pmid:99") + assert {r.id for r in records} == {r1.id, r2.id} From 046474a550b1d6085bd02b169fdd6fb0f7059f3e Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:15:45 +0000 Subject: [PATCH 07/12] feat(v2): record-scoped policy actions on SchemaPolicy --- .../api/policies/v1/schema_policy.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py b/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py index 15c7ba87e..c4a36f3eb 100644 --- a/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py +++ b/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py @@ -47,3 +47,27 @@ async def is_allowed(actor: User) -> bool: return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id)) return is_allowed + + # Record actions live here rather than on a new RecordPolicy: v2 records have no authz + # axis beyond their schema's workspace, and the RecordPolicy name is taken by v1. + + @classmethod + def upsert_records(cls, schema: Schema) -> PolicyAction: + async def is_allowed(actor: User) -> bool: + return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id)) + + return is_allowed + + @classmethod + def list_records(cls, schema: Schema) -> PolicyAction: + async def is_allowed(actor: User) -> bool: + return actor.is_owner or await actor.is_member(schema.workspace_id) + + return is_allowed + + @classmethod + def delete_records(cls, schema: Schema) -> PolicyAction: + async def is_allowed(actor: User) -> bool: + return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id)) + + return is_allowed From 4429338aba9befab7b7e3b43bfc2a894bea0a45f Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:19:04 +0000 Subject: [PATCH 08/12] feat(v2): /api/v2 records endpoints (bulk-upsert, list, delete) with authz --- .../extralit_server/api/schemas/v2/records.py | 7 + .../src/extralit_server/api/v2/__init__.py | 2 + .../src/extralit_server/api/v2/records.py | 87 +++++++++ .../extralit_server/contexts/v2/records.py | 4 + .../tests/integration/api/v2/test_records.py | 180 ++++++++++++++++++ 5 files changed, 280 insertions(+) create mode 100644 extralit-server/src/extralit_server/api/v2/records.py create mode 100644 extralit-server/tests/integration/api/v2/test_records.py diff --git a/extralit-server/src/extralit_server/api/schemas/v2/records.py b/extralit-server/src/extralit_server/api/schemas/v2/records.py index 1a022bee5..8b56d7990 100644 --- a/extralit-server/src/extralit_server/api/schemas/v2/records.py +++ b/extralit-server/src/extralit_server/api/schemas/v2/records.py @@ -16,6 +16,13 @@ class RecordUpsert(BaseModel): + """One bulk-upsert item. + + `fields` and `reference` are always written. `metadata` and `status` are patch-like: + when omitted (None) on an update they preserve the existing row's values (they cannot + be cleared via upsert); on insert they default to no metadata / `pending`. + """ + fields: dict[str, Any] reference: Reference external_id: str | None = None diff --git a/extralit-server/src/extralit_server/api/v2/__init__.py b/extralit-server/src/extralit_server/api/v2/__init__.py index a3972b217..c5679d131 100644 --- a/extralit-server/src/extralit_server/api/v2/__init__.py +++ b/extralit-server/src/extralit_server/api/v2/__init__.py @@ -3,6 +3,7 @@ from extralit_server._version import __version__ as extralit_version from extralit_server.api.errors.v1.exception_handlers import add_exception_handlers as add_exception_handlers_v1 from extralit_server.api.handlers.v1 import authentication as authentication_v1 +from extralit_server.api.v2 import records as records_v2 from extralit_server.api.v2 import schemas as schemas_v2 from extralit_server.errors.base_errors import __ALL__ from extralit_server.errors.error_handler import APIErrorHandler @@ -21,6 +22,7 @@ def create_api_v2() -> FastAPI: # Auth endpoints are reused from v1 so v2 tokens work identically. api_v2.include_router(authentication_v1.router) api_v2.include_router(schemas_v2.router) + api_v2.include_router(records_v2.router) return api_v2 diff --git a/extralit-server/src/extralit_server/api/v2/records.py b/extralit-server/src/extralit_server/api/v2/records.py new file mode 100644 index 000000000..0d837b043 --- /dev/null +++ b/extralit-server/src/extralit_server/api/v2/records.py @@ -0,0 +1,87 @@ +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Depends, Query, Security, status +from sqlalchemy.ext.asyncio import AsyncSession + +from extralit_server.api.policies.v1 import SchemaPolicy, authorize +from extralit_server.api.schemas.v2.records import ( + DELETE_RECORDS_LIMIT, + LIST_RECORDS_LIMIT_DEFAULT, + LIST_RECORDS_LIMIT_LE, + Records, + RecordsBulkUpsert, +) +from extralit_server.contexts import files as files_ctx +from extralit_server.contexts.v2 import records as records_ctx +from extralit_server.contexts.v2 import schemas as schemas_ctx +from extralit_server.database import get_async_db +from extralit_server.enums import V2RecordStatus +from extralit_server.errors.future import NotFoundError, UnprocessableEntityError +from extralit_server.models import User, Workspace +from extralit_server.models.v2 import Schema +from extralit_server.security import auth +from extralit_server.utils import parse_uuids + +router = APIRouter(tags=["v2: records"]) + + +async def _get_schema_or_404(db: AsyncSession, schema_id: UUID) -> Schema: + schema = await schemas_ctx.get_schema(db, schema_id) + if schema is None: + raise NotFoundError(f"Schema with id `{schema_id}` not found") + return schema + + +# AIP-136-style custom method (spec §7); Starlette treats the `:` as a literal character. +@router.post("/schemas/{schema_id}/records:bulk-upsert", response_model=Records) +async def bulk_upsert_schema_records( + *, + schema_id: UUID, + payload: RecordsBulkUpsert, + db: Annotated[AsyncSession, Depends(get_async_db)], + s3_client=Depends(files_ctx.get_s3_client), + current_user: Annotated[User, Security(auth.get_current_user)], +): + schema = await _get_schema_or_404(db, schema_id) + await authorize(current_user, SchemaPolicy.upsert_records(schema)) + workspace = await Workspace.get_or_raise(db, schema.workspace_id) + records = await records_ctx.bulk_upsert_records(db, s3_client, schema, items=payload.items, bucket=workspace.name) + return Records(items=records, total=len(records)) + + +@router.get("/schemas/{schema_id}/records", response_model=Records) +async def list_schema_records( + *, + schema_id: UUID, + db: Annotated[AsyncSession, Depends(get_async_db)], + current_user: Annotated[User, Security(auth.get_current_user)], + offset: Annotated[int, Query(ge=0)] = 0, + limit: Annotated[int, Query(ge=1, le=LIST_RECORDS_LIMIT_LE)] = LIST_RECORDS_LIMIT_DEFAULT, + status_filter: Annotated[V2RecordStatus | None, Query(alias="status")] = None, + reference: Annotated[str | None, Query()] = None, +): + schema = await _get_schema_or_404(db, schema_id) + await authorize(current_user, SchemaPolicy.list_records(schema)) + records, total = await records_ctx.list_records( + db, schema, offset=offset, limit=limit, status=status_filter, reference=reference + ) + return Records(items=records, total=total) + + +@router.delete("/schemas/{schema_id}/records", status_code=status.HTTP_204_NO_CONTENT) +async def delete_schema_records( + *, + schema_id: UUID, + db: Annotated[AsyncSession, Depends(get_async_db)], + current_user: Annotated[User, Security(auth.get_current_user)], + ids: Annotated[str, Query(description="Comma-separated record ids to delete")], +): + schema = await _get_schema_or_404(db, schema_id) + await authorize(current_user, SchemaPolicy.delete_records(schema)) + record_ids = parse_uuids(ids) + if len(record_ids) == 0: + raise UnprocessableEntityError("No record IDs provided") + if len(record_ids) > DELETE_RECORDS_LIMIT: + raise UnprocessableEntityError(f"Cannot delete more than {DELETE_RECORDS_LIMIT} records at once") + await records_ctx.delete_records(db, schema, record_ids) diff --git a/extralit-server/src/extralit_server/contexts/v2/records.py b/extralit-server/src/extralit_server/contexts/v2/records.py index 1ee56f2cc..5c889d983 100644 --- a/extralit-server/src/extralit_server/contexts/v2/records.py +++ b/extralit-server/src/extralit_server/contexts/v2/records.py @@ -40,6 +40,10 @@ async def bulk_upsert_records( Identity for updates is (schema_id, external_id); items without an external_id always insert. Each distinct schema version's body is fetched from the object store once per request, never per record. + + Update semantics are patch-like for `metadata`/`status`: an omitted (None) value + preserves the existing row's value rather than clearing it (see RecordUpsert docs); + `fields`, `reference`, and the resolved schema_version_id are always overwritten. """ default_version_id = schema.current_version_id if default_version_id is None: diff --git a/extralit-server/tests/integration/api/v2/test_records.py b/extralit-server/tests/integration/api/v2/test_records.py new file mode 100644 index 000000000..727875f6e --- /dev/null +++ b/extralit-server/tests/integration/api/v2/test_records.py @@ -0,0 +1,180 @@ +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pandera.pandas as pa +import pytest + +from extralit_server.enums import V2RecordStatus +from extralit_server.models.v2 import Schema, SchemaVersion +from tests.factories import SchemaFactory, SchemaVersionFactory, V2RecordFactory, WorkspaceUserFactory + +pytestmark = pytest.mark.asyncio + +BODY = pa.DataFrameSchema( + columns={ + "name": pa.Column(pa.String, nullable=False), + "age": pa.Column(pa.Int, nullable=True), + } +).to_json() + + +def _patch_fetch(monkeypatch, body: str = BODY) -> AsyncMock: + fetch = AsyncMock(return_value=body) + monkeypatch.setattr("extralit_server.contexts.v2.records._fetch_body_json", fetch) + return fetch + + +async def _published_schema(db) -> tuple[Schema, SchemaVersion]: + schema = await SchemaFactory.create() + version = await SchemaVersionFactory.create(schema=schema, version=1) + await schema.update(db, current_version_id=version.id) + return schema, version + + +async def test_bulk_upsert_creates_and_updates_records(async_client, owner_auth_header, db, monkeypatch): + _patch_fetch(monkeypatch) + schema, version = await _published_schema(db) + + resp = await async_client.post( + f"/api/v2/schemas/{schema.id}/records:bulk-upsert", + headers=owner_auth_header, + json={ + "items": [ + {"fields": {"name": "Ada", "age": 36}, "reference": "pmid:1", "external_id": "x-1"}, + {"fields": {"name": "Grace", "age": None}, "reference": "pmid:2"}, + ] + }, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["total"] == 2 + assert body["items"][0]["external_id"] == "x-1" + assert body["items"][0]["fields"] == {"name": "Ada", "age": 36} + assert body["items"][0]["schema_version_id"] == str(version.id) + assert body["items"][0]["status"] == V2RecordStatus.pending.value + + # Re-upsert by external_id updates in place instead of duplicating. + resp = await async_client.post( + f"/api/v2/schemas/{schema.id}/records:bulk-upsert", + headers=owner_auth_header, + json={"items": [{"fields": {"name": "Ada L.", "age": 37}, "reference": "pmid:1", "external_id": "x-1"}]}, + ) + assert resp.status_code == 200, resp.text + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records", headers=owner_auth_header) + assert resp.status_code == 200 + assert resp.json()["total"] == 2 + + +async def test_bulk_upsert_validation_failure_returns_422(async_client, owner_auth_header, db, monkeypatch): + _patch_fetch(monkeypatch) + schema, _ = await _published_schema(db) + + resp = await async_client.post( + f"/api/v2/schemas/{schema.id}/records:bulk-upsert", + headers=owner_auth_header, + json={"items": [{"fields": {"name": "Bob", "age": "not-a-number"}, "reference": "pmid:1"}]}, + ) + assert resp.status_code == 422, resp.text + assert "items[0]" in resp.json()["detail"] + assert "age" in resp.json()["detail"] + + +async def test_bulk_upsert_without_published_version_returns_422(async_client, owner_auth_header, monkeypatch): + _patch_fetch(monkeypatch) + schema = await SchemaFactory.create() + + resp = await async_client.post( + f"/api/v2/schemas/{schema.id}/records:bulk-upsert", + headers=owner_auth_header, + json={"items": [{"fields": {"name": "Ada"}, "reference": "pmid:1"}]}, + ) + assert resp.status_code == 422, resp.text + assert "no published version" in resp.json()["detail"] + + +async def test_bulk_upsert_unknown_schema_returns_404(async_client, owner_auth_header, monkeypatch): + _patch_fetch(monkeypatch) + resp = await async_client.post( + f"/api/v2/schemas/{uuid4()}/records:bulk-upsert", + headers=owner_auth_header, + json={"items": [{"fields": {"name": "Ada"}, "reference": "pmid:1"}]}, + ) + assert resp.status_code == 404, resp.text + + +async def test_list_records_paginates_and_filters(async_client, owner_auth_header, db): + schema, version = await _published_schema(db) + await V2RecordFactory.create(version=version, reference="pmid:1") + await V2RecordFactory.create(version=version, reference="pmid:1", status=V2RecordStatus.completed) + await V2RecordFactory.create(version=version, reference="pmid:2") + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records?offset=1&limit=1", headers=owner_auth_header) + assert resp.status_code == 200, resp.text + assert resp.json()["total"] == 3 + assert len(resp.json()["items"]) == 1 + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records?reference=pmid:1", headers=owner_auth_header) + assert resp.json()["total"] == 2 + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records?status=completed", headers=owner_auth_header) + assert resp.json()["total"] == 1 + + +async def test_delete_records(async_client, owner_auth_header, db): + schema, version = await _published_schema(db) + r1 = await V2RecordFactory.create(version=version) + r2 = await V2RecordFactory.create(version=version) + r3 = await V2RecordFactory.create(version=version) + + resp = await async_client.delete( + f"/api/v2/schemas/{schema.id}/records?ids={r1.id},{r2.id}", headers=owner_auth_header + ) + assert resp.status_code == 204, resp.text + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records", headers=owner_auth_header) + assert resp.json()["total"] == 1 + assert resp.json()["items"][0]["id"] == str(r3.id) + + +async def test_delete_records_validates_ids(async_client, owner_auth_header, db): + schema, version = await _published_schema(db) + + resp = await async_client.delete(f"/api/v2/schemas/{schema.id}/records?ids=", headers=owner_auth_header) + assert resp.status_code == 422 + + too_many = ",".join(str(uuid4()) for _ in range(101)) + resp = await async_client.delete(f"/api/v2/schemas/{schema.id}/records?ids={too_many}", headers=owner_auth_header) + assert resp.status_code == 422 + assert "more than" in resp.json()["detail"] + + +async def test_member_annotator_can_read_but_not_write_records( + async_client, annotator_auth_header, annotator, db, monkeypatch +): + _patch_fetch(monkeypatch) + schema, version = await _published_schema(db) + await WorkspaceUserFactory.create(workspace_id=schema.workspace_id, user_id=annotator.id) + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records", headers=annotator_auth_header) + assert resp.status_code == 200, resp.text + + resp = await async_client.post( + f"/api/v2/schemas/{schema.id}/records:bulk-upsert", + headers=annotator_auth_header, + json={"items": [{"fields": {"name": "Ada"}, "reference": "pmid:1"}]}, + ) + assert resp.status_code == 403, resp.text + + record = await V2RecordFactory.create(version=version) + resp = await async_client.delete( + f"/api/v2/schemas/{schema.id}/records?ids={record.id}", headers=annotator_auth_header + ) + assert resp.status_code == 403, resp.text + + +async def test_non_member_cannot_read_records(async_client, annotator_auth_header, db): + schema, _ = await _published_schema(db) + + resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records", headers=annotator_auth_header) + assert resp.status_code == 403, resp.text From 1d16bdd141fa2666349e1e545ac3278f0796dee6 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:21:33 +0000 Subject: [PATCH 09/12] feat(v2): GET /api/v2/references/{reference} cross-schema document view Also surfaces the specific 'No record IDs provided' 422 on empty delete ids (roborev job 70: the post-parse length check was unreachable). --- .../src/extralit_server/api/v2/records.py | 43 +++++++++- .../tests/integration/api/v2/test_records.py | 1 + .../integration/api/v2/test_references.py | 80 +++++++++++++++++++ 3 files changed, 122 insertions(+), 2 deletions(-) create mode 100644 extralit-server/tests/integration/api/v2/test_references.py diff --git a/extralit-server/src/extralit_server/api/v2/records.py b/extralit-server/src/extralit_server/api/v2/records.py index 0d837b043..76b192c17 100644 --- a/extralit-server/src/extralit_server/api/v2/records.py +++ b/extralit-server/src/extralit_server/api/v2/records.py @@ -9,8 +9,11 @@ DELETE_RECORDS_LIMIT, LIST_RECORDS_LIMIT_DEFAULT, LIST_RECORDS_LIMIT_LE, + RecordRead, Records, RecordsBulkUpsert, + ReferenceGroup, + ReferenceView, ) from extralit_server.contexts import files as files_ctx from extralit_server.contexts.v2 import records as records_ctx @@ -79,9 +82,45 @@ async def delete_schema_records( ): schema = await _get_schema_or_404(db, schema_id) await authorize(current_user, SchemaPolicy.delete_records(schema)) - record_ids = parse_uuids(ids) - if len(record_ids) == 0: + # Reject an empty param up front: parse_uuids("") would 422 with a generic + # "Invalid UUID format" before a post-parse length check could run. + if not ids.strip(): raise UnprocessableEntityError("No record IDs provided") + record_ids = parse_uuids(ids) if len(record_ids) > DELETE_RECORDS_LIMIT: raise UnprocessableEntityError(f"Cannot delete more than {DELETE_RECORDS_LIMIT} records at once") await records_ctx.delete_records(db, schema, record_ids) + + +@router.get("/references/{reference}", response_model=ReferenceView) +async def get_reference_view( + *, + reference: str, + db: Annotated[AsyncSession, Depends(get_async_db)], + current_user: Annotated[User, Security(auth.get_current_user)], + workspace_id: Annotated[UUID, Query(description="Workspace to scope the cross-schema view (required)")], +): + """The document's project-level extraction view: all v2 records across every schema in + the workspace that share this `reference` (spec §6), grouped per schema. + + An unknown reference returns an empty view (200): the reference is a free-form join + key, not an entity, so "no extractions yet" is not an error. + """ + # workspace_id is required so the view is scoped + authorized (mirrors GET /schemas). + await authorize(current_user, SchemaPolicy.list(workspace_id)) + records = await records_ctx.list_records_by_reference(db, workspace_id=workspace_id, reference=reference) + + schema_ids = {r.schema_id for r in records} + schemas_by_id = { + s.id: s for s in await schemas_ctx.list_schemas(db, workspace_id=workspace_id) if s.id in schema_ids + } + + groups = [ + ReferenceGroup( + schema_id=schema_id, + schema_name=schemas_by_id[schema_id].name, + records=[RecordRead.model_validate(r) for r in records if r.schema_id == schema_id], + ) + for schema_id in sorted(schema_ids, key=lambda schema_id: schemas_by_id[schema_id].name) + ] + return ReferenceView(reference=reference, groups=groups, total_records=len(records)) diff --git a/extralit-server/tests/integration/api/v2/test_records.py b/extralit-server/tests/integration/api/v2/test_records.py index 727875f6e..74db211ac 100644 --- a/extralit-server/tests/integration/api/v2/test_records.py +++ b/extralit-server/tests/integration/api/v2/test_records.py @@ -142,6 +142,7 @@ async def test_delete_records_validates_ids(async_client, owner_auth_header, db) resp = await async_client.delete(f"/api/v2/schemas/{schema.id}/records?ids=", headers=owner_auth_header) assert resp.status_code == 422 + assert "No record IDs provided" in resp.json()["detail"] too_many = ",".join(str(uuid4()) for _ in range(101)) resp = await async_client.delete(f"/api/v2/schemas/{schema.id}/records?ids={too_many}", headers=owner_auth_header) diff --git a/extralit-server/tests/integration/api/v2/test_references.py b/extralit-server/tests/integration/api/v2/test_references.py new file mode 100644 index 000000000..1ffe029ee --- /dev/null +++ b/extralit-server/tests/integration/api/v2/test_references.py @@ -0,0 +1,80 @@ +import pytest + +from tests.factories import ( + SchemaFactory, + SchemaVersionFactory, + V2RecordFactory, + WorkspaceFactory, + WorkspaceUserFactory, +) + +pytestmark = pytest.mark.asyncio + + +async def _schema_with_version(workspace, name: str): + schema = await SchemaFactory.create(workspace=workspace, name=name) + version = await SchemaVersionFactory.create(schema=schema, version=1) + return schema, version + + +async def test_reference_view_groups_records_per_schema_in_workspace(async_client, owner_auth_header): + workspace_a = await WorkspaceFactory.create() + workspace_b = await WorkspaceFactory.create() + + schema_pop, version_pop = await _schema_with_version(workspace_a, "population") + schema_out, version_out = await _schema_with_version(workspace_a, "outcomes") + _, version_foreign = await _schema_with_version(workspace_b, "foreign") + + r1 = await V2RecordFactory.create(version=version_pop, reference="pmid:99") + r2 = await V2RecordFactory.create(version=version_pop, reference="pmid:99") + r3 = await V2RecordFactory.create(version=version_out, reference="pmid:99") + await V2RecordFactory.create(version=version_foreign, reference="pmid:99") # workspace B + await V2RecordFactory.create(version=version_pop, reference="pmid:other") + + resp = await async_client.get( + f"/api/v2/references/pmid:99?workspace_id={workspace_a.id}", headers=owner_auth_header + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["reference"] == "pmid:99" + assert body["total_records"] == 3 + # Groups ordered by schema name: outcomes, population. + assert [g["schema_name"] for g in body["groups"]] == ["outcomes", "population"] + assert {g["schema_id"] for g in body["groups"]} == {str(schema_out.id), str(schema_pop.id)} + by_name = {g["schema_name"]: g for g in body["groups"]} + assert {r["id"] for r in by_name["population"]["records"]} == {str(r1.id), str(r2.id)} + assert [r["id"] for r in by_name["outcomes"]["records"]] == [str(r3.id)] + + +async def test_reference_view_unknown_reference_returns_empty(async_client, owner_auth_header): + workspace = await WorkspaceFactory.create() + resp = await async_client.get( + f"/api/v2/references/pmid:nope?workspace_id={workspace.id}", headers=owner_auth_header + ) + assert resp.status_code == 200, resp.text + assert resp.json() == {"reference": "pmid:nope", "groups": [], "total_records": 0} + + +async def test_reference_view_requires_workspace_id(async_client, owner_auth_header): + resp = await async_client.get("/api/v2/references/pmid:99", headers=owner_auth_header) + assert resp.status_code == 422 + + +async def test_reference_view_authz(async_client, annotator_auth_header, annotator): + workspace = await WorkspaceFactory.create() + _, version = await _schema_with_version(workspace, "population") + await V2RecordFactory.create(version=version, reference="pmid:99") + + # Non-member: forbidden. + resp = await async_client.get( + f"/api/v2/references/pmid:99?workspace_id={workspace.id}", headers=annotator_auth_header + ) + assert resp.status_code == 403, resp.text + + # Member annotator: allowed to read. + await WorkspaceUserFactory.create(workspace_id=workspace.id, user_id=annotator.id) + resp = await async_client.get( + f"/api/v2/references/pmid:99?workspace_id={workspace.id}", headers=annotator_auth_header + ) + assert resp.status_code == 200, resp.text + assert resp.json()["total_records"] == 1 From 1becffabf699fb624a0451aafad73f4a61a09dff Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:24:42 +0000 Subject: [PATCH 10/12] fix(v2): match slash-containing references (DOIs) via :path converter; lint cleanups --- .../src/extralit_server/api/v2/records.py | 4 +++- .../tests/integration/api/v2/test_records.py | 2 +- .../tests/integration/api/v2/test_references.py | 16 ++++++++++++++++ .../contexts/v2/test_records_context.py | 6 +++--- .../integration/models/v2/test_record_models.py | 2 +- .../integration/models/v2/test_schema_models.py | 4 ++-- 6 files changed, 26 insertions(+), 8 deletions(-) diff --git a/extralit-server/src/extralit_server/api/v2/records.py b/extralit-server/src/extralit_server/api/v2/records.py index 76b192c17..87e573600 100644 --- a/extralit-server/src/extralit_server/api/v2/records.py +++ b/extralit-server/src/extralit_server/api/v2/records.py @@ -92,7 +92,9 @@ async def delete_schema_records( await records_ctx.delete_records(db, schema, record_ids) -@router.get("/references/{reference}", response_model=ReferenceView) +# `:path` converter: references are free-form join keys and DOIs contain slashes +# (e.g. 10.1000/j.foo.2020.01); the default converter would 404 on them. +@router.get("/references/{reference:path}", response_model=ReferenceView) async def get_reference_view( *, reference: str, diff --git a/extralit-server/tests/integration/api/v2/test_records.py b/extralit-server/tests/integration/api/v2/test_records.py index 74db211ac..d5eda531e 100644 --- a/extralit-server/tests/integration/api/v2/test_records.py +++ b/extralit-server/tests/integration/api/v2/test_records.py @@ -138,7 +138,7 @@ async def test_delete_records(async_client, owner_auth_header, db): async def test_delete_records_validates_ids(async_client, owner_auth_header, db): - schema, version = await _published_schema(db) + schema, _ = await _published_schema(db) resp = await async_client.delete(f"/api/v2/schemas/{schema.id}/records?ids=", headers=owner_auth_header) assert resp.status_code == 422 diff --git a/extralit-server/tests/integration/api/v2/test_references.py b/extralit-server/tests/integration/api/v2/test_references.py index 1ffe029ee..c500f064b 100644 --- a/extralit-server/tests/integration/api/v2/test_references.py +++ b/extralit-server/tests/integration/api/v2/test_references.py @@ -46,6 +46,22 @@ async def test_reference_view_groups_records_per_schema_in_workspace(async_clien assert [r["id"] for r in by_name["outcomes"]["records"]] == [str(r3.id)] +async def test_reference_view_supports_slash_containing_references(async_client, owner_auth_header): + # DOIs contain slashes; the route must use the `:path` converter to match them. + workspace = await WorkspaceFactory.create() + _, version = await _schema_with_version(workspace, "population") + record = await V2RecordFactory.create(version=version, reference="10.1000/j.foo.2020.01") + + resp = await async_client.get( + f"/api/v2/references/10.1000/j.foo.2020.01?workspace_id={workspace.id}", headers=owner_auth_header + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["reference"] == "10.1000/j.foo.2020.01" + assert body["total_records"] == 1 + assert body["groups"][0]["records"][0]["id"] == str(record.id) + + async def test_reference_view_unknown_reference_returns_empty(async_client, owner_auth_header): workspace = await WorkspaceFactory.create() resp = await async_client.get( diff --git a/extralit-server/tests/integration/contexts/v2/test_records_context.py b/extralit-server/tests/integration/contexts/v2/test_records_context.py index 87c4e77f1..d10093ead 100644 --- a/extralit-server/tests/integration/contexts/v2/test_records_context.py +++ b/extralit-server/tests/integration/contexts/v2/test_records_context.py @@ -142,7 +142,7 @@ async def test_bulk_upsert_requires_published_version(db, monkeypatch): async def test_bulk_upsert_rejects_foreign_schema_version_pin(db, monkeypatch): _patch_fetch(monkeypatch) schema, _ = await _published_schema(db) - other_schema, other_version = await _published_schema(db) + _, other_version = await _published_schema(db) with pytest.raises(UnprocessableEntityError, match="does not belong"): await records_ctx.bulk_upsert_records( @@ -154,7 +154,7 @@ async def test_bulk_upsert_rejects_duplicate_external_ids_in_payload(db, monkeyp _patch_fetch(monkeypatch) schema, _ = await _published_schema(db) - with pytest.raises(UnprocessableEntityError, match="[Dd]uplicate"): + with pytest.raises(UnprocessableEntityError, match=r"[Dd]uplicate"): await records_ctx.bulk_upsert_records( db, AsyncMock(), @@ -166,7 +166,7 @@ async def test_bulk_upsert_rejects_duplicate_external_ids_in_payload(db, monkeyp async def test_list_records_paginates_and_filters(db): schema, version = await _published_schema(db) - other_schema, other_version = await _published_schema(db) + _, other_version = await _published_schema(db) r1 = await V2RecordFactory.create(version=version, reference="pmid:1") r2 = await V2RecordFactory.create(version=version, reference="pmid:1", status=V2RecordStatus.completed) diff --git a/extralit-server/tests/integration/models/v2/test_record_models.py b/extralit-server/tests/integration/models/v2/test_record_models.py index b47f004ad..95c301add 100644 --- a/extralit-server/tests/integration/models/v2/test_record_models.py +++ b/extralit-server/tests/integration/models/v2/test_record_models.py @@ -47,7 +47,7 @@ async def test_duplicate_external_id_in_schema_raises(db): ), ] ) - with pytest.raises(IntegrityError, match="v2_record_schema_id_external_id_uq|UNIQUE"): + with pytest.raises(IntegrityError, match=r"v2_record_schema_id_external_id_uq|UNIQUE"): await db.commit() diff --git a/extralit-server/tests/integration/models/v2/test_schema_models.py b/extralit-server/tests/integration/models/v2/test_schema_models.py index 6f736b647..69411c5a5 100644 --- a/extralit-server/tests/integration/models/v2/test_schema_models.py +++ b/extralit-server/tests/integration/models/v2/test_schema_models.py @@ -59,7 +59,7 @@ async def test_duplicate_schema_name_in_workspace_raises(db): Schema(name="dup", workspace_id=workspace.id), ] ) - with pytest.raises(IntegrityError, match="schema_workspace_id_name_uq|UNIQUE"): + with pytest.raises(IntegrityError, match=r"schema_workspace_id_name_uq|UNIQUE"): await db.commit() @@ -71,5 +71,5 @@ async def test_duplicate_schema_version_raises(db): SchemaVersion(schema_id=schema.id, version=1, object_key="b", etag="e", checksum="c"), ] ) - with pytest.raises(IntegrityError, match="schema_version_schema_id_version_uq|UNIQUE"): + with pytest.raises(IntegrityError, match=r"schema_version_schema_id_version_uq|UNIQUE"): await db.commit() From 006c8bd5d287688b25f85d204bd9975005a4ed4e Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Fri, 3 Jul 2026 18:26:46 +0000 Subject: [PATCH 11/12] docs: add Phase 2 (v2 records) implementation plan --- .../plans/2026-07-03-v2-records.md | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-03-v2-records.md diff --git a/docs/superpowers/plans/2026-07-03-v2-records.md b/docs/superpowers/plans/2026-07-03-v2-records.md new file mode 100644 index 000000000..f860a88ad --- /dev/null +++ b/docs/superpowers/plans/2026-07-03-v2-records.md @@ -0,0 +1,60 @@ +# v2 Records Implementation Plan (Phase 2 of 6) + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Executed 2026-07-03 on branch `feat/v2-records` (stacked on `feat/v2-schema-registry`, PR #219). + +**Goal:** Add the v2 `record` entity — one typed row pinned to a `schema_version`, carrying the cross-schema `reference` join key and Pandera-validated `fields` JSONB — with a validated bulk-upsert, paginated listing, bulk delete, and the `GET /references/{reference}` cross-schema document view (spec §5 record row, §6 write flow, §7 API surface). + +**Out of scope:** LanceDB search/sync (Phase 3), questions/suggestions/responses (Phase 4). + +## Architecture + +Extends the isolated v2 module set from Phase 1 (`models/v2/`, `contexts/v2/`, `api/v2/`, `api/schemas/v2/`). Postgres is the source of truth. On bulk-upsert, the context resolves each item's `schema_version_id` (default = `schema.current_version_id`), fetches each **distinct** version's Pandera body from the object store **once per request** (dict keyed by version id, via `contexts.files.get_object` pinned to the stored S3 `object_version_id`), validates every item via Phase 1's `validate_record_fields` *before any write* (all-or-nothing 422), then persists v1-style (fetch-existing-by-external_id → merge → flush → single commit). + +## Key decisions (as built) + +| Decision | Choice | +|---|---| +| Table name | `v2_records` — v1 owns `records`; renamed at Phase 6 retirement | +| ORM class | `V2Record` (not `Record`): a second declarative class named `Record` breaks v1's string-based `relationship("Record")` lookups in the shared registry. `models/v2/__init__` also exports it as `Record` for v2-namespace ergonomics; `models/__init__` registers `V2Record` | +| Status enum | New `V2RecordStatus` (`pending\|completed\|discarded`), PG type `v2_record_status_enum` — v1's `record_status_enum` untouched | +| Bulk route | `POST /api/v2/schemas/{schema_id}/records:bulk-upsert` (spec §7, AIP-136 style; Starlette treats `:` as literal) → `{items, total}`, 200 | +| List route | `GET /api/v2/schemas/{schema_id}/records?offset&limit&status&reference` → `{items, total}` (50 default / 1000 max, exact total) | +| Delete route | `DELETE /api/v2/schemas/{schema_id}/records?ids=` → 204; empty param → specific 422; cap 100 | +| References | `GET /api/v2/references/{reference:path}?workspace_id=` — `:path` converter because DOIs contain slashes; unknown reference → 200 empty view (a reference is a join key, not an entity); required `workspace_id` mirrors `GET /schemas` scoping | +| Upsert identity | `(schema_id, external_id)`, nullable `external_id` always inserts; duplicates within one payload → 422. Fetch-then-merge (not `upsert_many`): NULLs defeat ON CONFLICT, and validation must run before any write | +| Update semantics | Patch-like for `metadata`/`status` (omitted preserves existing; documented on `RecordUpsert` and the context); `fields`/`reference`/version pin always overwritten | +| Policies | `SchemaPolicy.{upsert_records,list_records,delete_records}` (writes: owner or admin member; reads: member); references view reuses `SchemaPolicy.list(workspace_id)`. `RecordPolicy` name belongs to v1 | +| FKs | `schema_id` and `schema_version_id` both `ondelete=CASCADE` | +| `metadata` column | Attr `metadata_` → column `"metadata"` (v1 pattern); `RecordRead` accepts either via `AliasChoices`, serializes as `metadata` | + +## Tasks (all completed, TDD: red → green → commit) + +1. **`V2RecordStatus` enum** — `tests/integration/test_enums_v2.py`; `enums.py`. `feat(v2): add V2RecordStatus enum` +2. **`V2Record` ORM model** — `models/v2/records.py`, exports, aliased registration in `models/__init__.py`; `tests/integration/models/v2/test_record_models.py` (defaults, uniq constraint, NULL external_ids don't collide). `feat(v2): V2Record ORM model (v2_records) pinned to schema versions` +3. **Alembic migration** — `8136bc88ee3a` (down_revision `9f3010c649c8`): table, enum, FKs, uniq, 4 indexes; upgrade→downgrade→upgrade round-trip verified. `feat(v2): alembic migration for v2_records table` +4. **`V2RecordFactory`** — SubFactory(SchemaVersionFactory) with `_create` override awaiting the version to wire `schema_id`/`schema_version_id` (LazyAttribute sees a coroutine). `test(v2): V2RecordFactory` +5. **Pydantic schemas** — `api/schemas/v2/records.py`: limits (500 bulk / 50-1000 list / 100 delete), `RecordUpsert`, `RecordsBulkUpsert`, `RecordRead`, `Records`, `ReferenceGroup`+`ReferenceView` (flattened `schema_id`/`schema_name` to avoid `BaseModel.schema` shadowing). `feat(v2): pydantic request/response schemas for v2 records and references` +6. **Records context** — `contexts/v2/records.py`: `_fetch_body_json` (the test seam; monkeypatched in tests), `bulk_upsert_records`, `list_records`, `delete_records` (schema-scoped), `list_records_by_reference` (workspace-scoped join). 10 tests in `tests/integration/contexts/v2/test_records_context.py`. `feat(v2): records context with validated bulk-upsert, list, delete, reference query` +7. **Policy actions** — `SchemaPolicy` additions. `feat(v2): record-scoped policy actions on SchemaPolicy` +8. **Records router** — `api/v2/records.py` + mount in `create_api_v2()`; 9 API tests incl. member/non-member authz negatives and 422 validation detail. `feat(v2): /api/v2 records endpoints (bulk-upsert, list, delete) with authz` +9. **References endpoint** — same router; 5 tests incl. DOI (slash) reference and workspace scoping. `feat(v2): GET /api/v2/references/{reference} cross-schema document view` +10. **Gate** — 55 v2 tests green; `configure_mappers()` + app-import smoke (v1/v2 registry coexistence); ruff clean except pre-existing `helpers.py` ASYNC240 (untouched by this branch). + +## Review findings addressed (roborev jobs 63–71) + +- **HIGH missing migration (job 64)** — migration landed in the adjacent commit (same pattern as Phase 1); `__repr__` now says `V2Record(...)`. +- **LOW upsert asymmetry (job 68)** — patch-like preserve-on-omit kept intentionally; contract documented on `RecordUpsert` and the context docstring. +- **LOW dead empty-ids branch (job 70)** — empty `ids=` rejected before `parse_uuids`, so the specific "No record IDs provided" 422 surfaces; asserted in tests. +- **MEDIUM slash references unreachable (job 71)** — `{reference:path}` converter + DOI test. + +## Accepted risks + +- Concurrent bulk-upserts racing on `(schema_id, external_id)` can raise IntegrityError (500) — same class as v1's behavior and Phase 1's accepted publish race; hardened when record writes move to the job queue. +- `:` custom-verb route is Starlette-safe; if a proxy ever mangles it, fall back to `/records/bulk` (one-line change). + +## Downstream phases (unchanged from Phase 1 plan) + +3. **LanceDB index** — `index/` engine, inline sync on record write, `:search`/similarity; delete `search_engine/`. +4. **Annotation v2** — `question` (column binding), `suggestion`, `response`. +5. **Queue** — `queue`/`queue_item`/`queue_assignment`, `GET /queues/{id}/next`. +6. **Migrator + retirement** — v1→v2 per-dataset migrator; delete v1 tables/handlers. From 62ecf5c1e1ed0398e136fef853a0df96c90c0453 Mon Sep 17 00:00:00 2001 From: JonnyTran Date: Mon, 6 Jul 2026 23:07:07 -0700 Subject: [PATCH 12/12] docs: update schema-centric data model design with new v2 table definitions and record write process - Clarified table names and purposes for `schemas`, `schema_versions`, and `v2_records`. - Enhanced the description of the record write process, emphasizing validation and error handling. - Updated API endpoint descriptions for schemas, schema versions, and records to reflect recent changes. --- ...-06-27-schema-centric-data-model-design.md | 84 +++++++++++++++---- 1 file changed, 67 insertions(+), 17 deletions(-) diff --git a/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md b/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md index 203681f3e..38084a522 100644 --- a/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md +++ b/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md @@ -84,14 +84,14 @@ Three stores, clear roles: ## 5. Relational schema (new v2 tables, additive Alembic) -Distinct/`v2_`-prefixed names during the parallel phase; renamed to canonical names on -v1 retirement. +As built: tables `schemas`, `schema_versions`, `v2_records` (v1 owns `records`); renamed +to canonical names on v1 retirement. | Table | Purpose | Key columns | |---|---|---| -| `schema` | **Core entity** = extraction table (≙ "Dataset" in UI). Registry pointer to the Pandera body. | `id`, `workspace_id`→workspaces, `name`, `kind` (`singleton`\|`table`), `status` (`draft`\|`published`), `current_version_id`→schema_version (nullable until first publish), `settings` JSONB (guidelines, etc.), `inserted_at`, `updated_at`; uniq(`workspace_id`,`name`) | -| `schema_version` | Immutable version → object-store body + lineage + denormalized column cache. | `id`, `schema_id`, `version`, `object_key`, `object_version_id`, `etag`, `checksum`, `parent_version_id` (lineage, nullable), `columns_cache` JSONB (per-column name/dtype/nullable/review-widget, derived from the S3 body for fast validation + UI), `created_by`→users, `inserted_at`; uniq(`schema_id`,`version`) | -| `record` | One typed row; pins its version; carries the cross-schema join key. | `id`, `schema_id`, `schema_version_id` (pinned), `reference` (indexed), `external_id` (nullable), `fields` JSONB (cells keyed by column), `metadata` JSONB, `status` (`pending`\|`completed`\|`discarded`), `inserted_at`, `updated_at`; idx(`schema_id`,`reference`), idx(`reference`); uniq(`schema_id`,`external_id`) | +| `schema` (`schemas`) | **Core entity** = extraction table (≙ "Dataset" in UI). Registry pointer to the Pandera body. Singleton-vs-table is *emergent* from question/column bindings, not a stored kind (§14). | `id`, `workspace_id`→workspaces, `name`, `status` (`draft`\|`published`), `current_version_id`→schema_version (nullable until first publish), `settings` JSONB (guidelines, etc.), `inserted_at`, `updated_at`; uniq(`workspace_id`,`name`) | +| `schema_version` (`schema_versions`) | Immutable version → object-store body + lineage + denormalized column cache. | `id`, `schema_id`, `version` (monotonic int), `object_key`, `object_version_id`, `etag`, `checksum`, `parent_version_id` (lineage, nullable), `columns_cache` JSONB (per-column name/dtype/nullable/review-widget, derived from the S3 body — UI/inspection cache, not a validation source), `review_widgets` JSONB (out-of-band per-column widget overlay, §13), `created_by`→users, `inserted_at`; uniq(`schema_id`,`version`) | +| `record` (`v2_records`, ORM `V2Record`) | One typed row; pins its version; carries the cross-schema join key. | `id`, `schema_id` (CASCADE), `schema_version_id` (pinned, CASCADE), `reference` (indexed), `external_id` (nullable), `fields` JSONB (cells keyed by column; permissive — unknown fields pass through, §14), `metadata` JSONB, `status` (`pending`\|`completed`\|`discarded`, PG enum `v2_record_status_enum`), `inserted_at`, `updated_at`; idx(`schema_id`,`reference`), idx(`reference`); uniq(`schema_id`,`external_id`) | | `question` | Review config bound to column(s): 1 normally, N if `type=table`. | `id`, `schema_id`, `name`, `title`, `description`, `type` (`text`\|`label`\|`multi_label`\|`rating`\|`ranking`\|`span`\|`table`), `columns` JSONB (bound column names), `settings` JSONB, `required`, `inserted_at`, `updated_at`; uniq(`schema_id`,`name`) | | `suggestion` | LLM output per (record, question). | `id`, `record_id`, `question_id`, `value` JSONB, `score` JSONB (float \| float[]), `agent`, `type`, `inserted_at`, `updated_at`; uniq(`record_id`,`question_id`) | | `response` | Human review per (record, user); `values` keyed by question/column → resolves to cells; multiple users per record = overlap. | `id`, `record_id`, `user_id`, `values` JSONB, `status` (`draft`\|`submitted`\|`discarded`), `inserted_at`, `updated_at`; uniq(`record_id`,`user_id`) | @@ -113,11 +113,18 @@ schema body + `columns_cache`). `parent_version_id`, and advances `schema.current_version_id`. The schema's LanceDB table is created or **evolved** to the new column superset. -**Record write (bulk upsert).** Resolve the record's `schema_version_id` (default = -`current_version_id`). Validate `fields` against that version's Pandera schema -(coerce + check) using `columns_cache` (fall back to fetching the S3 body when needed). -Persist to Postgres (truth). Then **sync** the row into the schema's LanceDB table -(record_id, reference, schema_version_id, status, column values, embeddings, metadata). +**Record write (bulk upsert).** Resolve each record's `schema_version_id` (default = +`current_version_id`). Validation always fetches the Pandera body from the object store +(pinned to the stored S3 `object_version_id`), once per distinct version per request — +Pandera checks (ranges, regexes) exist only in the body, so `columns_cache` is a +UI/inspection cache, not a validation source. Every item is validated (coerce + check) +*before any write*, so one invalid item fails the whole request (all-or-nothing 422). +Non-null fields absent from the schema pass through to `record.fields` unvalidated — +permissive JSONB is intentional (§14). Persist to Postgres (truth). Then **sync** the +row into the schema's LanceDB table (record_id, reference, schema_version_id, status, +column values, embeddings, metadata): Postgres commits first, Lance sync is best-effort, +and `rebuild(schema_id)` is the recovery path — search is eventually consistent by +design. Validation-throughput optimization is deferred (§14 future-work ledger). **Index engine.** New `index/` package: a LanceDB engine exposing `upsert_records`, `delete_records`, `search` (scalar filter + FTS), `similarity_search` (vector), and @@ -133,11 +140,16 @@ extraction view that the Queue UI renders. - **Schemas (= Datasets):** `POST /schemas`, `GET /schemas`, `GET /schemas/{id}`, `PUT /schemas/{id}`, `DELETE /schemas/{id}`; `GET /schemas/{id}/columns`. - **Schema versions:** `POST /schemas/{id}/versions` (register body + evolve Lance - table), `GET /schemas/{id}/versions`, `GET /schemas/{id}/versions/{version}`. -- **Records:** `POST /schemas/{id}/records:bulk-upsert`, `GET /schemas/{id}/records` - (filter/paginate), `DELETE /schemas/{id}/records`, - `POST /schemas/{id}/records:search` (LanceDB: text / vector / scalar filter). -- **References:** `GET /references/{reference}` (cross-schema document view). + table), `GET /schemas/{id}/versions`, `GET /schemas/{id}/versions/{version}` + (single-version read: unimplemented; the Phase-4 annotation UI needs it to render + old-version shapes — see §14 future-work ledger). +- **Records:** `POST /schemas/{id}/records:bulk-upsert` (AIP-136 custom method; ≤500 + items), `GET /schemas/{id}/records?offset&limit&status&reference` (50 default / + 1000 max, exact total), `DELETE /schemas/{id}/records?ids=` (≤100, 204), + `POST /schemas/{id}/records:search` (Phase 3, LanceDB: text / vector / scalar filter). +- **References:** `GET /references/{reference:path}?workspace_id=` (cross-schema + document view; `:path` because DOIs contain slashes; unknown reference → empty 200 — + a reference is a join key, not an entity). - **Questions:** `POST|GET|PUT|DELETE /schemas/{id}/questions`. - **Annotation:** `PUT /records/{id}/suggestions`, `PUT /records/{id}/responses`. - **Queues:** `POST|GET|PUT|DELETE /queues`; `GET /queues/{id}/next` (next reference @@ -219,8 +231,8 @@ retirements; job-queue offload of index sync; RAG/chat rewrite. ## 12. Open items to resolve during planning -- Exact `version` format (monotonic int vs semver) and whether a draft version can be - edited in place before publish. +- Whether a draft version can be edited in place before publish. (The `version` format + half of this item is resolved: monotonic int, implemented in Phase 1.) - `reference` derivation rules during migration when no `documents` link exists. - Whether `response.values` stays keyed-by-question (v1-style) or moves per-question rows — current design keeps the v1-style keyed map for minimal churn. @@ -261,3 +273,41 @@ retirements; job-queue offload of index sync; RAG/chat rewrite. - **Models + their migration live in adjacent commits (NOTED).** The roborev per-commit review of the models commit flagged the absence of the table migration in that same commit; the migration lands in the immediately following commit, and the branch is internally consistent and green. + +## 14. Resolved during Phase 2 implementation & 2026-07-06 spec review + +- **`kind` (`singleton`|`table`) removed as a schema concept (DECIDED 2026-07-06).** + Singleton behavior is not a special schema type: a record whose questions are all + non-table is *implicitly* one row per `reference`. The distinction is emergent from + question/column bindings (Phase 4), not a stored discriminator, so nothing enforces + "one row per reference" at the record layer. Code follow-up: drop `schemas.kind` + + `SchemaKind` (small migration) before Phase 4 builds question bindings on top. +- **Unknown-field passthrough is intentional (DECIDED 2026-07-06).** `record.fields` + keeps permissive-JSONB semantics: non-null fields absent from the Pandera schema pass + through unvalidated. No strict/reject mode. +- **Concurrency races stay accepted (REAFFIRMED 2026-07-06).** Publish `max(version)+1` + and concurrent bulk-upserts on `(schema_id, external_id)` can 500 on unique-constraint + collision; hardened when record writes move to the job queue. +- **As-built naming.** Tables `schemas` / `schema_versions` / `v2_records`; ORM class + `V2Record` (a second declarative class named `Record` breaks v1's string-based + `relationship("Record")` lookups in the shared registry); PG enum + `v2_record_status_enum`. Canonical names return at Phase 6 retirement. Full Phase 2 + decision table: `docs/superpowers/plans/2026-07-03-v2-records.md`. +- **Upsert update semantics.** Patch-like for `metadata`/`status` (omitted preserves the + existing value); `fields`/`reference`/version pin always overwritten. + +### Future work ledger (after the extraction-records model stabilizes) + +- **Documents import → `reference` mapping (needs its own spec + PRD).** How PDF(s) / + imported documents map to the `reference` join key, beyond what the current documents + import provides. +- **`reference` normalization.** A canonicalization rule for the join key (trim, DOI + case-folding) applied on the write path — needed before the Queue (Phase 5) walks + references, since `10.1000/ABC` vs `10.1000/abc` are currently distinct documents. +- **Question↔column binding validation.** Publish-time validation of `question.columns` + against the new version's `columns_cache`; bindings dangle on column rename/drop. +- **Validation throughput.** Per-request S3 body fetch is fine for now; later options: + Pandera lazy/batched validation or ibis + DuckDB backend with incremental loading + (an immutable-version body cache is the cheap intermediate step). +- **`GET /schemas/{id}/versions/{version}`.** Specced, unimplemented; the Phase-4 + annotation UI will need it to render records pinned to old versions.