diff --git a/README.md b/README.md index 18a729b..1571fce 100644 --- a/README.md +++ b/README.md @@ -5,21 +5,9 @@ mapping workflows. ## Current status -The package provides source-independent Pydantic contracts, SQLAlchemy models, -and a resumable `MappingStore` for runs, inputs, candidates, separate evidence -records, versioned decisions, decision history, provenance, and lifecycle -state. It can use an explicit SQLAlchemy URL for tests or resolve a named -database resource through `oa-configurator` for consuming packages. +The package provides source-independent Pydantic contracts, SQLAlchemy models, and a resumable `MappingStore` for runs, inputs, candidates, separate evidence records, versioned decisions, decision history, provenance, and lifecycle state. It can use an explicit SQLAlchemy URL for tests or resolve a named database resource through `oa-configurator` for consuming packages. -`MappingReadContext` exposes common read-only status, coverage, evidence-packet, -and review-handoff operations. Packets are JSON-safe and carry the full run, -input, candidate, evidence, and decision-history lineage, so Groundworkers or a -standalone reviewer can consume the same handoff without importing SQLAlchemy -models. - -Representative contract fixtures under `tests/fixtures/` cover one-to-one, -one-to-many, ambiguous, redirected/incomplete, unmappable, and retryable -failure outcomes. +`MappingReadContext` exposes common read-only status, coverage, paginated review summaries, evidence-packet, and review-handoff operations. Packets are JSON-safe and carry the full run, input, candidate, evidence, and decision-history lineage, so Groundworkers or a standalone reviewer can consume the same handoff without importing SQLAlchemy models. Review pages are query-level paginated and accept source-defined decision-status filters; `pending` is synthesized for inputs without a decision. ## Development @@ -29,6 +17,3 @@ uv run ty check src/ uv run ruff check . uv run pytest -q ``` - -The package version is derived from a `vX.Y.Z` or `X.Y.Z` Git tag, with a -`0.1.0` fallback when Git metadata is unavailable. diff --git a/src/groundstore/__init__.py b/src/groundstore/__init__.py index b01280a..775ae7f 100644 --- a/src/groundstore/__init__.py +++ b/src/groundstore/__init__.py @@ -1,6 +1,11 @@ """Shared mapping-task contracts and persistence for Groundworkers workflows.""" -from .context import MappingEvidencePacket, MappingReadContext, MappingReviewHandoff +from .context import ( + MappingEvidencePacket, + MappingReadContext, + MappingReviewHandoff, + MappingReviewPage, +) from .contracts import ( DecisionStatus, LifecycleStatus, @@ -23,6 +28,7 @@ "MappingInputSpec", "MappingReadContext", "MappingReviewHandoff", + "MappingReviewPage", "MappingRunSpec", "MappingStore", "create_groundstore_engine", diff --git a/src/groundstore/context.py b/src/groundstore/context.py index 721232b..866ae33 100644 --- a/src/groundstore/context.py +++ b/src/groundstore/context.py @@ -53,6 +53,21 @@ class MappingReviewHandoff(BaseModel): metadata: dict[str, Any] = Field(default_factory=dict) +class MappingReviewPage(BaseModel): + """Stable, paginated summary for selecting mapping inputs to review.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: str = "groundstore.mapping-review-page.v1" + run: dict[str, Any] | None = None + page: int = Field(ge=1) + page_size: int = Field(ge=1) + page_count: int = Field(ge=0) + total_items: int = Field(ge=0) + decision_status: str | None = None + items: list[dict[str, Any]] = Field(default_factory=list) + + @dataclass(frozen=True, slots=True) class MappingReadContext: """Read-only Groundstore façade for host tools and review consumers.""" @@ -128,6 +143,73 @@ def evidence_packet(self, input_id: str) -> MappingEvidencePacket: decision=decision_payload, ) + def review_page( + self, + source_namespace: str, + *, + run_id: str | None = None, + target_system: str | None = None, + page: int = 1, + page_size: int = 20, + decision_status: str | None = None, + ) -> MappingReviewPage: + """Return a database-paginated review summary for one mapping run.""" + if page < 1: + raise ValueError("page must be positive") + if page_size < 1: + raise ValueError("page_size must be positive") + + run = self.store.get_run(run_id) if run_id is not None else None + if run_id is None: + run = self.store.latest_successful_run( + source_namespace, target_system=target_system + ) + if run is None: + return MappingReviewPage( + run=None, + page=page, + page_size=page_size, + page_count=0, + total_items=0, + decision_status=decision_status, + ) + if run.source_namespace != source_namespace: + raise ValueError(f"mapping run {run.id} does not belong to {source_namespace}") + if target_system is not None and run.target_system != target_system: + raise ValueError(f"mapping run {run.id} does not target {target_system}") + + records, total = self.store.get_review_inputs( + run.id, + offset=(page - 1) * page_size, + limit=page_size, + decision_status=decision_status, + ) + page_count = (total + page_size - 1) // page_size + return MappingReviewPage( + run=_run_payload(run), + page=page, + page_size=page_size, + page_count=page_count, + total_items=total, + decision_status=decision_status, + items=[ + { + "input_id": input_record.id, + "source_namespace": input_record.source_namespace, + "source_kind": input_record.source_kind, + "source_key": input_record.source_key, + "lifecycle_status": input_record.lifecycle_status, + "decision_status": status, + "candidate_count": len(candidates), + "candidates": [ + _review_candidate_payload(candidate) for candidate in candidates + ], + "normalized_projection": input_record.normalized_projection, + } + for input_record, status, candidates in records + ], + ) + def review_handoff( self, input_id: str, @@ -207,6 +289,26 @@ def _candidate_payload(candidate: MappingCandidate) -> dict[str, Any]: } +def _review_candidate_payload(candidate: MappingCandidate) -> dict[str, Any]: + payload = _candidate_payload(candidate) + return { + key: payload[key] + for key in ( + "id", + "target_namespace", + "target_vocabulary_id", + "target_concept_id", + "target_code", + "target_grain", + "target_role", + "method", + "rank", + "score", + "confidence", + ) + } + + def _evidence_payload(evidence: MappingEvidence) -> dict[str, Any]: return { "id": evidence.id, diff --git a/src/groundstore/store.py b/src/groundstore/store.py index 86ab990..b72ec6d 100644 --- a/src/groundstore/store.py +++ b/src/groundstore/store.py @@ -10,8 +10,8 @@ from uuid import uuid4 from oa_configurator import ResolvedDatabase, Resolver -from sqlalchemy import Engine, select -from sqlalchemy.orm import sessionmaker +from sqlalchemy import Engine, and_, func, select +from sqlalchemy.orm import aliased, sessionmaker from .contracts import ( MappingCandidateSpec, @@ -362,6 +362,105 @@ def get_inputs(self, run_id: str) -> list[MappingInput]: with self._session_factory() as session: return list(session.scalars(select(MappingInput).where(MappingInput.run_id == run_id))) + def get_review_inputs( + self, + run_id: str, + *, + offset: int = 0, + limit: int = 20, + decision_status: str | None = None, + ) -> tuple[list[tuple[MappingInput, str, list[MappingCandidate]]], int]: + """Return one stable, database-paginated review page. + + The synthetic ``pending`` status represents an input without a + decision. Other statuses are source-independent strings so adapters + may add meaningful outcomes without changing Groundstore's contract. + """ + if offset < 0: + raise ValueError("offset cannot be negative") + if limit < 1: + raise ValueError("limit must be positive") + + latest_versions = ( + select( + MappingDecision.input_id, + func.max(MappingDecision.decision_version).label("decision_version"), + ) + .group_by(MappingDecision.input_id) + .subquery() + ) + latest_decision = aliased(MappingDecision) + decision_status_expression = func.coalesce(latest_decision.decision_status, "pending") + with self._session_factory() as session: + count_query = ( + select(func.count(MappingInput.id)) + .select_from(MappingInput) + .outerjoin( + latest_versions, + latest_versions.c.input_id == MappingInput.id, + ) + .outerjoin( + latest_decision, + and_( + latest_decision.input_id == MappingInput.id, + latest_decision.decision_version + == latest_versions.c.decision_version, + ), + ) + .where(MappingInput.run_id == run_id) + ) + page_query = ( + select(MappingInput, decision_status_expression) + .select_from(MappingInput) + .outerjoin( + latest_versions, + latest_versions.c.input_id == MappingInput.id, + ) + .outerjoin( + latest_decision, + and_( + latest_decision.input_id == MappingInput.id, + latest_decision.decision_version + == latest_versions.c.decision_version, + ), + ) + .where(MappingInput.run_id == run_id) + .order_by( + MappingInput.source_kind, + MappingInput.source_key, + MappingInput.id, + ) + .offset(offset) + .limit(limit) + ) + if decision_status is not None: + count_query = count_query.where(decision_status_expression == decision_status) + page_query = page_query.where(decision_status_expression == decision_status) + + total = int(session.scalar(count_query) or 0) + rows = list(session.execute(page_query)) + input_ids = [input_record.id for input_record, _ in rows] + candidates_by_input: dict[str, list[MappingCandidate]] = { + input_id: [] for input_id in input_ids + } + if input_ids: + candidates = session.scalars( + select(MappingCandidate) + .where(MappingCandidate.input_id.in_(input_ids)) + .order_by(MappingCandidate.input_id, MappingCandidate.rank) + ) + for candidate in candidates: + candidates_by_input[candidate.input_id].append(candidate) + + return [ + ( + input_record, + str(status), + candidates_by_input[input_record.id], + ) + for input_record, status in rows + ], total + def latest_successful_run( self, source_namespace: str, *, target_system: str | None = None ) -> MappingRun | None: diff --git a/tests/test_context.py b/tests/test_context.py index b7a213f..f8cd6a1 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -79,3 +79,75 @@ def test_read_context_builds_json_safe_packet_and_review_handoff(store): assert handoff.source_namespace == "pbs" assert handoff.metadata == {"queue": "pbs"} assert handoff.model_dump(mode="json")["packet"] == payload + + +def test_read_context_paginates_review_inputs_and_filters_latest_status(store): + run = store.get_or_create_run( + MappingRunSpec( + source_namespace="pbs", + source_fingerprint="snapshot-page", + target_system="omop", + algorithm_version="mapper-1", + policy_version="policy-1", + ) + ) + for key in ("B", "A", "C"): + input_record = store.upsert_input( + run.id, + MappingInputSpec( + source_namespace="pbs", + source_kind="drug", + source_key=key, + source_fingerprint=f"input-{key}", + normalized_projection={"name": key}, + ), + ) + candidate = store.upsert_candidate( + input_record.id, + MappingCandidateSpec( + target_namespace="omop", + target_vocabulary_id="RxNorm", + target_concept_id=key, + target_grain="ingredient", + method="exact", + rank=1, + ), + ) + store.record_decision( + input_record.id, + MappingDecisionSpec( + decision_status=( + DecisionStatus.AMBIGUOUS if key != "B" else DecisionStatus.MAPPED + ), + selected_candidate_ids=[candidate.id], + outcome_code="pbs.test", + ), + ) + store.upsert_input( + run.id, + MappingInputSpec( + source_namespace="pbs", + source_kind="drug", + source_key="D", + source_fingerprint="input-D", + normalized_projection={"name": "D"}, + ), + ) + store.update_run(run.id, lifecycle_status="complete") + + context = MappingReadContext(store) + page = context.review_page( + "pbs", run_id=run.id, page=1, page_size=1, decision_status="ambiguous" + ) + payload = page.model_dump(mode="json") + + assert payload["schema_version"] == "groundstore.mapping-review-page.v1" + assert payload["total_items"] == 2 + assert payload["page_count"] == 2 + assert payload["items"][0]["source_key"] == "A" + assert payload["items"][0]["candidate_count"] == 1 + assert payload["items"][0]["decision_status"] == "ambiguous" + + pending = context.review_page("pbs", run_id=run.id, decision_status="pending") + assert pending.total_items == 1 + assert pending.items[0]["source_key"] == "D"