def enrich_pgx(
spec_dir: Path,
*,
mode: str = "best_effort",
offline: bool = False,
declared_use: str = "unstated",
use_pharmvar: bool = True,
use_cpic: bool = True,
write: bool = True,
cpic_cache: Path | None = None,
pharmvar_cache: Path | None = None,
pharmvar_client: PharmVarClient | PharmVarSnapshotClient | None = None,
cpic_client: CpicClient | CpicSnapshotClient | None = None,
) -> PgxResult:
"""Cross-check the module's PGx tables and record what was consulted, into `sources.csv`.
`declared_use` is a third orthogonal axis, never folded into `mode`: `mode` says how hard to fail
on a finding, this says who is using the data and why. A source that forbids sale is *skipped*
when nothing was declared (conservative — the tool must not assert a purpose for the user) and
*refuses* when `commercial` was declared (a direct contradiction).
`offline` is real as of 0.5.1: each leg resolves a built snapshot first and only reaches the live
service when there is none and the run is online. Offline with no snapshot is a **skip with a
reason** (`skipped_offline`), never a silent pass and never a failure — a source the deployment
cannot reach is not a finding about the module.
"""
spec_dir = Path(spec_dir)
result = PgxResult(mode=mode, declared_use=declared_use)
genes = _module_genes(spec_dir)
if not genes:
note = (
"PGx cross-check skipped: the module names no genes in allele_function.csv or "
"haplotypes.csv, so there is nothing to check against."
)
result.warnings.append(note)
if not any((spec_dir / name).exists() for name, _, _ in _GENE_TABLES):
# **Not attested, and this is the one skip that must not be** — `clinpgx._attest`'s rule,
# which applies here for its own reason. A module carrying neither PGx table has no star
# allele for PharmVar or CPIC to have an opinion about, so the check does not *apply*
# rather than having failed to run; recording it would mine a nonce, create a
# `verification.json` on a module that never asked for one, and publish a
# `manifest.verification` block about a question this module cannot pose.
return result
# A table that is present and names no gene is a different answer: the check applies, and it
# had nothing in scope.
return _attest(
result,
spec_dir,
write=write,
record=skipped("allele_function", "nothing_to_check", detail=note),
)
authored = _authored_functions(spec_dir)
# Alleles carrying an authored `function_status`, which are the only claims this check has to
# put. A module that lists its alleles and states no function for any of them has nothing for an
# authority to disagree with, and saying so is not the same as saying they all agreed.
claims = {(row.gene, row.allele) for row in authored if row.function_status is not None}
compared: set[tuple[str, str]] = set()
# `source -> what became of its leg`, one of the `VALID_VERIFICATION_SKIPS` members or
# `_ANSWERED`. Recorded per leg because two authorities answer this one check and the record
# carries one reason: with nothing answering, *which* absence it was decides what a reader does
# next, and the ordering below is what picks between them.
legs: dict[str, tuple[str, str]] = {}
releases: dict[str, str | None] = {}
# Existing rows are authoritative and are never clobbered, matching `enrich()`. The path comes from
# the shared resolver like every other pass's (RM51): this is the one whose *primary* output is
# this table, so a private literal here would re-create the retired spelling beside the alias.
existing_path = sources_path(spec_dir, error=PgxEnrichmentError)
existing: dict[tuple[str, str], SourceRow] = {}
if existing_path.exists():
rows, errors, _ = load_csv_rows(existing_path, SourceRow, existing_path.name)
if errors:
raise PgxEnrichmentError(f"existing {existing_path.name} is invalid: {errors[0]}")
existing = {(r.source, r.layer): r for r in rows}
emitted: list[SourceRow] = []
def consult(terms: SourceTerms, enabled: bool, resolve, read) -> None:
"""Gate one source on the declared use, pick its route, read it. Records terms either way.
`resolve` returns `(client, owned, route)` or `None` when neither a snapshot nor a live route
is available — which is a *skip with a reason*, not a failure.
"""
if not enabled:
legs[terms.source] = ("not_requested", f"{terms.source}: the caller switched this leg off.")
return
# The module's own recorded declaration counts when the flag states none (S105, RM252): a
# module drafted under `--use non-commercial` carries that row for CPIC, and this check was
# saying "no use was declared" about the file it had just read. Per leg — a declaration for
# CPIC says nothing about PharmVar, which still asks.
use, declared_from = effective_declared_use(spec_dir, terms, declared_use)
reason = check_declared_use(terms, use) # raises LicenseRefusal on `commercial`
if reason is not None:
result.skipped.append(reason)
legs[terms.source] = ("not_permitted", reason)
logger.warning("%s", reason)
return
if declared_from is not None:
result.recorded_use[terms.source] = use
logger.info(
"%s: use %r read from %s, recorded by an earlier run.", terms.source, use, declared_from
)
try:
resolved = resolve()
except (PharmVarError, CpicError) as exc:
# **A resolve-time failure is `no_reference`, not `unreachable`.** Nothing was asked here:
# this leg could not produce a client at all — no snapshot, and no usable live route (the
# PharmVar key is the live instance of it) — which the vocabulary spells "no snapshot /
# sequence / list was provisioned to compare against". `unreachable` says the source was
# asked and never answered, so a reader would retry a request that was never made; the
# remedy here is a key or a built snapshot, and the sentence names which.
note = f"{terms.source} unavailable ({exc}); continuing without it."
result.warnings.append(note)
legs[terms.source] = ("no_reference", note)
return
if resolved is None:
note = (
f"{terms.source}: skipped — --offline and no built snapshot. Build one with "
f"`just-dna-enricher {terms.source} build`, or point at it with "
f"$JUST_DNA_{terms.source.upper()}_CACHE."
)
result.skipped_offline.append(note)
legs[terms.source] = ("offline", note)
logger.warning("%s", note)
return
client, owned, route = resolved
dataset = getattr(client, "dataset", None)
# **Per LEG, never per record (RM73).** `pgx_draft` copies `function_status` straight out of
# CPIC and this check then compares that very column against CPIC, so on a drafted module the
# CPIC leg cannot fail — RM4's tautology, unmarked here for two releases because this was the
# provider recording no release at all. PharmVar's leg is an independent authority and still
# runs; the record aggregates both. A whole-record skip would have thrown away a real
# comparison to suppress a hollow one, which is why the module-level shape did not fit.
tautological = _tautology_note(spec_dir, terms.source, dataset, existing)
if tautological is not None:
if owned:
client.close()
result.routes[terms.source] = route
legs[terms.source] = ("tautology", tautological)
releases[terms.source] = dataset
emitted.append(terms.row("annotation", declared_use=use, dataset=dataset))
# On `result.warnings`, not `logger.info` alone, and the mixed case is why: when PharmVar
# answers and CPIC skips, the record's `detail` names only the answered route, so a reader
# sees a clean comparison and no sign that half of it was hollow. Every sibling branch
# here surfaces its reason the same way — a skip that is not reported is the silent pass
# this whole item exists to end.
result.warnings.append(tautological)
logger.info("%s", tautological)
return
try:
reported, notes = read(client)
except (PharmVarError, CpicError) as exc:
# One source failing must not sink the pass — the other may still answer.
note = f"{terms.source} unavailable ({exc}); continuing without it."
result.warnings.append(note)
legs[terms.source] = ("unreachable", note)
return
finally:
if owned:
client.close()
result.routes[terms.source] = route
result.warnings.extend(notes)
conflicts, checked = _compare(authored, reported, terms.source)
result.conflicts.extend(conflicts)
compared.update(checked)
legs[terms.source] = (_ANSWERED, route)
releases[terms.source] = dataset
emitted.append(terms.row("annotation", declared_use=use, dataset=dataset))
def _injected(client) -> tuple[object, bool, str] | None:
"""An injected client's route — and `None` when `offline` forbids using it.
**`offline` wins over an injection, and the type is what decides.** An injected client is the
inject-only escape hatch, but a *live* one under `--offline` would egress from a run documented
as making none, which is the whole failure RM38 exists to close. A snapshot client is exempt
because reading a local parquet is not egress. Not decided on `configured`: a live client with a
perfectly good key is exactly the one that must not be used here.
"""
if client is None:
return None
is_snapshot = isinstance(client, CpicSnapshotClient | PharmVarSnapshotClient)
if offline and not is_snapshot:
return None
return client, False, "snapshot" if is_snapshot else "injected"
def _resolve_pharmvar():
if pharmvar_client is not None:
resolved = _injected(pharmvar_client)
# "No key" is still the caller's answer to hear: a keyless client would 401 every gene, and
# the CPIC leg must survive that. `PharmVarSnapshotClient.configured` is True, so a
# snapshot passes unchanged.
if resolved is not None and not pharmvar_client.configured:
raise PharmVarError(
"no PharmVar API key: set PHARMVAR_API_KEY (the key is personal to your account "
"and is never stored in a module)"
)
return resolved
reference = resolve_pharmvar_reference(pharmvar_cache)
if reference is not None:
return PharmVarSnapshotClient(reference), True, "snapshot"
if offline:
return None
client = PharmVarClient()
if not client.configured:
client.close()
raise PharmVarError(
"no PharmVar API key: set PHARMVAR_API_KEY (the key is personal to your account "
"and is never stored in a module), or build a snapshot with "
"`just-dna-enricher pharmvar build`"
)
return client, True, "live"
def _resolve_cpic():
if cpic_client is not None:
return _injected(cpic_client)
reference = resolve_cpic_reference(cpic_cache)
if reference is not None:
return CpicSnapshotClient(reference), True, "snapshot"
if offline:
return None
return CpicClient(), True, "live"
def _read_pharmvar(client) -> tuple[dict[tuple[str, str], str | None], list[str]]:
reported: dict[tuple[str, str], str | None] = {}
for gene in genes:
for allele in client.alleles_for_gene(gene):
key = (allele.gene, _normalize_allele(allele.gene, allele.allele))
reported[key] = (allele.function or "").replace(" ", "_").lower() or None
return reported, []
def _read_cpic(client) -> tuple[dict[tuple[str, str], str | None], list[str]]:
reported: dict[tuple[str, str], str | None] = {}
for gene in genes:
for allele in client.alleles_for_gene(gene):
reported[(allele.gene, allele.allele)] = allele.function_status
return reported, []
# PharmVar first on data-authority grounds (the naming authority for CYP star alleles), not
# licensing — neither source is sellable.
consult(PHARMVAR_TERMS, use_pharmvar, _resolve_pharmvar, _read_pharmvar)
consult(CPIC_TERMS, use_cpic, _resolve_cpic, _read_cpic)
for conflict in result.conflicts:
logger.warning("PGx allele-function difference — %s", conflict)
result.compared = len(compared)
# Merge, never clobber: an existing row for the same (source, layer) wins.
merged: dict[tuple[str, str], SourceRow] = dict(existing)
for row in emitted:
merged.setdefault((row.source, row.layer), row)
result.rows = [merged[key] for key in sorted(merged)]
if write and result.rows:
write_sources_csv(result.rows, existing_path)
return _attest(
result,
spec_dir,
write=write,
record=_function_check_record(result, claims=claims, legs=legs, releases=releases),
)