diff --git a/.env.example b/.env.example index 54c576b3..8cffbcb2 100644 --- a/.env.example +++ b/.env.example @@ -78,9 +78,12 @@ GOOGLE_APPLICATION_CREDENTIALS=/path/to/gcs_credentials.json # chemistry ingestion: Google Drive folder an engineer ingests LIMS .xlsx # workbooks from, on demand (no polling/scheduling; service account / ADC must # have read access to this folder). The ingested-file manifest is stored in -# GCS_BUCKET_NAME at CHEMISTRY_INGEST_MANIFEST_PATH. +# GCS_BUCKET_NAME, scoped per POSTGRES_DB (chemistry-ingest/manifest..json) +# so a local ingest never marks a file done for staging or production. Leave +# CHEMISTRY_INGEST_MANIFEST_PATH unset unless you need to force one specific +# manifest object. CHEMISTRY_DRIVE_FOLDER_ID= -CHEMISTRY_INGEST_MANIFEST_PATH=chemistry-ingest/manifest.json +#CHEMISTRY_INGEST_MANIFEST_PATH=chemistry-ingest/manifest.json # set to development for lexicon and parameter to be populated and enable the enums to work MODE=development diff --git a/cli/cli.py b/cli/cli.py index b4ff204c..385c24af 100644 --- a/cli/cli.py +++ b/cli/cli.py @@ -1163,6 +1163,78 @@ def water_chemistry_sync_drive( raise typer.Exit(result.exit_code) +@water_chemistry.command("manifest-status") +def water_chemistry_manifest_status( + name: str = typer.Option( + None, + "--name", + help="Only show workbooks whose file name contains this text.", + ), + theme: ThemeMode = typer.Option( + ThemeMode.auto, "--theme", help="Color theme: auto, light, dark." + ), +): + """ + show which databases each chemistry workbook has been ingested into. Reads + every per-database manifest in GCS and merges them for display; writes + nothing, so it is safe to run against any environment. + """ + from services.chemistry_drive import ChemistryDriveConfigError, manifest_overview + + colors = _palette(theme) + try: + overview = manifest_overview() + except ChemistryDriveConfigError as exc: + typer.secho(str(exc), fg=colors["issue"], bold=True, err=True) + raise typer.Exit(1) from exc + + databases = overview.databases + if not databases and not overview.unreadable: + typer.secho("No chemistry ingest manifests found.", fg=colors["muted"]) + return + + records = sorted(overview.files.items(), key=lambda kv: kv[1]["name"].lower()) + if name: + needle = name.lower() + records = [r for r in records if needle in r[1]["name"].lower()] + + typer.secho("[CHEMISTRY MANIFEST STATUS]", fg=colors["accent"], bold=True) + typer.secho("=" * 72, fg=colors["accent"]) + typer.secho(f"Databases: {', '.join(databases) or 'none'}", fg=colors["accent"]) + if overview.unreadable: + typer.secho( + f"Unreadable manifests (not shown below): " + f"{', '.join(overview.unreadable)}", + fg=colors["issue"], + ) + typer.echo() + + if not records: + typer.secho("No workbooks match.", fg=colors["muted"]) + return + + for file_id, record in records: + typer.secho(record["name"], fg=colors["field"], bold=True) + for db in databases: + entry = record["databases"].get(db) + if entry is None: + typer.secho(f" {db:<28} | not ingested", fg=colors["muted"]) + continue + status = entry.get("status", "unknown") + color = colors["ok"] if status == "success" else colors["issue"] + detail = f"{entry.get('rows_imported', 0)} row(s)" + if status != "success": + detail = entry.get("error") or "ingestion failed" + when = (entry.get("ingested_at") or "")[:19] + typer.secho( + f" {db:<28} | {status:<8} | {detail} | {when}", + fg=color, + ) + typer.echo() + + typer.secho("=" * 72, fg=colors["accent"]) + + @data_migrations.command("list") def data_migrations_list( theme: ThemeMode = typer.Option( diff --git a/docs/chemistry-ingestion-runbook.md b/docs/chemistry-ingestion-runbook.md index 6831d3f3..31b6c0f7 100644 --- a/docs/chemistry-ingestion-runbook.md +++ b/docs/chemistry-ingestion-runbook.md @@ -152,8 +152,9 @@ Br); resolve the base `SamplePointID → Thing`. Then, per distinct lab sample otherwise create a new `NMA_Chemistry_SampleInfo` whose `nma_sample_point_id` is the base PointID with the **next letter incrementor** appended (`A`, `B`, ... `Z`, `AA`, ...), and insert the analyte rows under it. -A data-quality problem (a row that fails to map, or a `SamplePointID` with no -matching well) aborts the whole file — nothing is written. +A data-quality problem aborts the whole file and nothing is written: a row that +fails to map, a row with no `SampleNumber`, or a `SamplePointID` with no +matching well. --- @@ -162,7 +163,9 @@ matching well) aborts the whole file — nothing is written. - **Duplicate detection is WCLab_ID-only.** A re-ingest is recognized by the lab `WCLab_ID` (SampleNumber). A genuinely new lab sample with a reused SampleNumber would be treated as a duplicate and skipped; a re-run of the same sample under a - new SampleNumber would append a spurious extra lettered sample. + new SampleNumber would append a spurious extra lettered sample. A row with no + SampleNumber at all is rejected rather than loaded, since there would be + nothing to recognize it by on the next run. - **`.xlsx` only.** Legacy `.xls` LIMS exports are not read; the file must be a modern `.xlsx`. - **Fixed analyte map.** Unknown `Param` names fail until engineering adds them diff --git a/services/chemistry_drive.py b/services/chemistry_drive.py index f32ce1d0..6a3093e2 100644 --- a/services/chemistry_drive.py +++ b/services/chemistry_drive.py @@ -20,14 +20,17 @@ polling or scheduling. New/changed ``.xlsx`` files under ``CHEMISTRY_DRIVE_FOLDER_ID`` are downloaded and handed to :func:`services.chemistry_lims.bulk_upload_chemistry`. A manifest of -already-ingested files is kept as a JSON object in the GCS bucket -(``CHEMISTRY_INGEST_MANIFEST_PATH``, default ``chemistry-ingest/manifest.json``) -so re-runs only process files that are new or whose contents changed. +already-ingested files is kept as a JSON object in the GCS bucket. By default +the manifest path is scoped per target database (``chemistry-ingest/manifest. +.json``) so pointing ``.env`` at a different database -- local +copy, staging, production -- never skips a file on the strength of an ingest +into a different database. Configuration (environment variables): * ``CHEMISTRY_DRIVE_FOLDER_ID`` - Drive folder id to scan (shared-drive or My-Drive folder shared with the service account). * ``CHEMISTRY_INGEST_MANIFEST_PATH`` - GCS object key for the manifest. + Overrides the per-database default below. * ``GCS_BUCKET_NAME`` - bucket that holds the manifest (shared with gcs_helper). Authentication mirrors ``services.gcs_helper``: in production the base64 @@ -43,6 +46,7 @@ import json import logging import os +import re from dataclasses import dataclass, field from datetime import datetime, timezone from functools import lru_cache @@ -56,7 +60,11 @@ DRIVE_SCOPES = ["https://www.googleapis.com/auth/drive.readonly"] XLSX_MIME = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" -DEFAULT_MANIFEST_PATH = "chemistry-ingest/manifest.json" +MANIFEST_PREFIX = "chemistry-ingest/" +# Recovers the database name from a per-database manifest object key. Only the +# generated names match, so an explicit CHEMISTRY_INGEST_MANIFEST_PATH override +# is left out of the cross-database view rather than reported under a guess. +MANIFEST_NAME_RE = re.compile(r"^manifest\.(?P[a-z0-9-]+)\.json$") class ChemistryDriveConfigError(Exception): @@ -167,8 +175,23 @@ def download_drive_file(file_id: str, service=None) -> bytes: # --- manifest (GCS JSON object) ------------------------------------------------ +def _manifest_db_suffix() -> str: + """Slug the target database name for use in a manifest path. + + Scoping by ``POSTGRES_DB`` (rather than a separate "which environment am + I" variable) means the manifest can never disagree with the database + ``.env`` is actually pointed at -- there is nothing to keep in sync. + """ + db_name = os.environ.get("POSTGRES_DB", "").strip().lower() + slug = re.sub(r"[^a-z0-9-]+", "-", db_name).strip("-") + return slug or "unknown" + + def _manifest_path() -> str: - return os.environ.get("CHEMISTRY_INGEST_MANIFEST_PATH", DEFAULT_MANIFEST_PATH) + override = os.environ.get("CHEMISTRY_INGEST_MANIFEST_PATH") + if override: + return override + return f"{MANIFEST_PREFIX}manifest.{_manifest_db_suffix()}.json" def _manifest_bucket(): @@ -206,6 +229,61 @@ def save_manifest(manifest: dict[str, dict], bucket=None) -> None: ) +def manifest_databases(bucket=None) -> list[str]: + """Database names that have a manifest object, sorted by name.""" + bucket = bucket or _manifest_bucket() + names = [] + for blob in bucket.list_blobs(prefix=MANIFEST_PREFIX): + match = MANIFEST_NAME_RE.match(blob.name[len(MANIFEST_PREFIX) :]) + if match: + names.append(match.group("db")) + return sorted(names) + + +@dataclass +class ManifestOverview: + """A read-only cross-database view of the per-database manifests.""" + + # Databases whose manifest parsed. Unreadable ones are reported separately + # so a corrupt manifest is never rendered as "this file is not ingested + # there", which is a different and much more alarming claim. + databases: list[str] = field(default_factory=list) + unreadable: list[str] = field(default_factory=list) + files: dict[str, dict] = field(default_factory=dict) + + +def manifest_overview(bucket=None) -> ManifestOverview: + """Merge every per-database manifest into one read-only cross-db view. + + ``files`` maps ``file_id -> {"name": str, "databases": {db: entry}}``. This + only reads, and is derived from the per-database manifests rather than + replacing them: the authoritative skip decision stays scoped to a single + database so concurrent runs cannot clobber each other's bookkeeping. + """ + bucket = bucket or _manifest_bucket() + overview = ManifestOverview() + for db in manifest_databases(bucket): + blob = bucket.blob(f"{MANIFEST_PREFIX}manifest.{db}.json") + if not blob.exists(): + continue + try: + manifest = json.loads(blob.download_as_text()) + except (ValueError, json.JSONDecodeError): + logger.warning("Chemistry ingest manifest for %s is corrupt; skipping.", db) + overview.unreadable.append(db) + continue + overview.databases.append(db) + for file_id, entry in manifest.items(): + record = overview.files.setdefault( + file_id, {"name": entry.get("name", file_id), "databases": {}} + ) + # Later manifests may carry a renamed file; keep a name over a bare id. + if entry.get("name"): + record["name"] = entry["name"] + record["databases"][db] = entry + return overview + + # --- orchestration ------------------------------------------------------------- diff --git a/services/chemistry_lims.py b/services/chemistry_lims.py index 665a447f..68699651 100644 --- a/services/chemistry_lims.py +++ b/services/chemistry_lims.py @@ -277,6 +277,14 @@ def prep_record(record: dict) -> dict: if not pointid: raise ChemistryMappingError("Missing SamplePointID") + # The WCLab_ID is the only thing that makes a re-ingest recognizable, so a + # row without one is not loadable data: it would be appended again under a + # fresh lettered sample point on every run. Same policy as a missing + # SamplePointID, a row error that aborts the file before anything is written. + wclab_id = _get(record, "SampleNumber") + if not wclab_id: + raise ChemistryMappingError("Missing SampleNumber") + units = mapping.units or _get(record, "Results_Units") reported = _get(record, "ReportedND") @@ -300,7 +308,6 @@ def prep_record(record: dict) -> dict: analysis_date = _to_datetime(_get(record, "AnalysisTime")) sample_date = _to_datetime(_get(record, "SampleDate")) or analysis_date - wclab_id = _get(record, "SampleNumber") return { "analyte": mapping.analyte, @@ -311,7 +318,7 @@ def prep_record(record: dict) -> dict: "analysis_method": str(analysis_method) if analysis_method else None, "analysis_date": analysis_date, "sample_date": sample_date, - "wclab_id": str(wclab_id) if wclab_id is not None else None, + "wclab_id": str(wclab_id), "samplepointid": str(pointid), "test": _get(record, "Test"), } @@ -433,6 +440,9 @@ def _sample_exists_for_wclab( session: Session, thing_id: int, wclab_id: str | None ) -> bool: """True if this lab sample (WCLab_ID) is already recorded for the Thing.""" + # Unreachable via prep_record, which rejects a blank SampleNumber. Kept + # because a None would compare as IS NULL and match legacy rows, silently + # skipping a real sample. if wclab_id is None: return False return ( @@ -483,8 +493,9 @@ def bulk_upload_chemistry( ...). A lab sample already recorded for the well (same ``WCLab_ID``) is skipped, so re-running is idempotent. - A data-quality problem (a row that fails to map, or a ``SamplePointID`` with - no matching Thing) aborts the whole file -- nothing is written. + A data-quality problem aborts the whole file and nothing is written: a row + that fails to map, a row with no ``SampleNumber`` (the WCLab_ID that makes a + re-ingest recognizable), or a ``SamplePointID`` with no matching Thing. """ if isinstance(source, str): source = Path(source) diff --git a/tests/test_chemistry_drive.py b/tests/test_chemistry_drive.py index 12a672c5..6659c663 100644 --- a/tests/test_chemistry_drive.py +++ b/tests/test_chemistry_drive.py @@ -98,6 +98,13 @@ def __init__(self): def blob(self, name): return FakeBlob(self.store, name) + def list_blobs(self, prefix=""): + return [ + SimpleNamespace(name=name) + for name in sorted(self.store) + if name.startswith(prefix) + ] + @pytest.fixture() def fake_bucket(): @@ -136,6 +143,77 @@ def test_manifest_roundtrip(fake_bucket): assert load_manifest(fake_bucket) == {"F1": {"status": "success"}} +def test_manifest_path_scoped_to_database(monkeypatch): + """The manifest key follows POSTGRES_DB so one database cannot mask another.""" + monkeypatch.delenv("CHEMISTRY_INGEST_MANIFEST_PATH", raising=False) + + monkeypatch.setenv("POSTGRES_DB", "ocotillo-staging") + assert ( + chemistry_drive._manifest_path() + == "chemistry-ingest/manifest.ocotillo-staging.json" + ) + + # Underscores and case are slugged so the key is a safe object name. + monkeypatch.setenv("POSTGRES_DB", "Ocotillo_Prod_Copy") + assert ( + chemistry_drive._manifest_path() + == "chemistry-ingest/manifest.ocotillo-prod-copy.json" + ) + + +def test_manifest_path_override_wins(monkeypatch): + monkeypatch.setenv("POSTGRES_DB", "ocotillo-staging") + monkeypatch.setenv("CHEMISTRY_INGEST_MANIFEST_PATH", "custom/manifest.json") + assert chemistry_drive._manifest_path() == "custom/manifest.json" + + +def test_manifest_overview_merges_databases(monkeypatch, fake_bucket): + """The cross-database view answers 'has this workbook reached staging?'.""" + monkeypatch.delenv("CHEMISTRY_INGEST_MANIFEST_PATH", raising=False) + + monkeypatch.setenv("POSTGRES_DB", "ocotillo_prod_copy") + save_manifest( + { + "F1": {"name": "wells.xlsx", "status": "success", "rows_imported": 12}, + "F2": {"name": "other.xlsx", "status": "failed", "error": "no Thing"}, + }, + fake_bucket, + ) + + monkeypatch.setenv("POSTGRES_DB", "ocotillo-staging") + save_manifest( + {"F1": {"name": "wells.xlsx", "status": "success", "rows_imported": 12}}, + fake_bucket, + ) + + assert chemistry_drive.manifest_databases(fake_bucket) == [ + "ocotillo-prod-copy", + "ocotillo-staging", + ] + + overview = chemistry_drive.manifest_overview(fake_bucket) + assert overview.files["F1"]["name"] == "wells.xlsx" + assert set(overview.files["F1"]["databases"]) == { + "ocotillo-prod-copy", + "ocotillo-staging", + } + # F2 only ever ingested locally, so staging is absent rather than blank. + assert set(overview.files["F2"]["databases"]) == {"ocotillo-prod-copy"} + + +def test_manifest_overview_separates_corrupt_manifest(monkeypatch, fake_bucket): + """A corrupt manifest must not read as 'not ingested there'.""" + monkeypatch.delenv("CHEMISTRY_INGEST_MANIFEST_PATH", raising=False) + monkeypatch.setenv("POSTGRES_DB", "ocotillo-staging") + save_manifest({"F1": {"name": "wells.xlsx", "status": "success"}}, fake_bucket) + fake_bucket.store["chemistry-ingest/manifest.broken.json"] = "{not json" + + overview = chemistry_drive.manifest_overview(fake_bucket) + assert overview.databases == ["ocotillo-staging"] + assert overview.unreadable == ["broken"] + assert set(overview.files["F1"]["databases"]) == {"ocotillo-staging"} + + def test_missing_folder_raises(monkeypatch): monkeypatch.delenv("CHEMISTRY_DRIVE_FOLDER_ID", raising=False) with pytest.raises(ChemistryDriveConfigError): diff --git a/tests/test_chemistry_lims.py b/tests/test_chemistry_lims.py index 086383b5..632d3687 100644 --- a/tests/test_chemistry_lims.py +++ b/tests/test_chemistry_lims.py @@ -328,6 +328,43 @@ def test_bulk_upload_reports_missing_thing(tmp_path, _cleanup_chemistry): assert any("no matching Thing" in e for e in result.payload["validation_errors"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_prep_record_missing_sample_number_raises(blank): + """No WCLab_ID means no way to recognize a re-ingest, so reject the row.""" + from services.chemistry_lims import ChemistryMappingError + + with pytest.raises(ChemistryMappingError, match="Missing SampleNumber"): + prep_record(_lims_row("calcium", "12.5", SampleNumber=blank)) + + +def test_bulk_upload_reports_missing_sample_number( + tmp_path, water_well_thing, _cleanup_chemistry +): + """A blank SampleNumber aborts the file rather than loading un-redoable rows.""" + path = _write_workbook( + tmp_path / "lims.xlsx", + [ + _lims_row("calcium", "12.5"), + _lims_row("magnesium", "3.3", SampleNumber=None), + ], + ) + + result = bulk_upload_chemistry(path) + + assert result.exit_code == 1 + assert result.payload["summary"]["total_rows_imported"] == 0 + assert any("Missing SampleNumber" in e for e in result.payload["validation_errors"]) + + # Nothing was written, including the row that would have mapped cleanly. + with session_ctx() as session: + infos = session.scalars( + select(NMA_Chemistry_SampleInfo).where( + NMA_Chemistry_SampleInfo.nma_wclab_id == "LAB-1" + ) + ).all() + assert infos == [] + + def test_bulk_upload_reports_unmapped_analyte( tmp_path, water_well_thing, _cleanup_chemistry ):