overall cleanup

README.md

@@ -35,7 +35,7 @@ subscription_key = "your-subscription-key"
 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. Tests use local fixtures and never call the API.
+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

src/pbs_client/cli/main.py

@@ -2,31 +2,101 @@
 
 from __future__ import annotations
 
+from collections import Counter
+from typing import Any
+
 import typer
 from sqlalchemy.engine import Engine
 
 from pbs_client.config import PBSSettings, get_pbs_context
 from pbs_client.db import init_db, make_session_factory
-from pbs_client.errors import PBSSyncError, PBSTransportError
-from pbs_client.http import PBSClient
+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
 
 app = typer.Typer(help="Maintain a local offline mirror of the PBS Public Data API v3.")
 
 
+def _status_lines(rows: list[dict[str, Any]]) -> list[str]:
+    """Render a compact operational summary followed by grouped resource details."""
+
+    counts = Counter(row["status"] for row in rows)
+    complete = counts.get("complete", 0)
+    total_rows = sum(int(row["rows"]) for row in rows)
+    state_summary = " · ".join(
+        f"{counts.get(status, 0)} {label.lower()}"
+        for status, label in (
+            ("complete", "complete"),
+            ("in_progress", "in progress"),
+            ("failed", "failed"),
+            ("pending", "pending"),
+        )
+        if counts.get(status, 0)
+    )
+    lines = [
+        f"PBS mirror: {complete}/{len(rows)} resources complete · {total_rows:,} rows",
+        f"State: {state_summary}",
+    ]
+    attention = [
+        row["resource"]
+        for row in rows
+        if row["status"] in {"failed", "in_progress"}
+    ]
+    if attention:
+        lines.append(f"Attention: {', '.join(attention)}")
+
+    headings = {
+        "failed": "Failed",
+        "in_progress": "In progress",
+        "pending": "Pending",
+        "complete": "Complete",
+    }
+    for status in ("failed", "in_progress", "pending", "complete"):
+        group = [row for row in rows if row["status"] == status]
+        if not group:
+            continue
+        lines.extend(["", f"{headings[status]} ({len(group)})"])
+        for row in group:
+            if status == "complete":
+                detail = (
+                    f"{int(row['rows']):,} rows · page {row['page']} · "
+                    f"completed {_format_status_time(row['completed_at'])}"
+                )
+            elif status == "in_progress":
+                detail = (
+                    f"{int(row['rows']):,} rows · page {row['page']} · "
+                    f"started {_format_status_time(row['started_at'])}"
+                )
+            elif status == "failed":
+                error = str(row["last_error"] or "unknown error").splitlines()[0]
+                detail = f"{int(row['rows']):,} rows · page {row['page']} · {error}"
+            else:
+                detail = "not started"
+            lines.append(f"  {row['resource']:<30} {detail}")
+    return lines
+
+
+def _format_status_time(value: Any) -> str:
+    if value is None:
+        return "-"
+    return value.strftime("%Y-%m-%d %H:%M")
+
+
 def _runtime() -> tuple[PBSSettings, Engine, str]:
     """Resolve the shared oa-configurator config, engine, and database name."""
 
     config, database = get_pbs_context()
     return PBSSettings.from_config(config), database.create_engine(future=True), config.pbs_db
 
 
-def _find_transport_error(error: BaseException) -> PBSTransportError | None:
-    """Find a transport failure retained in a wrapped sync exception."""
+def _find_cause[Error: BaseException](
+    error: BaseException, expected: type[Error]
+) -> Error | None:
+    """Find a cause of the requested type retained in a wrapped exception."""
 
     current: BaseException | None = error
     while current is not None:
-        if isinstance(current, PBSTransportError):
+        if isinstance(current, expected):
             return current
         current = current.__cause__
     return None
@@ -36,7 +106,9 @@ def _report_sync_failure(error: PBSSyncError) -> None:
     """Print an actionable, traceback-free message for a failed CLI sync."""
 
     resource = error.resource or "the current resource"
-    transport = _find_transport_error(error)
+    transport = _find_cause(error, PBSTransportError)
+    http_error = _find_cause(error, PBSHTTPError)
+    invalid_response = _find_cause(error, PBSInvalidResponseError)
     typer.echo(f"PBS sync paused on {resource}.", err=True)
     if transport is not None and transport.timed_out:
         typer.echo(
@@ -50,6 +122,27 @@ def _report_sync_failure(error: PBSSyncError) -> None:
             err=True,
         )
         limit = " --limit 1000"
+    elif http_error is not None and http_error.status_code == 429:
+        typer.echo(str(error), err=True)
+        typer.echo(
+            "The PBS API rate limit was reached. Wait before retrying; committed "
+            "pages and completed resources are already saved.",
+            err=True,
+        )
+        limit = ""
+    elif invalid_response is not None:
+        typer.echo(str(error), err=True)
+        typer.echo(
+            "The PBS API returned an empty or non-JSON response; this can "
+            "happen when a page is too large or the service returns an error page.",
+            err=True,
+        )
+        typer.echo(
+            "Completed resources and committed pages are already saved. "
+            "Retry with a smaller page size.",
+            err=True,
+        )
+        limit = " --limit 1000"
     else:
         typer.echo(str(error), err=True)
         typer.echo("Completed resources are already saved and can be resumed.", err=True)
