Completion

README.md

@@ -9,12 +9,14 @@ uv sync --extra dev
 uv run omop-config configure pbs_client
 uv run pbs-client init-db
 uv run pbs-client sync
+uv run pbs-client verify
 ```
 
-The package registers `PBSClientConfig` with `oa-configurator` under the
-`pbs_client` tool name. The configuration wizard creates a `[tools.pbs_client]`
-section and a named generic database entry (`pbs_db`) that can be shared with
-downstream packages such as Groundworkers.
+`verify` is a read-only check that each recorded sync has written at least the API-reported `total_records`; it compares checkpoint writes, not distinct table row counts, because some resources intentionally upsert duplicate keys.
+
+In an interactive terminal, `sync` also shows a per-resource record progress bar from the API totals, advancing after each page is committed.
+
+The package registers `PBSClientConfig` with `oa-configurator` under the `pbs_client` tool name. The configuration wizard creates a `[tools.pbs_client]` section and a named generic database entry (`pbs_db`) that can be shared with downstream packages such as Groundworkers.
 
 For a non-interactive local SQLite setup, the relevant stack configuration is:
 
@@ -32,12 +34,8 @@ pbs_db = "pbs_db"
 subscription_key = "your-subscription-key"
 ```
 
-Set `OA_CONFIG_PATH` before invoking the commands when using a config file
-outside `~/.config/omop/config.toml`.
+Set `OA_CONFIG_PATH` before invoking the commands when using a config file outside `~/.config/omop/config.toml`.
 
 The public API is deliberately rate limited to one request per twenty seconds. The client enforces that interval process-wide, including retries and page continuations, and rejects configured intervals below three seconds because the quota is shared across users. The default page size is 5,000 records: this keeps large responses manageable without creating unnecessary calls against the shared quota. Use `--limit 1000` when an endpoint still returns an empty or non-JSON response; if a page-size change is made during a resume, the affected resource safely restarts from page one. Refreshes are upserts and intentionally retain rows no longer returned by a later response, preserving local PBS history. Tests use local fixtures and never call the API.
 
-All configuration — the subscription key, base URL, rate limit, and the
-shared mirror database — is read from `oa-configurator`. There is no
-environment-variable or CLI-flag fallback; run `uv run omop-config configure
-pbs_client` before using the library or CLI.
+All configuration — the subscription key, base URL, rate limit, and the shared mirror database — is read from `oa-configurator`. There is no environment-variable or CLI-flag fallback; run `uv run omop-config configure pbs_client` before using the library or CLI.

src/pbs_client/cli/main.py

@@ -12,7 +12,7 @@
 from pbs_client.db import init_db, make_session_factory
 from pbs_client.errors import PBSHTTPError, PBSInvalidResponseError, PBSSyncError, PBSTransportError
 from pbs_client.http import DEFAULT_PAGE_SIZE, PBSClient
-from pbs_client.sync import SyncOrchestrator, mirror_status
+from pbs_client.sync import SyncOrchestrator, mirror_status, sync_integrity_issues
 
 app = typer.Typer(help="Maintain a local offline mirror of the PBS Public Data API v3.")
 
@@ -203,5 +203,31 @@ def status() -> None:
             typer.echo(line)
 
 
+@app.command()
+def verify() -> None:
+    """Check that recorded sync writes are not below API-reported totals."""
+
+    _, engine, _ = _runtime()
+    with make_session_factory(engine)() as session:
+        issues = sync_integrity_issues(session)
+    if not issues:
+        typer.echo("PBS mirror verification passed: no recorded resource is below its API total.")
+        return
+
+    typer.echo(
+        f"PBS mirror verification failed for {len(issues)} resource(s):",
+        err=True,
+    )
+    for state in issues:
+        total_records = state.metadata_json["total_records"]
+        missing_records = total_records - state.records_written
+        typer.echo(
+            f"  {state.resource}: {state.records_written:,}/{total_records:,} "
+            f"records written (missing {missing_records:,}; status={state.status})",
+            err=True,
+        )
+    raise typer.Exit(code=1)
+
+
 def main() -> None:
     app()

src/pbs_client/db/engine.py

@@ -5,7 +5,6 @@
 from collections.abc import Iterator
 from contextlib import contextmanager
 
-import sqlalchemy as sa
 from orm_loader.backends.resolve import resolve_backend
 from orm_loader.helpers.bulk import engine_with_replica_role
 from sqlalchemy import Engine, event
@@ -14,8 +13,26 @@
 from pbs_client.db.model import Base
 
 
