File size: 4,866 Bytes
74d4e89 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """Validate-before-availability for a quarantined upload (ADR-0011 items 3, 4, 5).
A staged file stays quarantined until it clears the SAME grounding gate a
registered dataset clears:
1. manifest schema validation (src.datasets.manifest_schema.validate_manifest)
2. semantic against-data checks (src.workflows.manifest_data_validation)
Item 4 ("never execute uploaded content") is honoured structurally: the file is
opened ONLY by the vetted scanpy loader — the same loader registered data uses —
never ``exec``/``eval``/``pickle``. Passing both checks flips the record to
``validated``; failing keeps it quarantined with the errors surfaced.
"""
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
from src.uploads.records import (
STATUS_QUARANTINED,
STATUS_VALIDATED,
UploadRecord,
persist_record,
)
def _report(status: str, errors: list[str], **extra: Any) -> dict:
return {"status": status, "errors": errors, **extra}
def validate_upload(record: UploadRecord) -> tuple[UploadRecord, dict]:
"""Run the manifest-schema + against-data gate on a quarantined upload.
Returns ``(record, report)``. The record's ``status`` becomes ``validated``
only when both gates pass; otherwise it stays ``quarantined`` and the reasons
are in ``record.errors`` / ``report['errors']``.
"""
if record.status != STATUS_QUARANTINED:
return record, _report(
"error",
[f"Upload is '{record.status}', not 'quarantined' — nothing to validate."],
)
if not record.staged_path or not Path(record.staged_path).is_file():
record.errors = [f"Staged file missing: {record.staged_path}"]
persist_record(record)
return record, _report("error", record.errors)
# ── Gate 1: manifest schema validity ─────────────────────────────────────
from src.datasets.manifest_schema import DatasetManifest, validate_manifest
schema_result = validate_manifest(record.manifest)
if not schema_result.valid:
record.errors = [f"manifest schema: {e}" for e in schema_result.errors]
persist_record(record)
return record, _report(
"error", record.errors, schema_warnings=schema_result.warnings
)
manifest_obj = DatasetManifest.from_dict(record.manifest)
# ── Gate 2: against-data validation via the VETTED loader only (item 4) ──
suffix = Path(record.staged_path).suffix.lower()
if suffix != ".h5ad":
# The gate accepts tabular for staging, but auto against-data validation
# is h5ad-only this phase (every served dataset is an h5ad). Stay
# quarantined and say so honestly rather than pretend it validated.
record.errors = [
f"Auto-validation for '{suffix}' uploads is not implemented yet — "
"convert to .h5ad or validate manually. File stays quarantined."
]
persist_record(record)
return record, _report("error", record.errors)
try:
import numpy as np
import scanpy as sc
from src.workflows.manifest_data_validation import validate_manifest_against_data
adata = sc.read_h5ad(record.staged_path) # vetted loader — never exec
X = adata.X
if hasattr(X, "toarray"):
X = X.toarray()
X_flat = X.flatten()
if len(X_flat) > 50_000:
rng = np.random.default_rng(0)
X_flat = rng.choice(X_flat, size=50_000, replace=False)
data_report = validate_manifest_against_data(
X_flat, list(adata.var.index), adata.obs.copy(), manifest_obj
)
except Exception as exc: # noqa: BLE001 — a load/validate failure is a rejection
record.errors = [f"against-data validation failed to run: {exc}"]
persist_record(record)
return record, _report("error", record.errors)
if not data_report.get("overall_valid"):
record.errors = list(data_report.get("errors", [])) or [
"against-data validation reported the manifest inconsistent with the file."
]
persist_record(record)
return record, _report(
"error", record.errors, against_data=data_report,
schema_warnings=schema_result.warnings,
)
# ── Passed both gates ────────────────────────────────────────────────────
record.status = STATUS_VALIDATED
record.validated_at = datetime.now().isoformat(timespec="seconds")
record.errors = []
persist_record(record)
return record, _report(
"pass", [], against_data=data_report, schema_warnings=schema_result.warnings
)
|