@@ -76,7 +169,7 @@ def sync_command(
     refresh: bool = typer.Option(
         True, "--refresh/--resume-only", help="Refresh completed resources."
     ),
-    limit: int = typer.Option(100_000, min=1, help="API page size."),
+    limit: int = typer.Option(DEFAULT_PAGE_SIZE, min=1, help="API page size."),
 ) -> None:
     """Synchronize all PBS resources, or one resource, into the local mirror."""
 
@@ -106,12 +199,8 @@ def status() -> None:
     _, engine, _ = _runtime()
     init_db(engine)
     with make_session_factory(engine)() as session:
-        for row in mirror_status(session):
-            completed = row["completed_at"].isoformat() if row["completed_at"] else "-"
-            typer.echo(
-                f"{row['resource']}: {row['status']}; rows={row['rows']}; "
-                f"page={row['page']}; completed={completed}"
-            )
+        for line in _status_lines(mirror_status(session)):
+            typer.echo(line)
 
 
 def main() -> None:

src/pbs_client/config.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from math import isfinite
 from typing import Annotated, ClassVar
 
 from oa_configurator import (
@@ -19,6 +20,7 @@
 DEFAULT_BASE_URL = "https://data-api.health.gov.au/pbs/api/v3"
 DEFAULT_PUBLIC_KEY = "2384af7c667342ceb5a736fe29f1dc6b"
 DEFAULT_RATE_LIMIT_SECONDS = 20.0
+MIN_RATE_LIMIT_SECONDS = 3.0
 
 
 class PBSClientConfig(PackageConfigBase):
@@ -45,7 +47,7 @@ class PBSClientConfig(PackageConfigBase):
     )
     rate_limit_seconds: float = Field(
         default=DEFAULT_RATE_LIMIT_SECONDS,
-        gt=0,
+        ge=MIN_RATE_LIMIT_SECONDS,
         description="Minimum delay between PBS API requests in seconds.",
     )
 
@@ -80,6 +82,13 @@ class PBSSettings:
     base_url: str = DEFAULT_BASE_URL
     rate_limit_seconds: float = DEFAULT_RATE_LIMIT_SECONDS
 
+    def __post_init__(self) -> None:
+        if not isfinite(self.rate_limit_seconds) or self.rate_limit_seconds < MIN_RATE_LIMIT_SECONDS:
+            raise ValueError(
+                "rate_limit_seconds must be at least "
+                f"{MIN_RATE_LIMIT_SECONDS:g} seconds to respect the PBS API quota"
+            )
+
     @classmethod
     def from_config(cls, config: PBSClientConfig) -> PBSSettings:
         """Build HTTP settings from the resolved package configuration."""

src/pbs_client/db/schema.py

@@ -189,3 +189,9 @@ def primary_key(self) -> tuple[str, ...]:
     "SummaryOfChanges",
     "ApiChangelog",
 )
+
+if len(SYNC_ORDER) != len(set(SYNC_ORDER)):
+    raise RuntimeError("SYNC_ORDER contains duplicate resources")
+unknown = sorted(set(SYNC_ORDER) - set(RESOURCE_BY_NAME))
+if unknown:
+    raise RuntimeError(f"SYNC_ORDER contains unregistered resources: {unknown}")

src/pbs_client/db/state.py

@@ -26,10 +26,12 @@ class SyncState(Base):
     last_error: Mapped[str | None] = mapped_column(Text)
     metadata_json: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict)
 
-    def begin(self) -> None:
+    def begin(self, *, page_limit: int) -> None:
         self.status = "in_progress"
         self.started_at = datetime.now(UTC)
+        self.completed_at = None
         self.last_error = None
+        self.metadata_json = {**self.metadata_json, "page_limit": page_limit}
 
     def checkpoint(self, page: int, count: int, metadata: dict[str, Any]) -> None:
         self.page = page

src/pbs_client/errors.py

@@ -15,6 +15,31 @@ class PBSAPIError(PBSClientError):
     """The PBS API returned an error or an unusable response."""
 
 
+class PBSHTTPError(PBSAPIError):
+    """The PBS API returned an HTTP error response."""
+
+    def __init__(
+        self,
+        url: str,
+        status_code: int,
+        attempts: int,
+        *,
+        retryable: bool,
+        retry_after_seconds: float | None = None,
+    ) -> None:
+        self.url = url
+        self.status_code = status_code
+        self.attempts = attempts
+        self.retryable = retryable
+        self.retry_after_seconds = retry_after_seconds
+        suffix = f" after {attempts} attempts" if retryable else ""
+        super().__init__(f"PBS API returned HTTP {status_code}{suffix}: {url}")
+
+
+class PBSInvalidResponseError(PBSAPIError):
+    """The PBS API response could not be decoded as the expected document."""
+
+
 class PBSTransportError(PBSAPIError):
     """The PBS API could not be reached after all transport retries."""
 

src/pbs_client/http/__init__.py

@@ -1,5 +1,11 @@
 """PBS API v3 HTTP client."""
 
-from pbs_client.http.client import GlobalRateLimiter, Page, PBSClient, TransportResponse
+from pbs_client.http.client import (
+    DEFAULT_PAGE_SIZE,
+    GlobalRateLimiter,
+    Page,
+    PBSClient,
+    TransportResponse,
+)
 
-__all__ = ["GlobalRateLimiter", "PBSClient", "Page", "TransportResponse"]
+__all__ = ["DEFAULT_PAGE_SIZE", "GlobalRateLimiter", "PBSClient", "Page", "TransportResponse"]