+def _set_sqlite_foreign_keys(dbapi_connection, enabled: bool) -> int:
+    """Set this connection's FK flag outside any SQLite transaction."""
+
+    autocommit = dbapi_connection.autocommit
+    dbapi_connection.rollback()
+    dbapi_connection.autocommit = True
+    try:
+        value = "ON" if enabled else "OFF"
+        dbapi_connection.execute(f"PRAGMA foreign_keys = {value}").close()
+        cursor = dbapi_connection.execute("PRAGMA foreign_keys")
+        try:
+            return cursor.fetchone()[0]
+        finally:
+            cursor.close()
+    finally:
+        dbapi_connection.autocommit = autocommit
+
+
 def _enable_sqlite_foreign_keys(dbapi_connection, _connection_record) -> None:
-    dbapi_connection.execute("PRAGMA foreign_keys = ON")
+    _set_sqlite_foreign_keys(dbapi_connection, True)
 
 
 @contextmanager
@@ -24,9 +41,9 @@ def fk_checks_disabled_for_refresh(engine: Engine) -> Iterator[None]:
 
     The orchestrator commits each API page in a fresh session, so a session
     scoped PRAGMA/replication-role change would only affect one arbitrary pool
-    connection.  SQLite uses pool checkout/checkin listeners; Postgres uses
-    orm_loader's engine-scoped replica role and disposes the pool afterwards so
-    no connection carrying the disabled role can be reused.
+    connection. SQLite disables checks on checkout and restores them after the
+    refresh; Postgres disposes the pool so disabled replica connections cannot
+    be reused.
     """
 
     backend = resolve_backend(engine)
@@ -42,29 +59,34 @@ def fk_checks_disabled_for_refresh(engine: Engine) -> Iterator[None]:
     if backend.name != "sqlite":
         raise NotImplementedError(f"Unsupported database backend for PBS refresh: {backend.name}")
 
-    def disable_on_pool_event(dbapi_connection, *_args) -> None:
-        dbapi_connection.execute("PRAGMA foreign_keys = OFF")
-
-    def enable_on_checkin(dbapi_connection, *_args) -> None:
-        if dbapi_connection is not None:
-            dbapi_connection.execute("PRAGMA foreign_keys = ON")
+    def disable_on_checkout(dbapi_connection, *_args) -> None:
+        cursor = dbapi_connection.execute("PRAGMA foreign_keys")
+        try:
+            if cursor.fetchone()[0] != 0:
+                _set_sqlite_foreign_keys(dbapi_connection, False)
+        finally:
+            cursor.close()
 
-    event.listen(engine, "connect", disable_on_pool_event)
-    event.listen(engine, "checkout", disable_on_pool_event)
-    event.listen(engine, "checkin", enable_on_checkin)
+    event.listen(engine, "checkout", disable_on_checkout)
     try:
         yield
     finally:
-        event.remove(engine, "connect", disable_on_pool_event)
-        event.remove(engine, "checkout", disable_on_pool_event)
-        try:
-            with engine.connect() as connection:
-                connection.execute(sa.text("PRAGMA foreign_keys = ON"))
-                state = connection.execute(sa.text("PRAGMA foreign_keys")).scalar_one()
-                if state != 1:
-                    raise RuntimeError("Failed to restore SQLite foreign-key enforcement")
-        finally:
-            event.remove(engine, "checkin", enable_on_checkin)
+        event.remove(engine, "checkout", disable_on_checkout)
+        database = engine.url.database
+        in_memory = (
+            database in (None, "", ":memory:", "file::memory:")
+            or engine.url.query.get("mode") == "memory"
+        )
+        if not in_memory:
+            # File-backed pool connections may still have FK checks disabled.
+            engine.dispose()
+        with engine.connect() as connection:
+            state = _set_sqlite_foreign_keys(
+                connection.connection.driver_connection,
+                True,
+            )
+            if state != 1:
+                raise RuntimeError("Failed to restore SQLite foreign-key enforcement")
 
 
 def make_session_factory(engine: Engine):
@@ -80,5 +102,5 @@ def init_db(engine: Engine) -> None:
         if not event.contains(engine, "connect", _enable_sqlite_foreign_keys):
             event.listen(engine, "connect", _enable_sqlite_foreign_keys)
         with engine.connect() as connection:
-            connection.execute(sa.text("PRAGMA foreign_keys = ON"))
+            _set_sqlite_foreign_keys(connection.connection.driver_connection, True)
     Base.metadata.create_all(engine)

src/pbs_client/sync/__init__.py

@@ -1,5 +1,17 @@
 """PBS API to local database synchronization."""
 
-from pbs_client.sync.orchestrator import SyncOrchestrator, SyncResult, mirror_status, upsert_records
+from pbs_client.sync.orchestrator import (
+    SyncOrchestrator,
+    SyncResult,
+    mirror_status,
+    sync_integrity_issues,
+    upsert_records,
+)
 
-__all__ = ["SyncOrchestrator", "SyncResult", "mirror_status", "upsert_records"]
+__all__ = [
+    "SyncOrchestrator",
+    "SyncResult",
+    "mirror_status",
+    "sync_integrity_issues",
+    "upsert_records",
+]

src/pbs_client/sync/orchestrator.py

@@ -3,10 +3,19 @@
 from __future__ import annotations
 
 import logging
+import sys
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from typing import Any
 
+from rich.progress import (
+    BarColumn,
+    MofNCompleteColumn,
+    Progress,
+    TextColumn,
+    TimeElapsedColumn,
+    TimeRemainingColumn,
+)
 from sqlalchemy import Engine, func, select
 from sqlalchemy.orm import Session
 
@@ -55,6 +64,17 @@ def upsert_records(session: Session, model: type[Base], records: Iterable[dict[s
     return written
 
 
+def _underreported_total(state: SyncState) -> int | None:
+    total_records = state.metadata_json.get("total_records")
+    if (
+        isinstance(total_records, int)
+        and not isinstance(total_records, bool)
+        and state.records_written < total_records
+    ):
+        return total_records
+    return None
+
+
 class SyncOrchestrator:
     """Coordinate API pages and local transactions in the required order."""
 
@@ -125,24 +145,53 @@ def _sync_resource(self, name: str, state: SyncState, *, limit: int) -> SyncResu
         self._save_state(state)
         pages = 0
         try:
-            for page in self.client.iter_pages(spec.endpoint, limit=limit, start_page=start_page):
-                pages += 1
-                with self.session_factory() as session:
-                    current = session.get(SyncState, name)
-                    if current is None:
-                        raise PBSSyncError(f"sync state disappeared for {name}")
-                    count = upsert_records(session, model, page.records)
-                    metadata = {
-                        "total_records": page.total_records,
-                        "messages": page.messages,
-                        "links": page.links,
-                        "synced_at": page.metadata.get("synced_at"),
-                        "page_limit": limit,
-                    }
-                    current.checkpoint(page.page, count, metadata)
-                    session.commit()
-                    state = current
-                logger.info("Synced %s page %s (%s records)", name, page.page, count)
+            initial_total = state.metadata_json.get("total_records") if can_resume else None
+            if isinstance(initial_total, bool) or not isinstance(initial_total, int):
+                initial_total = None
+            with Progress(
+                TextColumn("{task.description}"),
+                BarColumn(),
+                MofNCompleteColumn(),
+                TextColumn("records"),
+                TimeElapsedColumn(),
+                TimeRemainingColumn(),
+                disable=not sys.stderr.isatty(),
+                transient=True,
+            ) as progress:
+                task_id = progress.add_task(
+                    name,
+                    total=initial_total,
+                    completed=state.records_written,
+                )
+                for page in self.client.iter_pages(
+                    spec.endpoint,
+                    limit=limit,
+                    start_page=start_page,
+                ):
+                    pages += 1
+                    with self.session_factory() as session:
+                        current = session.get(SyncState, name)
+                        if current is None:
+                            raise PBSSyncError(f"sync state disappeared for {name}")
+                        count = upsert_records(session, model, page.records)
+                        metadata = {
+                            "total_records": page.total_records,
+                            "messages": page.messages,
+                            "links": page.links,
+                            "synced_at": page.metadata.get("synced_at"),
+                            "page_limit": limit,
+                        }
+                        current.checkpoint(page.page, count, metadata)
+                        session.commit()
+                        state = current
+                    progress.update(task_id, total=page.total_records)
+                    progress.advance(task_id, len(page.records))
+                    logger.info("Synced %s page %s (%s records)", name, page.page, count)
+            if (total_records := _underreported_total(state)) is not None:
+                raise PBSSyncError(
+                    f"{name} wrote {state.records_written:,} of the API-reported "
+                    f"{total_records:,} records"
+                )
             state.complete()
             self._save_state(state)
         except Exception as exc:
@@ -226,3 +275,17 @@ def mirror_status(session: Session) -> list[dict[str, Any]]:
             }
         )
     return rows
+
+
+def sync_integrity_issues(session: Session) -> list[SyncState]:
+    """Find checkpoints that wrote fewer API records than the final page reports.
+
+    Compare the API page count with ``records_written``, not table row counts:
+    upserts can legitimately collapse repeated records across schedules.
+    """
+
+    issues = []
+    for state in session.scalars(select(SyncState).order_by(SyncState.resource)):
+        if _underreported_total(state) is not None:
+            issues.append(state)
+    return issues