From d9974cc969aaea8388a357ffab526ad54ae1feab Mon Sep 17 00:00:00 2001 From: Fang Chen Date: Sat, 29 Aug 2026 07:03:43 -0700 Subject: [PATCH 1/5] PR-A: assembly version schema & path contracts Behaviour-neutral fixes to the column names Phases 0-3 read and write and to where their files land. Addresses F1, F2, F3, F4, F6 and F9 from the Phase 4 plan; no change to which assemblies are detected or what the pipeline decides. F1/F2 - Phase 1 read the non-existent `accession` column and wrote snake_case version metadata, so the Phase 0 -> Phase 1 handoff raised KeyError on the first real row. Adds get_accession / get_assembly_id / get_version_status to assembly_versions_utils and routes every Phase 1 TSV read through them; Phase 1 now emits the canonical `versionStatus` / `assemblyID`. The JSONL read is left alone, where `accession` is the correct NCBI key. F3 - append_superseded_to_tsv derived its header from one arbitrary row, so merging Phase 0 and Phase 1 row sets silently dropped columns. merge_fieldnames now takes the union across all rows, ordered by the YAML header or the order already on disk. F4 - Phases 1-3 hardcoded `assembly_current.tsv` while the production config emits `ncbi_datasets_eukaryota.tsv.gz`, and Phase 1 opened the gzipped `.previous` snapshot as plain text. resolve_current_tsv_paths derives the name from config.meta["file_name"], and open_tsv detects gzip by magic bytes (the snapshot is compressed but does not end `.gz`). F6 - Phase 0 wrote relative to the YAML's directory. resolve_output_path pins the output to --work_dir, so the copy-the-YAML workaround is gone from the READMEs. F9 - Phase 0 wrote with mode "w". Harmless for a one-time backfill, but the F5 decision makes the gap-fill path re-invoke it daily, where it would truncate the ~8,500-row file. write_or_append_parsed writes the whole file only when it does not exist (so the header is produced) and otherwise appends the rows not already present, deduplicating on genbankAccession via a cheap accession-only pass. Also adds `superseded_by`, `superseded_by_version` and `superseded_date` to assembly_historical.types.yaml so they survive the GenomeHubs write path. Tests: fixtures rebuilt on the real headers - building them inline with the wrong keys is why F1-F3 went undetected - plus a schema-conformance test against the YAML and a non-destructive-write test. 76 -> 109 passing. --- configs/assembly_historical.types.yaml | 11 + flows/lib/assembly_versions_utils.py | 202 ++++++++- flows/lib/compute_taxon_milestones.py | 45 +- flows/lib/generate_assembly_summary.py | 50 ++- flows/parsers/parse_assembly_versions.py | 183 ++++++-- .../parse_backfill_historical_versions.py | 115 ++++- tests/README_phase_0_backfill.md | 18 +- tests/README_phase_1_daily_updates.md | 48 ++- tests/test_assembly_versions.py | 408 ++++++++++++++++-- 9 files changed, 948 insertions(+), 132 deletions(-) diff --git a/configs/assembly_historical.types.yaml b/configs/assembly_historical.types.yaml index b98f0fd..f46b222 100644 --- a/configs/assembly_historical.types.yaml +++ b/configs/assembly_historical.types.yaml @@ -136,6 +136,17 @@ attributes: submitter: header: submitter path: assemblyInfo.submitter + # Supersession metadata stamped by the Phase 1 daily parser. Declared here + # so the columns survive the GenomeHubs write path instead of being dropped. + superseded_by: + header: superseded_by + path: processedAssemblyInfo.supersededBy + superseded_by_version: + header: superseded_by_version + path: processedAssemblyInfo.supersededByVersion + superseded_date: + header: superseded_date + path: processedAssemblyInfo.supersededDate ungapped_span: header: totalUngappedLength path: assemblyStats.totalUngappedLength diff --git a/flows/lib/assembly_versions_utils.py b/flows/lib/assembly_versions_utils.py index 070e7c9..73a6a9f 100644 --- a/flows/lib/assembly_versions_utils.py +++ b/flows/lib/assembly_versions_utils.py @@ -1,18 +1,214 @@ -"""Shared utilities for assembly version discovery and fetching. +"""Shared utilities for assembly version discovery, schema and paths. -Used by both the backfill parser and the incremental updater to discover -assembly versions via NCBI FTP and fetch per-version metadata. +Used by the backfill parser, the incremental parser, the summary generator and +the milestone flow to discover assembly versions via NCBI FTP, fetch +per-version metadata, read rows written by any phase, and resolve the +current-assembly TSV paths from a loaded config rather than a hardcoded name. """ +import glob as globmod +import gzip import json import os import re import time +from typing import Callable, Optional from flows.lib import utils ACCESSION_PATTERN = re.compile(r"^GC[AF]_\d{9}\.\d+$") +# Leading bytes of a gzip member, used to detect compression by content. +GZIP_MAGIC = b"\x1f\x8b" + +# Canonical column names, as declared in assembly_historical.types.yaml. Every +# phase writes these; the ``*_ALIASES`` tuples additionally tolerate the +# snake_case spellings written by earlier revisions of the Phase 1 parser. +COL_ACCESSION = "genbankAccession" +COL_ASSEMBLY_ID = "assemblyID" +COL_VERSION_STATUS = "versionStatus" +COL_SUPERSEDED_BY = "superseded_by" +COL_SUPERSEDED_BY_VERSION = "superseded_by_version" +COL_SUPERSEDED_DATE = "superseded_date" + +ACCESSION_ALIASES = (COL_ACCESSION, "accession") +ASSEMBLY_ID_ALIASES = (COL_ASSEMBLY_ID, "assembly_id") +VERSION_STATUS_ALIASES = (COL_VERSION_STATUS, "version_status") + +# Filename of the historical TSV, and the outputs derived from it. Used to +# exclude the pipeline's own products when discovering the current TSV. +HISTORICAL_TSV_NAME = "assembly_historical.tsv" +CURRENT_TSV_DEFAULT = "assembly_current.tsv" +DERIVED_TSV_NAMES = frozenset({ + HISTORICAL_TSV_NAME, + "assembly_version_summary.tsv", + "taxon_milestone_summary.tsv", +}) + + +def _first_present(row: dict, keys: tuple) -> str: + """Return the first non-empty value in ``row`` for any of ``keys``.""" + for key in keys: + value = row.get(key) + if value: + return value + return "" + + +def get_accession(row: dict) -> str: + """Return a row's versioned accession, whichever column holds it. + + Real assembly TSVs use ``genbankAccession``; ``accession`` is accepted for + rows that originated from an NCBI JSONL record or an older parse. + + Args: + row (dict): A TSV row. + + Returns: + str: The versioned accession, or "" when absent. + """ + return _first_present(row, ACCESSION_ALIASES) + + +def get_assembly_id(row: dict) -> str: + """Return a row's assembly ID, tolerating either naming convention. + + Args: + row (dict): A TSV row. + + Returns: + str: The assembly ID, or "" when absent. + """ + return _first_present(row, ASSEMBLY_ID_ALIASES) + + +def get_version_status(row: dict) -> str: + """Return a row's version status, tolerating either naming convention. + + Args: + row (dict): A TSV row. + + Returns: + str: The version status, or "" when absent. + """ + return _first_present(row, VERSION_STATUS_ALIASES) + + +def is_gzipped(path: str) -> bool: + """Report whether a TSV is gzip-compressed. + + The name alone is not enough: ``snapshot_previous_output`` copies a + gzipped current TSV verbatim to ``.gz.previous``, so the compressed + file does not end ``.gz``. Sniff the magic bytes, and fall back to the + extension when the file cannot be read. + + Args: + path (str): Path to the TSV. + + Returns: + bool: True when the file is gzip-compressed. + """ + name = str(path) + try: + with open(name, "rb") as f: + return f.read(2) == GZIP_MAGIC + except OSError: + return name.endswith(".gz") + + +def open_tsv(path: str, mode: str = "rt"): + """Open a TSV transparently, decompressing gzipped files. + + Args: + path (str): Path to the TSV. + mode (str): Text mode to open with. + + Returns: + IO: An open text-mode file object. + """ + if is_gzipped(path): + return gzip.open(path, mode, encoding="utf-8") + return open(path, mode, encoding="utf-8") + + +def discover_current_tsv_name(work_dir: str) -> str: + """Find the current-assembly TSV filename in ``work_dir``. + + Fallback for callers that have no config to resolve from. Considers every + ``*.tsv`` / ``*.tsv.gz`` in ``work_dir`` except the pipeline's own derived + outputs. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + str: Basename of the current TSV. Falls back to + ``assembly_current.tsv`` when nothing is on disk yet, so callers + still produce a meaningful not-found message. + + Raises: + ValueError: If several candidates are present and none of them is the + conventional ``assembly_current.tsv``. + """ + candidates = sorted( + globmod.glob(os.path.join(work_dir, "*.tsv")) + + globmod.glob(os.path.join(work_dir, "*.tsv.gz")) + ) + names = [ + os.path.basename(c) + for c in candidates + if os.path.basename(c) not in DERIVED_TSV_NAMES + ] + if not names: + return CURRENT_TSV_DEFAULT + if len(names) == 1: + return names[0] + if CURRENT_TSV_DEFAULT in names: + return CURRENT_TSV_DEFAULT + raise ValueError( + f"Cannot identify the current assembly TSV in {work_dir}: {names}. " + "Pass a config so the filename can be resolved from meta['file_name']." + ) + + +def resolve_current_tsv_paths( + work_dir: str, config: Optional[object] = None +) -> tuple[str, str, Callable]: + """Resolve the current-assembly TSV paths and the matching open function. + + The filename comes from ``config.meta["file_name"]`` when a config is + supplied, so nothing hardcodes ``assembly_current.tsv``. The previous-run + snapshot is ``.previous``, written verbatim by + ``snapshot_previous_output`` — so it carries the same compression as the + current file, which is why the open function is returned alongside. + + Args: + work_dir (str): Directory holding the pipeline files. + config (Config, optional): Loaded YAML config whose + ``meta["file_name"]`` names the current TSV. + + Returns: + tuple: ``(current_tsv, previous_tsv, open_fn)``. + """ + if config is not None: + name = os.path.basename(config.meta["file_name"]) + else: + name = discover_current_tsv_name(work_dir) + current_tsv = os.path.join(work_dir, name) + return current_tsv, f"{current_tsv}.previous", open_tsv + + +def resolve_historical_tsv_path(work_dir: str) -> str: + """Return the historical TSV path inside ``work_dir``. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + str: Path to assembly_historical.tsv. + """ + return os.path.join(work_dir, HISTORICAL_TSV_NAME) + def parse_version(accession: str) -> int: """Extract the version number from a dotted accession string. diff --git a/flows/lib/compute_taxon_milestones.py b/flows/lib/compute_taxon_milestones.py index f34a473..44b5752 100644 --- a/flows/lib/compute_taxon_milestones.py +++ b/flows/lib/compute_taxon_milestones.py @@ -30,13 +30,19 @@ import csv import os +from flows.lib.assembly_versions_utils import ( + get_accession, + get_version_status, + open_tsv, + resolve_current_tsv_paths, + resolve_historical_tsv_path, +) from flows.lib.conditional_import import emit_event, flow from flows.lib.load_taxonomy import CANONICAL_RANKS, build_taxonomy -from flows.lib.shared_args import TAXDUMP_PATH, WORK_DIR +from flows.lib.shared_args import TAXDUMP_PATH, WORK_DIR, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args +from flows.lib.utils import load_config -CURRENT_TSV = "assembly_current.tsv" -HISTORICAL_TSV = "assembly_historical.tsv" OUTPUT_TSV = "taxon_milestone_summary.tsv" # EBP umbrella BioProject — affiliation is membership in this project. @@ -77,7 +83,7 @@ def _accession(row: dict) -> str: """Return the assembly accession, normalised across phases.""" - return row.get("accession") or row.get("genbankAccession") or "" + return get_accession(row) def _release_date(row: dict) -> str: @@ -88,7 +94,7 @@ def _release_date(row: dict) -> str: def _version_status(row: dict) -> str: """Return versionStatus across camelCase / snake_case, default 'current'.""" - return row.get("versionStatus") or row.get("version_status") or "current" + return get_version_status(row) or "current" def _is_affiliated(row: dict) -> bool: @@ -132,7 +138,7 @@ def load_assemblies(current_tsv: str, historical_tsv: str) -> list[dict]: print(f" Warning: {label} TSV not found: {path}") continue count = 0 - with open(path, encoding="utf-8") as f: + with open_tsv(path) as f: for row in csv.DictReader(f, delimiter="\t"): rows.append(row) count += 1 @@ -325,17 +331,26 @@ def build_output_rows(taxa: dict, species_extra: dict) -> list[dict]: @flow(log_prints=True) -def compute_taxon_milestones(work_dir: str = ".", taxdump_path: str | None = None) -> None: +def compute_taxon_milestones( + work_dir: str = ".", + taxdump_path: str | None = None, + yaml_path: str | None = None, +) -> None: """Compute per-taxon assembly milestones and write the summary TSV. + The current-TSV filename comes from the config when yaml_path is given, + and is otherwise discovered in work_dir, so nothing hardcodes it. + Args: - work_dir: Directory containing assembly_current.tsv and - assembly_historical.tsv; output is written there too. + work_dir: Directory containing the current and historical TSVs; + output is written there too. taxdump_path: Dev/test NCBI taxdump directory. In production the lineage arrives via the upstream contract and this is omitted. + yaml_path: Optional YAML config naming the current TSV. """ - current_tsv = os.path.join(work_dir, CURRENT_TSV) - historical_tsv = os.path.join(work_dir, HISTORICAL_TSV) + config = load_config(config_file=yaml_path) if yaml_path else None + current_tsv, _, _ = resolve_current_tsv_paths(work_dir, config=config) + historical_tsv = resolve_historical_tsv_path(work_dir) output_tsv = os.path.join(work_dir, OUTPUT_TSV) separator = "=" * 80 @@ -406,7 +421,11 @@ def compute_taxon_milestones(work_dir: str = ".", taxdump_path: str | None = Non if __name__ == "__main__": args = _parse_args( - [WORK_DIR, TAXDUMP_PATH], + [WORK_DIR, TAXDUMP_PATH, YAML_PATH], description="Compute per-taxon assembly milestones from version-tracked TSVs", ) - compute_taxon_milestones(work_dir=args.work_dir, taxdump_path=args.taxdump_path) + compute_taxon_milestones( + work_dir=args.work_dir, + taxdump_path=args.taxdump_path, + yaml_path=getattr(args, "yaml_path", None), + ) diff --git a/flows/lib/generate_assembly_summary.py b/flows/lib/generate_assembly_summary.py index 4eff2d7..36dd658 100644 --- a/flows/lib/generate_assembly_summary.py +++ b/flows/lib/generate_assembly_summary.py @@ -16,13 +16,19 @@ import os from collections import defaultdict -from flows.lib.assembly_versions_utils import parse_accession +from flows.lib.assembly_versions_utils import ( + get_accession, + get_version_status, + open_tsv, + parse_accession, + resolve_current_tsv_paths, + resolve_historical_tsv_path, +) from flows.lib.conditional_import import emit_event, flow -from flows.lib.shared_args import WORK_DIR +from flows.lib.shared_args import WORK_DIR, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args +from flows.lib.utils import load_config -CURRENT_TSV = "assembly_current.tsv" -HISTORICAL_TSV = "assembly_historical.tsv" OUTPUT_TSV = "assembly_version_summary.tsv" # EBP metric columns to check (either naming convention may appear) @@ -54,11 +60,10 @@ def load_assemblies(current_tsv: str, historical_tsv: str) -> list[dict]: print(f" Warning: {label} TSV not found: {path}") continue count = 0 - with open(path, encoding="utf-8") as f: + with open_tsv(path) as f: for row in csv.DictReader(f, delimiter="\t"): # Normalise to 'accession' so the rest of the code is uniform - if "accession" not in row or not row["accession"]: - row["accession"] = row.get("genbankAccession", "") + row["accession"] = get_accession(row) rows.append(row) count += 1 print(f" Loaded {count} {label} rows") @@ -95,11 +100,9 @@ def generate_summary_for_base(base_accession: str, rows: list[dict]) -> dict: first_row = rows[0] current_row = rows[-1] - # versionStatus column may be camelCase (Phase 0 YAML) or snake_case (Phase 1 copy) - def version_status(row): - return row.get("versionStatus") or row.get("version_status") or "current" - - superseded_count = sum(1 for r in rows if version_status(r) == "superseded") + superseded_count = sum( + 1 for r in rows if (get_version_status(r) or "current") == "superseded" + ) # First version that met EBP metric criteria first_ebp_row = next((r for r in rows if _has_ebp_metric(r)), None) @@ -141,15 +144,22 @@ def version_status(row): @flow(log_prints=True) -def generate_assembly_summary(work_dir: str = ".") -> None: +def generate_assembly_summary( + work_dir: str = ".", yaml_path: str | None = None +) -> None: """Combine current and historical TSVs into a per-base-accession summary. + The current-TSV filename comes from the config when yaml_path is given, + and is otherwise discovered in work_dir, so nothing hardcodes it. + Args: - work_dir: Directory containing assembly_current.tsv and - assembly_historical.tsv; output is written there too. + work_dir: Directory containing the current and historical TSVs; + output is written there too. + yaml_path: Optional YAML config naming the current TSV. """ - current_tsv = os.path.join(work_dir, CURRENT_TSV) - historical_tsv = os.path.join(work_dir, HISTORICAL_TSV) + config = load_config(config_file=yaml_path) if yaml_path else None + current_tsv, _, _ = resolve_current_tsv_paths(work_dir, config=config) + historical_tsv = resolve_historical_tsv_path(work_dir) output_tsv = os.path.join(work_dir, OUTPUT_TSV) separator = "=" * 80 @@ -225,7 +235,9 @@ def generate_assembly_summary(work_dir: str = ".") -> None: if __name__ == "__main__": args = _parse_args( - [WORK_DIR], + [WORK_DIR, YAML_PATH], description="Aggregate current and historical assembly TSVs into version summary", ) - generate_assembly_summary(work_dir=args.work_dir) + generate_assembly_summary( + work_dir=args.work_dir, yaml_path=getattr(args, "yaml_path", None) + ) diff --git a/flows/parsers/parse_assembly_versions.py b/flows/parsers/parse_assembly_versions.py index 5b2f4bb..081b9b3 100644 --- a/flows/parsers/parse_assembly_versions.py +++ b/flows/parsers/parse_assembly_versions.py @@ -16,34 +16,57 @@ from pathlib import Path from typing import Optional -from flows.lib.assembly_versions_utils import parse_accession +from flows.lib.assembly_versions_utils import ( + COL_ASSEMBLY_ID, + COL_SUPERSEDED_BY, + COL_SUPERSEDED_BY_VERSION, + COL_SUPERSEDED_DATE, + COL_VERSION_STATUS, + get_accession, + get_assembly_id, + open_tsv, + parse_accession, + resolve_current_tsv_paths, + resolve_historical_tsv_path, +) from flows.lib.conditional_import import flow -from flows.lib.shared_args import INPUT_PATH +from flows.lib.shared_args import INPUT_PATH, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args from flows.lib.shared_args import required -from flows.lib.utils import Parser +from flows.lib.utils import Config, Parser, load_config -def derive_assembly_version_paths(input_path: str) -> tuple[str, str]: +DELIMITER = "\t" + + +def derive_assembly_version_paths( + input_path: str, config: Optional[Config] = None +) -> tuple[str, str]: """Derive previous_tsv and historical_tsv paths from the input JSONL path. - Both files are assumed to live alongside the JSONL in the same directory, - following the same convention used by parse_ncbi_assemblies. + Both files live alongside the JSONL, following the convention used by + parse_ncbi_assemblies. The current-TSV filename — and therefore the name + of its ".previous" snapshot — comes from config.meta["file_name"] when a + config is supplied, so nothing here hardcodes it. Args: input_path (str): Path to the current assembly_data_report.jsonl. + config (Config, optional): Loaded YAML config naming the current TSV. Returns: tuple: (previous_tsv, historical_tsv) absolute paths. """ work_dir = os.path.dirname(os.path.abspath(input_path)) - previous_tsv = os.path.join(work_dir, "assembly_current.tsv.previous") - historical_tsv = os.path.join(work_dir, "assembly_historical.tsv") - return previous_tsv, historical_tsv + _, previous_tsv, _ = resolve_current_tsv_paths(work_dir, config=config) + return previous_tsv, resolve_historical_tsv_path(work_dir) def load_previous_parsed_by_base(previous_tsv: str) -> dict[str, dict[int, dict]]: """Load previous parsed results indexed by base accession and version. + The snapshot is opened through open_tsv, so a gzipped snapshot of a + gzipped current TSV is read transparently, and the accession is read via + get_accession rather than assuming a single column name. + Args: previous_tsv (str): Path to assembly_current.tsv from the previous run. @@ -55,9 +78,11 @@ def load_previous_parsed_by_base(previous_tsv: str) -> dict[str, dict[int, dict] previous_by_base: dict[str, dict[int, dict]] = {} try: - with open(previous_tsv, encoding="utf-8") as f: - for row in csv.DictReader(f, delimiter="\t"): - accession = row["accession"] + with open_tsv(previous_tsv) as f: + for row in csv.DictReader(f, delimiter=DELIMITER): + accession = get_accession(row) + if not accession: + continue base_acc, version = parse_accession(accession) if base_acc not in previous_by_base: previous_by_base[base_acc] = {} @@ -90,16 +115,21 @@ def build_superseded_row( new_version (int): Version number of the superseding assembly. release_date (str): Release date of the superseding assembly. + Emits the canonical camelCase column names declared in + assembly_historical.types.yaml, so Phase 1 rows are indistinguishable + from Phase 0 rows in the historical TSV. + Returns: - dict: Updated row with version_status, assembly_id, and superseded_by fields. + dict: Updated row with versionStatus, assemblyID and the three + supersession columns set. """ - base_acc, _ = parse_accession(previous_row["accession"]) + base_acc, _ = parse_accession(get_accession(previous_row)) row = previous_row.copy() - row["version_status"] = "superseded" - row["assembly_id"] = f"{base_acc}_{previous_version}" - row["superseded_by"] = new_accession - row["superseded_by_version"] = new_version - row["superseded_date"] = release_date + row[COL_VERSION_STATUS] = "superseded" + row[COL_ASSEMBLY_ID] = f"{base_acc}_{previous_version}" + row[COL_SUPERSEDED_BY] = new_accession + row[COL_SUPERSEDED_BY_VERSION] = new_version + row[COL_SUPERSEDED_DATE] = release_date return row @@ -189,37 +219,78 @@ def identify_newly_superseded( return newly_superseded, missing_versions +def merge_fieldnames( + rows: list[dict], preferred_order: Optional[list[str]] = None +) -> list[str]: + """Build an output header covering every key present in any row. + + Deriving the header from a single arbitrary row silently drops any column + that row happens to lack, so the union is taken across all rows. Columns + are emitted in preferred_order first (the YAML header order, or the order + already on disk), then any remaining keys in first-seen order. + + Args: + rows (list): All row dicts destined for the output file. + preferred_order (list, optional): Column order to honour where + possible. + + Returns: + list: Ordered fieldnames covering the union of all row keys. + """ + union: dict[str, None] = {} + for row in rows: + for key in row: + union.setdefault(key, None) + + order = list(preferred_order or []) + ordered = [col for col in order if col in union] + ordered += [col for col in union if col not in set(order)] + return ordered + + def append_superseded_to_tsv( - newly_superseded: list[dict], historical_tsv: str + newly_superseded: list[dict], + historical_tsv: str, + headers: Optional[list[str]] = None, ) -> None: - """Append newly superseded rows to the historical TSV, deduplicating by assembly_id. + """Merge newly superseded rows into the historical TSV, keyed by assembly ID. Reads the existing file if present, merges new rows (new rows take - precedence on duplicate assembly_id), and writes the combined result back. + precedence on duplicate assembly ID), and writes the combined result back. + The output header is the union of the columns across all rows, so merging + Phase 0 and Phase 1 row sets cannot silently drop a column. Args: newly_superseded (list): Row dicts from identify_newly_superseded. historical_tsv (str): Path to assembly_historical.tsv. + headers (list, optional): Preferred column order, e.g. + config.headers from assembly_historical.types.yaml. Defaults to + the order already present in the file on disk. """ if not newly_superseded: print(" No newly superseded versions to add.") return existing: dict[str, dict] = {} + file_order: list[str] = [] historical_path = Path(historical_tsv) if historical_path.exists(): - with open(historical_tsv, encoding="utf-8") as f: - for row in csv.DictReader(f, delimiter="\t"): - existing[row["assembly_id"]] = dict(row) + with open_tsv(historical_tsv) as f: + reader = csv.DictReader(f, delimiter=DELIMITER) + for row in reader: + existing[get_assembly_id(row)] = dict(row) + file_order = list(reader.fieldnames or []) for row in newly_superseded: - existing[row["assembly_id"]] = row + existing[get_assembly_id(row)] = row - fieldnames = list(next(iter(existing.values())).keys()) + fieldnames = merge_fieldnames( + list(existing.values()), headers or file_order + ) with open(historical_tsv, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( - f, fieldnames=fieldnames, delimiter="\t", extrasaction="ignore" + f, fieldnames=fieldnames, delimiter=DELIMITER, extrasaction="ignore" ) writer.writeheader() writer.writerows(existing.values()) @@ -241,7 +312,10 @@ def print_superseded_summary(newly_superseded: list[dict]) -> None: print(f" Found: {len(newly_superseded)} newly superseded versions.") print(" Examples:") for row in newly_superseded[:5]: - print(f" {row['accession']} -> superseded by v{row['superseded_by_version']}") + print( + f" {get_accession(row)} -> " + f"superseded by v{row[COL_SUPERSEDED_BY_VERSION]}" + ) if len(newly_superseded) > 5: print(f" ... and {len(newly_superseded) - 5} more") @@ -273,6 +347,7 @@ def parse_assembly_versions( new_jsonl: str, previous_tsv: str, historical_tsv: str, + historical_headers: Optional[list[str]] = None, ) -> dict: """Daily incremental update of the historical assembly TSV. @@ -283,6 +358,8 @@ def parse_assembly_versions( new_jsonl (str): Path to the current assembly_data_report.jsonl. previous_tsv (str): Path to assembly_current.tsv from the previous run. historical_tsv (str): Path to assembly_historical.tsv to update. + historical_headers (list, optional): Preferred column order for the + historical TSV. Defaults to the order already on disk. Returns: dict: Summary with keys newly_superseded_count, missing_versions_count, @@ -311,7 +388,9 @@ def parse_assembly_versions( print_missing_versions_warning(missing) print("\n[3/3] Updating historical TSV...") - append_superseded_to_tsv(newly_superseded, historical_tsv) + append_superseded_to_tsv( + newly_superseded, historical_tsv, headers=historical_headers + ) print(f"\n{separator}") print( @@ -328,6 +407,29 @@ def parse_assembly_versions( } +def load_assembly_versions_config(working_yaml: Optional[str]) -> Optional[Config]: + """Load the parser config, returning None when it is unavailable. + + The config is only used to resolve the current-TSV filename, so a missing + or unreadable YAML degrades to filename discovery in work_dir rather than + failing the run. + + Args: + working_yaml (str, optional): Path to the working YAML file. + + Returns: + Config or None: Loaded config, or None when it could not be loaded. + """ + if not working_yaml or not os.path.exists(working_yaml): + return None + try: + return load_config(config_file=working_yaml) + except Exception as e: + print(f" Warning: could not load {working_yaml} ({e});") + print(" falling back to filename discovery in work_dir.") + return None + + def parse_assembly_versions_wrapper( working_yaml: str, work_dir: str, @@ -337,12 +439,13 @@ def parse_assembly_versions_wrapper( ) -> None: """Wrapper matching the fetch_parse_validate parser signature. - Derives the previous TSV and historical TSV paths from work_dir and - delegates to parse_assembly_versions. + Loads working_yaml so the current-TSV filename — and the ".previous" + snapshot derived from it — comes from config.meta["file_name"] rather than + a hardcoded constant, then delegates to parse_assembly_versions. Args: - working_yaml (str): Path to the working YAML file (unused; accepted - for pipeline compatibility). + working_yaml (str): Path to the working YAML file, used to resolve the + current-TSV filename and its compression. work_dir (str): Path to the working directory containing the JSONL, the previous TSV, and the historical TSV. append (bool): Unused; accepted for pipeline compatibility. @@ -357,7 +460,10 @@ def parse_assembly_versions_wrapper( if len(paths) > 1: raise ValueError(f"More than one jsonl file found in {work_dir}") - previous_tsv, historical_tsv = derive_assembly_version_paths(paths[0]) + config = load_assembly_versions_config(working_yaml) + previous_tsv, historical_tsv = derive_assembly_version_paths( + paths[0], config=config + ) results = parse_assembly_versions( new_jsonl=paths[0], previous_tsv=previous_tsv, @@ -382,10 +488,13 @@ def plugin() -> Parser: if __name__ == "__main__": args = _parse_args( - [required(INPUT_PATH)], + [required(INPUT_PATH), YAML_PATH], description="Daily incremental update of historical assembly records", ) - previous_tsv, historical_tsv = derive_assembly_version_paths(args.input_path) + config = load_assembly_versions_config(getattr(args, "yaml_path", None)) + previous_tsv, historical_tsv = derive_assembly_version_paths( + args.input_path, config=config + ) results = parse_assembly_versions( new_jsonl=args.input_path, previous_tsv=previous_tsv, diff --git a/flows/parsers/parse_backfill_historical_versions.py b/flows/parsers/parse_backfill_historical_versions.py index 4b7acb0..79a85ef 100644 --- a/flows/parsers/parse_backfill_historical_versions.py +++ b/flows/parsers/parse_backfill_historical_versions.py @@ -1,7 +1,9 @@ -"""One-time historical backfill of superseded assembly versions from NCBI. +"""Historical backfill of superseded assembly versions from NCBI. Discovers and parses all superseded versions for assemblies with version > 1. -Run once before starting the daily incremental pipeline. +Run once before starting the daily incremental pipeline, and re-invoked by the +daily gap-fill path whenever parse_assembly_versions reports a missing version +— so the write into assembly_historical.tsv preserves the rows already there. Usage: python -m flows.parsers.parse_backfill_historical_versions \\ @@ -10,6 +12,7 @@ --work_dir tmp """ +import csv import json import os from datetime import datetime @@ -21,7 +24,10 @@ from flows.lib import utils from flows.lib.assembly_versions_utils import ( + COL_ACCESSION, find_all_assembly_versions, + get_accession, + open_tsv, parse_accession, parse_version, setup_cache_directories, @@ -30,13 +36,15 @@ from flows.lib.shared_args import INPUT_PATH, WORK_DIR, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args from flows.lib.shared_args import required -from flows.lib.utils import Config, Parser +from flows.lib.utils import Config, Parser, append_to_tsv from flows.parsers.parse_ncbi_assemblies import ( fetch_and_parse_sequence_report, process_assembly_report, write_to_tsv, ) +DELIMITER = "\t" + def parse_historical_version( version_data: dict, @@ -166,6 +174,90 @@ def identify_assemblies_needing_backfill(input_path: str) -> list[dict]: return assemblies +def resolve_output_path(config: Config, work_dir: str) -> str: + """Point the config's output at work_dir instead of the YAML's directory. + + ``config.meta["file_name"]`` is resolved relative to the YAML file, so an + out-of-tree config would write the historical TSV next to itself. Rewrite + it onto work_dir, mirroring what run_generic_tsv_parser does. + + Args: + config (Config): Loaded YAML configuration. + work_dir (str): Working directory the run was given. + + Returns: + str: The resolved output path, also stored back on the config. + """ + output_path = os.path.join( + work_dir, os.path.basename(config.meta["file_name"]) + ) + config.meta["file_name"] = output_path + return output_path + + +def load_existing_accessions(output_path: str) -> set: + """Collect the accessions already present in the historical TSV. + + Only the accession column is read, so this stays cheap on a file holding + thousands of rows. + + Args: + output_path (str): Path to assembly_historical.tsv. + + Returns: + set: Accessions already written; empty when the file does not exist. + """ + if not os.path.exists(output_path) or os.path.getsize(output_path) == 0: + return set() + + accessions = set() + with open_tsv(output_path) as f: + for row in csv.DictReader(f, delimiter=DELIMITER): + accession = get_accession(row) + if accession: + accessions.add(accession) + return accessions + + +def write_or_append_parsed(parsed: dict, config: Config) -> int: + """Write parsed rows without destroying rows already in the output. + + ``write_to_tsv`` opens the output with mode "w", which is correct for the + one-time backfill against an empty directory but destructive once the + gap-fill path re-invokes this parser daily against a populated + assembly_historical.tsv. Write the whole file only when it does not exist + yet — that branch produces the header line — and otherwise append the rows + that are not already present. + + Deduplication is required because appending is not idempotent: + find_all_assembly_versions re-parses every version below the current one, + and a completed checkpoint deliberately restarts from index 0. + + Args: + parsed (dict): Mapping of accession -> parsed row. + config (Config): Loaded YAML configuration. + + Returns: + int: Number of rows actually written. + """ + output_path = config.meta["file_name"] + existing = load_existing_accessions(output_path) + + if not existing: + write_to_tsv(parsed, config) + return len(parsed) + + new_rows = [ + row for accession, row in parsed.items() if accession not in existing + ] + skipped = len(parsed) - len(new_rows) + if skipped: + print(f" Skipping {skipped} versions already in {output_path}") + if new_rows: + append_to_tsv(config.headers, new_rows, config.meta) + return len(new_rows) + + @flow(log_prints=True) def backfill_historical_versions( input_path: str, @@ -173,12 +265,16 @@ def backfill_historical_versions( work_dir: str = ".", checkpoint_file: Optional[str] = None, ): - """One-time backfill of all historical assembly versions. + """Backfill historical assembly versions into assembly_historical.tsv. Accumulates all parsed rows in memory and writes the output TSV once at the end. Checkpoints are saved periodically so the run can be resumed after interruption but do not trigger intermediate TSV writes. + The output is written into work_dir, and rows already in the output are + preserved — the daily gap-fill path re-invokes this flow, so a truncating + write would destroy the backfill it is meant to extend. + Args: input_path (str): Path to assembly_data_report.jsonl. yaml_path (str): Path to assembly_historical.types.yaml. @@ -188,6 +284,7 @@ def backfill_historical_versions( """ setup_cache_directories(work_dir) config = utils.load_config(config_file=yaml_path) + output_path = resolve_output_path(config, work_dir) checkpoint_file = checkpoint_file or derive_checkpoint_path( input_path, yaml_path, work_dir, ) @@ -224,6 +321,7 @@ def backfill_historical_versions( parsed = {} processed = start_index + written = 0 for assembly_info in assemblies[start_index:]: base_acc = assembly_info["base_accession"] @@ -257,7 +355,7 @@ def backfill_historical_versions( version_num=version_num, current_accession=current_accession, ) - genbank_acc = row.get("genbankAccession", version_acc) + genbank_acc = row.get(COL_ACCESSION) or version_acc parsed[genbank_acc] = row print("done") except Exception as e: @@ -276,7 +374,7 @@ def backfill_historical_versions( if parsed: print(f"\nWriting {len(parsed)} records to TSV...") - write_to_tsv(parsed, config) + written = write_or_append_parsed(parsed, config) save_checkpoint(checkpoint_file, processed, completed=True) @@ -284,8 +382,9 @@ def backfill_historical_versions( print("BACKFILL COMPLETE") print(f"{'=' * 80}") print(f" Processed: {processed}/{total_assemblies} assemblies") - print(f" Records written: {len(parsed)}") - print(f" Output: {config.meta['file_name']}") + print(f" Records parsed: {len(parsed)}") + print(f" Records written: {written}") + print(f" Output: {output_path}") print("\n Next step: Run daily incremental pipeline") print(f"{'=' * 80}\n") diff --git a/tests/README_phase_0_backfill.md b/tests/README_phase_0_backfill.md index 4bac0ad..defdb7a 100644 --- a/tests/README_phase_0_backfill.md +++ b/tests/README_phase_0_backfill.md @@ -44,11 +44,10 @@ Test the complete backfill on the 3-assembly fixture already in the repo ```bash mkdir -p /tmp/assembly-versions -cp configs/assembly_historical.types.yaml /tmp/assembly-versions/ SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ --input_path tests/test_data/assembly_test_sample.jsonl \ - --yaml_path /tmp/assembly-versions/assembly_historical.types.yaml \ + --yaml_path configs/assembly_historical.types.yaml \ --work_dir /tmp/assembly-versions ``` @@ -56,18 +55,17 @@ SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ ```powershell New-Item -ItemType Directory -Force -Path C:\Temp\assembly-versions -Copy-Item configs\assembly_historical.types.yaml C:\Temp\assembly-versions\ $env:SKIP_PREFECT = "true" python -m flows.parsers.parse_backfill_historical_versions ` --input_path tests\test_data\assembly_test_sample.jsonl ` - --yaml_path C:\Temp\assembly-versions\assembly_historical.types.yaml ` + --yaml_path configs\assembly_historical.types.yaml ` --work_dir C:\Temp\assembly-versions ``` -> **Note on output path**: the output TSV (`assembly_historical.tsv`) is written -> relative to the *YAML file's directory*, not `--work_dir`. Copying the YAML -> into `--work_dir` (as above) puts the output there. +> **Note on output path**: the output TSV (`assembly_historical.tsv`) is +> written into `--work_dir`. The YAML's own directory is not used, so the +> config can be passed straight from `configs/`. Requires `datasets` CLI and network access (FTP version discovery + `datasets summary` per version). Without them each assembly reports a clean @@ -84,3 +82,9 @@ python3 -m flows.parsers.parse_backfill_historical_versions \ The run is resumable: if interrupted, re-running with the same arguments picks up from the last saved checkpoint in `/checkpoints/`. + +The write is non-destructive. When `assembly_historical.tsv` already exists the +parser appends only the versions not already in it, keyed on +`genbankAccession`, so re-running a completed backfill — or a daily gap-fill +that re-parses versions already on file — neither truncates the output nor +duplicates rows. diff --git a/tests/README_phase_1_daily_updates.md b/tests/README_phase_1_daily_updates.md index 9203f94..02f008b 100644 --- a/tests/README_phase_1_daily_updates.md +++ b/tests/README_phase_1_daily_updates.md @@ -50,10 +50,14 @@ missing_versions.json ──► update_assembly_versions ──► missing_ass ``` **`parse_assembly_versions.py`** (daily, no network): -- Reads the previous `assembly_current.tsv` (renamed `.previous` before each run) +- Reads the previous current TSV (snapshotted to `.previous` before each run); + its filename and compression come from `config.meta["file_name"]` - Reads today's JSONL; for every assembly at version > 1 checks whether its predecessor exists in the previous TSV -- If found: copies the row and marks it `superseded`, appends to `assembly_historical.tsv` +- If found: copies the row, stamps `versionStatus`, `assemblyID`, + `superseded_by`, `superseded_by_version` and `superseded_date` — the column + names declared in `assembly_historical.types.yaml` — and merges it into + `assembly_historical.tsv` - If missing: writes a record to `missing_versions.json` for the updater to handle **`update_assembly_versions.py`** (runs when `missing_versions.json` exists): @@ -69,21 +73,20 @@ both the "predecessor found" and "predecessor missing" code paths. ```bash mkdir -p /tmp/assembly-versions -cp configs/assembly_historical.types.yaml /tmp/assembly-versions/ cp tests/test_data/assembly_test_sample.jsonl /tmp/assembly-versions/assembly_data_report.jsonl ``` -> **Note on output path**: `parse_backfill_historical_versions.py` writes -> `assembly_historical.tsv` relative to the *YAML file's directory*, not -> `--work_dir`. Copying the YAML into `--work_dir` (as above) puts the output -> there. +> **Note on output path**: `parse_backfill_historical_versions.py` resolves +> `assembly_historical.tsv` against `--work_dir`, so the YAML can stay where it +> is in the repo. (Before PR-A the output landed next to the YAML instead, and +> this walkthrough copied the config into `--work_dir` to work around it.) ### Step 1 — one-time Phase 0 backfill (if not already done) ```bash SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ --input_path /tmp/assembly-versions/assembly_data_report.jsonl \ - --yaml_path /tmp/assembly-versions/assembly_historical.types.yaml \ + --yaml_path configs/assembly_historical.types.yaml \ --work_dir /tmp/assembly-versions # Writes: /tmp/assembly-versions/assembly_historical.tsv ``` @@ -95,13 +98,19 @@ run completes with 0 records written. ### Step 2 — simulate "yesterday's" parse ```bash -# Minimal previous TSV: only GCA_000222935.1 was known yesterday -printf "accession\tassembly_id\nGCA_000222935.1\tGCA_000222935_1\n" \ +# Minimal previous TSV: only GCA_000222935.1 was known yesterday. +# Column names must match what the assembly parsers actually emit. +printf "genbankAccession\tassemblyID\nGCA_000222935.1\tGCA_000222935_1\n" \ > /tmp/assembly-versions/assembly_current.tsv.previous ``` -In production this file is the `assembly_current.tsv` from the previous daily -run, renamed before the new run writes a fresh copy. +In production this file is a verbatim snapshot of the previous run's current +TSV, taken by `snapshot_previous_output` before the new run overwrites it. Its +name — and whether it is gzipped — follow `config.meta["file_name"]`, so with +the production NCBI config it is `ncbi_datasets_eukaryota.tsv.gz.previous`. +`parse_assembly_versions` resolves both from the config it is given and opens a +compressed snapshot transparently; the plain `assembly_current.tsv.previous` +above is what a config emitting `assembly_current.tsv` produces. ### Step 3 — run the daily incremental parser @@ -151,11 +160,16 @@ SKIP_PREFECT=true python3 -m flows.updaters.update_assembly_versions \ ```bash SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ --input_path /tmp/assembly-versions/missing_assembly_versions.jsonl \ - --yaml_path /tmp/assembly-versions/assembly_historical.types.yaml \ + --yaml_path configs/assembly_historical.types.yaml \ --work_dir /tmp/assembly-versions # Appends the missing predecessors to assembly_historical.tsv ``` +The append is genuine: when `assembly_historical.tsv` already exists the parser +adds only the versions it does not already hold, leaving every earlier row in +place. (Before PR-A this step rewrote the file wholesale, so a single gap-fill +would have discarded the whole backfill.) + ## Running the test suite ```bash @@ -163,6 +177,8 @@ cd "$(git rev-parse --show-toplevel)" SKIP_PREFECT=true python3 -m pytest tests/test_assembly_versions.py -v ``` -All 34 tests cover: accession parsing, path derivation, superseded-row -building, missing-version detection, TSV append/deduplication, updater JSONL -writing, and the `plugin()` registration hook. +The tests cover: accession parsing, config-driven path resolution, +superseded-row building, missing-version detection, TSV merge/deduplication, +schema conformance against `assembly_historical.types.yaml`, the +non-destructive Phase 0 write, updater JSONL writing, and the `plugin()` +registration hook. diff --git a/tests/test_assembly_versions.py b/tests/test_assembly_versions.py index 48be5a1..322b7e6 100644 --- a/tests/test_assembly_versions.py +++ b/tests/test_assembly_versions.py @@ -10,6 +10,7 @@ """ import csv +import gzip import json import os import sys @@ -18,12 +19,21 @@ import pytest +TAB = "\t" + +HISTORICAL_YAML = ( + Path(__file__).parent.parent / "configs" / "assembly_historical.types.yaml" +) + sys.path.insert(0, str(Path(__file__).parent.parent)) os.environ["SKIP_PREFECT"] = "true" -from flows.lib.utils import Parser # noqa: E402 +from flows.lib.utils import Parser, load_config # noqa: E402 from flows.parsers import parse_assembly_versions as incremental_module # noqa: E402 +from flows.parsers import ( # noqa: E402 + parse_backfill_historical_versions as backfill_module, +) from flows.parsers.parse_assembly_versions import ( # noqa: E402 append_superseded_to_tsv, build_missing_version_record, @@ -31,8 +41,21 @@ derive_assembly_version_paths, identify_newly_superseded, load_previous_parsed_by_base, + merge_fieldnames, parse_assembly_versions, ) +from flows.lib.assembly_versions_utils import ( # noqa: E402 + get_accession, + get_assembly_id, + get_version_status, + open_tsv, + resolve_current_tsv_paths, +) +from flows.parsers.parse_backfill_historical_versions import ( # noqa: E402 + load_existing_accessions, + resolve_output_path, + write_or_append_parsed, +) from flows.parsers.parse_ncbi_assemblies import snapshot_previous_output # noqa: E402 from flows.updaters import update_assembly_versions as updater_module # noqa: E402 from flows.updaters.update_assembly_versions import ( # noqa: E402 @@ -81,17 +104,17 @@ def test_missing_file_returns_empty(self, tmp_path): def test_single_version_indexed(self, tmp_path): tsv = tmp_path / "current.tsv" - write_tsv(tsv, [{"accession": "GCA_000222935.1", "taxon_id": "12345"}]) + write_tsv(tsv, [{"genbankAccession": "GCA_000222935.1", "taxId": "12345"}]) result = load_previous_parsed_by_base(str(tsv)) assert "GCA_000222935" in result assert 1 in result["GCA_000222935"] - assert result["GCA_000222935"][1]["taxon_id"] == "12345" + assert result["GCA_000222935"][1]["taxId"] == "12345" def test_multi_version_same_base(self, tmp_path): tsv = tmp_path / "current.tsv" write_tsv(tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"}, - {"accession": "GCA_000222935.2", "taxon_id": "1"}, + {"genbankAccession": "GCA_000222935.1", "taxId": "1"}, + {"genbankAccession": "GCA_000222935.2", "taxId": "1"}, ]) result = load_previous_parsed_by_base(str(tsv)) assert len(result["GCA_000222935"]) == 2 @@ -101,14 +124,42 @@ def test_multi_version_same_base(self, tmp_path): def test_multiple_base_accessions(self, tmp_path): tsv = tmp_path / "current.tsv" write_tsv(tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"}, - {"accession": "GCA_000412225.1", "taxon_id": "2"}, + {"genbankAccession": "GCA_000222935.1", "taxId": "1"}, + {"genbankAccession": "GCA_000412225.1", "taxId": "2"}, ]) result = load_previous_parsed_by_base(str(tsv)) assert len(result) == 2 assert "GCA_000222935" in result assert "GCA_000412225" in result + def test_legacy_accession_column_still_read(self, tmp_path): + """Rows written before the schema fix used a bare 'accession' column.""" + tsv = tmp_path / "current.tsv" + write_tsv(tsv, [{"accession": "GCA_000222935.1", "taxId": "1"}]) + result = load_previous_parsed_by_base(str(tsv)) + assert 1 in result["GCA_000222935"] + + def test_rows_without_accession_skipped(self, tmp_path): + tsv = tmp_path / "current.tsv" + write_tsv(tsv, [ + {"genbankAccession": "", "taxId": "1"}, + {"genbankAccession": "GCA_000222935.1", "taxId": "2"}, + ]) + result = load_previous_parsed_by_base(str(tsv)) + assert list(result) == ["GCA_000222935"] + + def test_gzipped_previous_snapshot_read(self, tmp_path): + """A gzipped current TSV yields a gzipped .previous snapshot.""" + tsv = tmp_path / "ncbi_datasets_eukaryota.tsv.gz.previous" + with gzip.open(tsv, "wt", encoding="utf-8", newline="") as f: + writer = csv.DictWriter( + f, fieldnames=["genbankAccession", "taxId"], delimiter=TAB + ) + writer.writeheader() + writer.writerow({"genbankAccession": "GCA_000222935.1", "taxId": "1"}) + result = load_previous_parsed_by_base(str(tsv)) + assert 1 in result["GCA_000222935"] + # --------------------------------------------------------------------------- # TestBuildSupersededRow @@ -119,18 +170,18 @@ class TestBuildSupersededRow: def _base_row(self): return { - "accession": "GCA_000222935.1", - "taxon_id": "12345", - "assembly_level": "Chromosome", + "genbankAccession": "GCA_000222935.1", + "taxId": "12345", + "assemblyLevel": "Chromosome", } def test_version_status_set(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") - assert row["version_status"] == "superseded" + assert row["versionStatus"] == "superseded" def test_assembly_id_format(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") - assert row["assembly_id"] == "GCA_000222935_1" + assert row["assemblyID"] == "GCA_000222935_1" def test_superseded_by_fields(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") @@ -141,12 +192,12 @@ def test_superseded_by_fields(self): def test_original_row_not_mutated(self): original = self._base_row() build_superseded_row(original, 1, "GCA_000222935.2", 2, "2024-01-15") - assert "version_status" not in original + assert "versionStatus" not in original def test_existing_fields_preserved(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") - assert row["taxon_id"] == "12345" - assert row["assembly_level"] == "Chromosome" + assert row["taxId"] == "12345" + assert row["assemblyLevel"] == "Chromosome" # --------------------------------------------------------------------------- @@ -197,7 +248,9 @@ def test_superseded_found_when_previous_version_present(self, tmp_path): tmp_path, [{"accession": "GCA_000222935.2", "releaseDate": "2024-01-15"}] ) previous = { - "GCA_000222935": {1: {"accession": "GCA_000222935.1", "taxon_id": "1"}} + "GCA_000222935": { + 1: {"genbankAccession": "GCA_000222935.1", "taxId": "1"} + } } superseded, missing = identify_newly_superseded(jsonl, previous) assert len(superseded) == 1 @@ -207,7 +260,9 @@ def test_superseded_found_when_previous_version_present(self, tmp_path): def test_missing_with_version_gap(self, tmp_path): jsonl = self._write_jsonl(tmp_path, [{"accession": "GCA_000222935.3"}]) previous = { - "GCA_000222935": {1: {"accession": "GCA_000222935.1", "taxon_id": "1"}} + "GCA_000222935": { + 1: {"genbankAccession": "GCA_000222935.1", "taxId": "1"} + } } superseded, missing = identify_newly_superseded(jsonl, previous) assert superseded == [] @@ -228,7 +283,9 @@ def test_mixed_batch(self, tmp_path): {"accession": "GCA_999999999.2"}, ]) previous = { - "GCA_000222935": {1: {"accession": "GCA_000222935.1", "taxon_id": "1"}} + "GCA_000222935": { + 1: {"genbankAccession": "GCA_000222935.1", "taxId": "1"} + } } superseded, missing = identify_newly_superseded(jsonl, previous) assert len(superseded) == 1 @@ -244,9 +301,9 @@ class TestAppendSupersededToTsv: def _make_row(self, acc, assembly_id, status="superseded"): return { - "accession": acc, - "assembly_id": assembly_id, - "version_status": status, + "genbankAccession": acc, + "assemblyID": assembly_id, + "versionStatus": status, } def test_creates_new_file(self, tmp_path): @@ -256,7 +313,7 @@ def test_creates_new_file(self, tmp_path): assert tsv.exists() result = read_tsv(tsv) assert len(result) == 1 - assert result[0]["accession"] == "GCA_000222935.1" + assert result[0]["genbankAccession"] == "GCA_000222935.1" def test_appends_to_existing(self, tmp_path): tsv = tmp_path / "historical.tsv" @@ -270,9 +327,9 @@ def test_appends_to_existing(self, tmp_path): def test_dedup_on_assembly_id_keeps_new(self, tmp_path): tsv = tmp_path / "historical.tsv" old_row = { - "accession": "GCA_000222935.1", - "assembly_id": "GCA_000222935_1", - "version_status": "superseded", + "genbankAccession": "GCA_000222935.1", + "assemblyID": "GCA_000222935_1", + "versionStatus": "superseded", "superseded_by": "GCA_000222935.2", } write_tsv(tsv, [old_row]) @@ -311,7 +368,7 @@ def test_no_previous_tsv_returns_empty_result(self, tmp_path): def test_one_superseded_produces_correct_counts(self, tmp_path): previous_tsv = tmp_path / "previous.tsv" write_tsv(previous_tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"} + {"genbankAccession": "GCA_000222935.1", "taxId": "1"} ]) jsonl = tmp_path / "new.jsonl" write_jsonl(jsonl, [ @@ -329,7 +386,7 @@ def test_missing_version_detected_in_orchestrator_result(self, tmp_path): """v3 present, v2 missing → missing_versions_count == 1.""" previous_tsv = tmp_path / "previous.tsv" write_tsv(previous_tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"} + {"genbankAccession": "GCA_000222935.1", "taxId": "1"} ]) jsonl = tmp_path / "new.jsonl" write_jsonl(jsonl, [ @@ -347,7 +404,7 @@ def test_missing_version_detected_in_orchestrator_result(self, tmp_path): def test_historical_tsv_written(self, tmp_path): previous_tsv = tmp_path / "previous.tsv" write_tsv(previous_tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"} + {"genbankAccession": "GCA_000222935.1", "taxId": "1"} ]) jsonl = tmp_path / "new.jsonl" write_jsonl(jsonl, [ @@ -362,7 +419,7 @@ def test_historical_tsv_written(self, tmp_path): assert historical_tsv.exists() rows = read_tsv(historical_tsv) assert len(rows) == 1 - assert rows[0]["version_status"] == "superseded" + assert rows[0]["versionStatus"] == "superseded" # --------------------------------------------------------------------------- @@ -563,3 +620,296 @@ def test_overwrites_stale_previous_snapshot(self, tmp_path): snapshot_previous_output(self._config(output)) assert previous.read_bytes() == output.read_bytes() + + +# --------------------------------------------------------------------------- +# TestNormalisingAccessors +# --------------------------------------------------------------------------- + +class TestNormalisingAccessors: + """get_* accessors read either naming convention (PR-A item 1).""" + + def test_accession_prefers_canonical_column(self): + row = {"genbankAccession": "GCA_000222935.1", "accession": "GCA_000412225.1"} + assert get_accession(row) == "GCA_000222935.1" + + def test_accession_falls_back_to_legacy_column(self): + assert get_accession({"accession": "GCA_000412225.1"}) == "GCA_000412225.1" + + def test_accession_absent_returns_empty(self): + assert get_accession({"taxId": "1"}) == "" + + def test_assembly_id_both_conventions(self): + assert get_assembly_id({"assemblyID": "GCA_1_1"}) == "GCA_1_1" + assert get_assembly_id({"assembly_id": "GCA_1_1"}) == "GCA_1_1" + + def test_version_status_both_conventions(self): + assert get_version_status({"versionStatus": "superseded"}) == "superseded" + assert get_version_status({"version_status": "superseded"}) == "superseded" + + def test_empty_value_falls_through_to_the_next_alias(self): + row = {"genbankAccession": "", "accession": "GCA_000412225.1"} + assert get_accession(row) == "GCA_000412225.1" + + +# --------------------------------------------------------------------------- +# TestResolveCurrentTsvPaths +# --------------------------------------------------------------------------- + +class TestResolveCurrentTsvPaths: + """The current-TSV name comes from the config, not a hardcoded constant (F4).""" + + def test_name_and_snapshot_from_config(self, tmp_path): + config = MagicMock() + config.meta = {"file_name": "configs/ncbi_datasets_eukaryota.tsv.gz"} + current, previous, open_fn = resolve_current_tsv_paths( + str(tmp_path), config=config + ) + assert current.endswith("ncbi_datasets_eukaryota.tsv.gz") + assert previous == f"{current}.previous" + assert open_fn is open_tsv + + def test_discovery_ignores_derived_outputs(self, tmp_path): + (tmp_path / "assembly_historical.tsv").touch() + (tmp_path / "assembly_version_summary.tsv").touch() + (tmp_path / "ncbi_datasets_eukaryota.tsv.gz").touch() + current, _, _ = resolve_current_tsv_paths(str(tmp_path)) + assert current.endswith("ncbi_datasets_eukaryota.tsv.gz") + + def test_fallback_name_when_nothing_on_disk(self, tmp_path): + current, previous, _ = resolve_current_tsv_paths(str(tmp_path)) + assert current.endswith("assembly_current.tsv") + assert previous.endswith("assembly_current.tsv.previous") + + def test_ambiguous_directory_raises(self, tmp_path): + (tmp_path / "one.tsv").touch() + (tmp_path / "two.tsv").touch() + with pytest.raises(ValueError): + resolve_current_tsv_paths(str(tmp_path)) + + def test_derived_paths_use_the_config_name(self, tmp_path): + config = MagicMock() + config.meta = {"file_name": "configs/ncbi_datasets_eukaryota.tsv.gz"} + jsonl = tmp_path / "assembly_data_report.jsonl" + jsonl.touch() + previous_tsv, historical_tsv = derive_assembly_version_paths( + str(jsonl), config=config + ) + assert previous_tsv.endswith("ncbi_datasets_eukaryota.tsv.gz.previous") + assert historical_tsv.endswith("assembly_historical.tsv") + + +# --------------------------------------------------------------------------- +# TestMergeFieldnames +# --------------------------------------------------------------------------- + +class TestMergeFieldnames: + """merge_fieldnames covers every column present in any row (F3).""" + + def test_union_across_rows(self): + assert set(merge_fieldnames([{"a": 1}, {"b": 2}])) == {"a", "b"} + + def test_preferred_order_honoured(self): + rows = [{"b": 1, "a": 2, "z": 3}] + assert merge_fieldnames(rows, ["a", "b"]) == ["a", "b", "z"] + + def test_preferred_columns_absent_from_rows_are_dropped(self): + assert merge_fieldnames([{"a": 1}], ["a", "b"]) == ["a"] + + def test_no_rows_gives_no_columns(self): + assert merge_fieldnames([], ["a"]) == [] + + +# --------------------------------------------------------------------------- +# TestAppendPreservesColumns +# --------------------------------------------------------------------------- + +class TestAppendPreservesColumns: + """Merging Phase 0 and Phase 1 row sets must not drop columns (F3).""" + + def test_no_column_lost_when_merging_row_sets(self, tmp_path): + tsv = tmp_path / "historical.tsv" + write_tsv(tsv, [{ + "genbankAccession": "GCA_000412225.1", + "assemblyID": "GCA_000412225_1", + "versionStatus": "superseded", + }]) + new_row = { + "genbankAccession": "GCA_000222935.1", + "assemblyID": "GCA_000222935_1", + "versionStatus": "superseded", + "superseded_by": "GCA_000222935.2", + "superseded_by_version": 2, + "superseded_date": "2024-01-15", + } + append_superseded_to_tsv([new_row], str(tsv)) + result = read_tsv(tsv) + assert len(result) == 2 + by_accession = {r["genbankAccession"]: r for r in result} + assert by_accession["GCA_000222935.1"]["superseded_by"] == "GCA_000222935.2" + assert "superseded_date" in by_accession["GCA_000412225.1"] + + def test_explicit_headers_set_the_column_order(self, tmp_path): + tsv = tmp_path / "historical.tsv" + row = { + "versionStatus": "superseded", + "genbankAccession": "GCA_000222935.1", + "assemblyID": "GCA_000222935_1", + } + headers = ["genbankAccession", "assemblyID", "versionStatus"] + append_superseded_to_tsv([row], str(tsv), headers=headers) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert written == headers + + +# --------------------------------------------------------------------------- +# TestSchemaConformance +# --------------------------------------------------------------------------- + +class TestSchemaConformance: + """Phase 1 output columns must be declared in assembly_historical.types.yaml. + + This is the check that would have caught F1-F3: Phase 1 was written and + tested against a mock schema that production never produces. + """ + + @staticmethod + def _yaml_headers(): + return load_config(config_file=str(HISTORICAL_YAML)).headers + + def test_canonical_columns_declared(self): + headers = self._yaml_headers() + assert "genbankAccession" in headers + assert "assemblyID" in headers + assert "versionStatus" in headers + + def test_supersession_columns_declared(self): + headers = self._yaml_headers() + for column in ("superseded_by", "superseded_by_version", "superseded_date"): + assert column in headers + + def test_superseded_row_columns_are_all_declared(self): + headers = set(self._yaml_headers()) + previous_row = dict.fromkeys(headers, "") + previous_row["genbankAccession"] = "GCA_000222935.1" + row = build_superseded_row(previous_row, 1, "GCA_000222935.2", 2, "2024-01-15") + assert set(row) <= headers + + def test_snake_case_columns_not_emitted(self): + row = build_superseded_row( + {"genbankAccession": "GCA_000222935.1"}, + 1, "GCA_000222935.2", 2, "2024-01-15", + ) + assert "version_status" not in row + assert "assembly_id" not in row + + def test_written_header_is_a_subset_of_the_yaml_schema(self, tmp_path): + headers = self._yaml_headers() + previous_row = dict.fromkeys(headers, "") + previous_row["genbankAccession"] = "GCA_000222935.1" + row = build_superseded_row(previous_row, 1, "GCA_000222935.2", 2, "2024-01-15") + tsv = tmp_path / "historical.tsv" + append_superseded_to_tsv([row], str(tsv), headers=headers) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert set(written) <= set(headers) + assert written == list(headers) + + +# --------------------------------------------------------------------------- +# TestResolveOutputPath +# --------------------------------------------------------------------------- + +class TestResolveOutputPath: + """Phase 0 writes into work_dir, not the YAML file directory (F6).""" + + def test_output_lands_in_work_dir(self, tmp_path): + config = load_config(config_file=str(HISTORICAL_YAML)) + path = resolve_output_path(config, str(tmp_path)) + assert os.path.dirname(path) == str(tmp_path) + assert os.path.basename(path) == "assembly_historical.tsv" + + def test_config_meta_updated_in_place(self, tmp_path): + config = load_config(config_file=str(HISTORICAL_YAML)) + path = resolve_output_path(config, str(tmp_path)) + assert config.meta["file_name"] == path + + +# --------------------------------------------------------------------------- +# TestNonDestructiveHistoricalWrite +# --------------------------------------------------------------------------- + +class TestNonDestructiveHistoricalWrite: + """Phase 0 must not truncate an existing historical TSV (F9). + + The F5 decision makes the gap-fill path re-invoke this parser daily, so a + truncating write would wipe the backfill it is meant to extend. + """ + + @staticmethod + def _config(tmp_path): + config = load_config(config_file=str(HISTORICAL_YAML)) + resolve_output_path(config, str(tmp_path)) + return config + + @staticmethod + def _row(accession): + base, version = accession.split(".") + return { + "genbankAccession": accession, + "assemblyID": f"{base}_{version}", + "versionStatus": "superseded", + } + + def test_first_write_creates_the_file_with_a_header(self, tmp_path): + config = self._config(tmp_path) + written = write_or_append_parsed( + {"GCA_000412225.1": self._row("GCA_000412225.1")}, config + ) + output = tmp_path / "assembly_historical.tsv" + assert written == 1 + with open(output, encoding="utf-8") as f: + header = next(csv.reader(f, delimiter=TAB)) + assert header == list(config.headers) + + def test_existing_rows_survive_a_gap_fill(self, tmp_path): + config = self._config(tmp_path) + output = tmp_path / "assembly_historical.tsv" + write_or_append_parsed({ + "GCA_000412225.1": self._row("GCA_000412225.1"), + "GCA_000412225.2": self._row("GCA_000412225.2"), + }, config) + assert len(read_tsv(output)) == 2 + + written = write_or_append_parsed( + {"GCA_000222935.1": self._row("GCA_000222935.1")}, config + ) + rows = read_tsv(output) + assert written == 1 + assert len(rows) == 3 + assert {r["genbankAccession"] for r in rows} == { + "GCA_000412225.1", "GCA_000412225.2", "GCA_000222935.1", + } + + def test_reparsed_versions_are_not_duplicated(self, tmp_path): + config = self._config(tmp_path) + parsed = {"GCA_000412225.1": self._row("GCA_000412225.1")} + write_or_append_parsed(parsed, config) + written = write_or_append_parsed(parsed, config) + assert written == 0 + assert len(read_tsv(tmp_path / "assembly_historical.tsv")) == 1 + + def test_load_existing_accessions_reads_the_accession_column(self, tmp_path): + output = tmp_path / "assembly_historical.tsv" + write_tsv(output, [self._row("GCA_000412225.1"), self._row("GCA_000412225.2")]) + assert load_existing_accessions(str(output)) == { + "GCA_000412225.1", "GCA_000412225.2", + } + + def test_load_existing_accessions_on_an_absent_file(self, tmp_path): + assert load_existing_accessions(str(tmp_path / "nope.tsv")) == set() + + def test_flow_routes_its_write_through_the_helper(self): + source = Path(backfill_module.__file__).read_text(encoding="utf-8") + assert "written = write_or_append_parsed(parsed, config)" in source From 55349c6ebbb8dc420f1fe3380f33460fe4b88abf Mon Sep 17 00:00:00 2001 From: Fang Chen Date: Sat, 29 Aug 2026 08:17:45 -0700 Subject: [PATCH 2/5] Read assemblyId as well as assemblyID parse_ncbi_assemblies has written processedAssemblyInfo.assemblyId (lowercase d) since 5180090, while assembly_historical.types.yaml declares assemblyID and Phase 0 writes assemblyID. Nothing is broken today - the two spellings never meet in one file - but this is the class of defect PR-A exists to remove, and it bites as soon as anything joins current and historical rows on that key. get_assembly_id now reads all three spellings. The write side deliberately stays on assemblyID: it is what the YAML declares, so a row emitted with the lowercase form would be dropped by the GenomeHubs write path. A test pins that asymmetry so a future edit cannot quietly flip the write side. Narrow by inspection - versionStatus and genbankAccession have no case variants upstream, so assemblyID is the only divergence. Which case is canonical is a question for Rich; repointing is a one-line change to COL_ASSEMBLY_ID plus the YAML once he answers. --- flows/lib/assembly_versions_utils.py | 12 ++++++++++-- tests/test_assembly_versions.py | 23 ++++++++++++++++++++++- 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/flows/lib/assembly_versions_utils.py b/flows/lib/assembly_versions_utils.py index 73a6a9f..14d1b78 100644 --- a/flows/lib/assembly_versions_utils.py +++ b/flows/lib/assembly_versions_utils.py @@ -32,7 +32,11 @@ COL_SUPERSEDED_DATE = "superseded_date" ACCESSION_ALIASES = (COL_ACCESSION, "accession") -ASSEMBLY_ID_ALIASES = (COL_ASSEMBLY_ID, "assembly_id") +# "assemblyId" is the spelling parse_ncbi_assemblies has written since +# 5180090 (2026-07-27); assembly_historical.types.yaml and Phase 0 still use +# "assemblyID". Read both until the case is settled upstream — the write side +# stays on COL_ASSEMBLY_ID, which is what the YAML declares. +ASSEMBLY_ID_ALIASES = (COL_ASSEMBLY_ID, "assemblyId", "assembly_id") VERSION_STATUS_ALIASES = (COL_VERSION_STATUS, "version_status") # Filename of the historical TSV, and the outputs derived from it. Used to @@ -71,7 +75,11 @@ def get_accession(row: dict) -> str: def get_assembly_id(row: dict) -> str: - """Return a row's assembly ID, tolerating either naming convention. + """Return a row's assembly ID across all three spellings in the pipeline. + + ``assemblyID`` is canonical (declared in assembly_historical.types.yaml and + written by Phase 0); ``assemblyId`` is what the current-assembly parser + writes upstream; ``assembly_id`` is the legacy snake_case form. Args: row (dict): A TSV row. diff --git a/tests/test_assembly_versions.py b/tests/test_assembly_versions.py index 322b7e6..6b2d2ee 100644 --- a/tests/test_assembly_versions.py +++ b/tests/test_assembly_versions.py @@ -639,10 +639,16 @@ def test_accession_falls_back_to_legacy_column(self): def test_accession_absent_returns_empty(self): assert get_accession({"taxId": "1"}) == "" - def test_assembly_id_both_conventions(self): + def test_assembly_id_all_three_spellings(self): assert get_assembly_id({"assemblyID": "GCA_1_1"}) == "GCA_1_1" + assert get_assembly_id({"assemblyId": "GCA_1_1"}) == "GCA_1_1" assert get_assembly_id({"assembly_id": "GCA_1_1"}) == "GCA_1_1" + def test_assembly_id_prefers_the_yaml_spelling(self): + """assemblyID is what the historical YAML declares, so it wins.""" + row = {"assemblyID": "canonical", "assemblyId": "upstream"} + assert get_assembly_id(row) == "canonical" + def test_version_status_both_conventions(self): assert get_version_status({"versionStatus": "superseded"}) == "superseded" assert get_version_status({"version_status": "superseded"}) == "superseded" @@ -804,6 +810,21 @@ def test_snake_case_columns_not_emitted(self): assert "version_status" not in row assert "assembly_id" not in row + def test_write_side_uses_the_yaml_spelling_of_assembly_id(self): + """Reads tolerate assemblyId; writes must stay on the declared name. + + parse_ncbi_assemblies writes `assemblyId` upstream, but the historical + YAML declares `assemblyID`, so a row emitted with the lowercase form + would be dropped by the GenomeHubs write path. + """ + row = build_superseded_row( + {"genbankAccession": "GCA_000222935.1", "assemblyId": "stale"}, + 1, "GCA_000222935.2", 2, "2024-01-15", + ) + assert row["assemblyID"] == "GCA_000222935_1" + assert "assemblyID" in self._yaml_headers() + assert "assemblyId" not in self._yaml_headers() + def test_written_header_is_a_subset_of_the_yaml_schema(self, tmp_path): headers = self._yaml_headers() previous_row = dict.fromkeys(headers, "") From 3cc32cdc1e6d0e1fa3ce417de8fcb34b6c3be0d6 Mon Sep 17 00:00:00 2001 From: Fang Chen Date: Fri, 4 Sep 2026 10:56:45 -0700 Subject: [PATCH 3/5] Read every cell through one helper, and append under the header on disk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects in this PR's own surface, found running the phases together. Phase 2 counted an ebpStandardDate of "None" as an EBP metric while Phase 3 read the same cell as absent. Upstream writes that string wherever a value was missing when the row was formatted, so on real data the summary and the milestones would claim different things about the same assembly. Every phase now reads cells through assembly_versions_utils.cell, and a test asserts the two predicates agree on "", "None" and a real date. The gap fill appended under config.headers. That is right until the Phase 1 parser rewrites the file, which widens the header to the union of every row's columns — a superseded row copied from the current TSV brings columns the historical config never declared, and appended rows then come out a few columns short. The pass that collects existing accessions now keeps the on-disk header too, and appends against that; load_existing_accessions becomes read_existing_output to match. Phase 2.2 had no unit tests. It has them now, including the cross-phase agreement on the sentinel. The pytest CI job comes with this PR rather than the last of the three: F1-F3 survived review because nobody re-ran the suite at the right moment, which is an argument for the job gating the PR that fixes them. --- .github/workflows/pytest.yml | 25 +++ flows/lib/assembly_versions_utils.py | 64 +++++- flows/lib/generate_assembly_summary.py | 42 ++-- .../parse_backfill_historical_versions.py | 33 ++- tests/test_assembly_summary.py | 190 ++++++++++++++++++ tests/test_assembly_versions.py | 41 +++- 6 files changed, 362 insertions(+), 33 deletions(-) create mode 100644 .github/workflows/pytest.yml create mode 100644 tests/test_assembly_summary.py diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml new file mode 100644 index 0000000..4868200 --- /dev/null +++ b/.github/workflows/pytest.yml @@ -0,0 +1,25 @@ +name: pytest + +on: + workflow_dispatch: + pull_request: + +jobs: + pytest: + runs-on: ubuntu-latest + name: pytest + steps: + - name: Check out source repository + uses: actions/checkout@v4 + - name: Set up Python environment + uses: actions/setup-python@v5 + with: + python-version: "3.12" + # The suite needs no network and no datasets CLI, so pip is enough: + # genomehubs pulls in the rest of what flows/lib imports. + - name: Install dependencies + run: python -m pip install --upgrade pip "genomehubs>=2.10.14" requests pytest + - name: Run tests + env: + SKIP_PREFECT: "true" + run: python -m pytest tests/ -q diff --git a/flows/lib/assembly_versions_utils.py b/flows/lib/assembly_versions_utils.py index 14d1b78..44a6798 100644 --- a/flows/lib/assembly_versions_utils.py +++ b/flows/lib/assembly_versions_utils.py @@ -6,6 +6,7 @@ current-assembly TSV paths from a loaded config rather than a hardcoded name. """ +import csv import glob as globmod import gzip import json @@ -18,6 +19,13 @@ ACCESSION_PATTERN = re.compile(r"^GC[AF]_\d{9}\.\d+$") +# Cell values that mean "no value". Upstream writes the literal string +# "None" wherever a field was absent when the row was formatted -- +# genomehubs' format_entry stringifies a bare None -- so every phase has to +# read it as empty or it becomes a date, a taxid or a metric that is not +# there. +ABSENT_VALUES = frozenset({"", "None"}) + # Leading bytes of a gzip member, used to detect compression by content. GZIP_MAGIC = b"\x1f\x8b" @@ -50,15 +58,31 @@ }) -def _first_present(row: dict, keys: tuple) -> str: - """Return the first non-empty value in ``row`` for any of ``keys``.""" +def cell(row: dict, *keys: str) -> str: + """Return the first populated cell in ``row`` among ``keys``. + + Values in ABSENT_VALUES read as empty, so a column holding the literal + string "None" is treated the same as a column holding nothing. + + Args: + row (dict): A TSV row. + *keys (str): Column names to try, in precedence order. + + Returns: + str: The first populated value, stripped, or "" when there is none. + """ for key in keys: - value = row.get(key) - if value: + value = str(row.get(key) or "").strip() + if value not in ABSENT_VALUES: return value return "" +def _first_present(row: dict, keys: tuple) -> str: + """Return the first populated value in ``row`` for any of ``keys``.""" + return cell(row, *keys) + + def get_accession(row: dict) -> str: """Return a row's versioned accession, whichever column holds it. @@ -244,6 +268,38 @@ def parse_accession(accession: str) -> tuple[str, int]: return parts[0], int(parts[1]) if len(parts) > 1 else 1 +def load_versions_by_base(tsv_path: str) -> dict[str, set[int]]: + """Index the assembly versions already present in a TSV by base accession. + + Used against assembly_historical.tsv so the daily diff can tell a genuine + gap from one that a previous run — or the Phase 0 backfill — has already + parsed. Only the accession column is read, so this stays cheap on a file + holding tens of thousands of rows, and it opens no network connections. + + Args: + tsv_path (str): Path to a TSV written by any phase. + + Returns: + dict: Mapping of base_accession -> set of version numbers present. + Empty when the file is absent or empty, which is the expected + state before the first backfill. + """ + versions_by_base: dict[str, set[int]] = {} + + if not os.path.exists(tsv_path) or os.path.getsize(tsv_path) == 0: + return versions_by_base + + with open_tsv(tsv_path) as f: + for row in csv.DictReader(f, delimiter="\t"): + accession = get_accession(row) + if not accession: + continue + base_acc, version = parse_accession(accession) + versions_by_base.setdefault(base_acc, set()).add(version) + + return versions_by_base + + def setup_cache_directories(work_dir: str) -> None: """Create cache directory structure under work_dir. diff --git a/flows/lib/generate_assembly_summary.py b/flows/lib/generate_assembly_summary.py index 36dd658..8c1109b 100644 --- a/flows/lib/generate_assembly_summary.py +++ b/flows/lib/generate_assembly_summary.py @@ -17,6 +17,7 @@ from collections import defaultdict from flows.lib.assembly_versions_utils import ( + cell, get_accession, get_version_status, open_tsv, @@ -31,13 +32,24 @@ OUTPUT_TSV = "assembly_version_summary.tsv" -# EBP metric columns to check (either naming convention may appear) -EBP_METRIC_COLUMNS = {"ebpStandardDate", "ebp_standard_date", "ebpMetricDate", "ebp_metric_date"} +# EBP metric columns to check (either naming convention may appear). A +# tuple, not a set: cell() takes them in precedence order. +EBP_METRIC_COLUMNS = ( + "ebpStandardDate", + "ebp_standard_date", + "ebpMetricDate", + "ebp_metric_date", +) def _has_ebp_metric(row: dict) -> bool: - """Return True if any EBP metric date column is populated.""" - return any(row.get(col) for col in EBP_METRIC_COLUMNS) + """Return True if any EBP metric date column is populated. + + Read through ``cell`` so the literal string "None" counts as absent, + which is what Phase 3 does with the same column. Reading it as a value + here would have the summary claim an EBP metric the milestones deny. + """ + return bool(cell(row, *EBP_METRIC_COLUMNS)) def load_assemblies(current_tsv: str, historical_tsv: str) -> list[dict]: @@ -109,18 +121,24 @@ def generate_summary_for_base(base_accession: str, rows: list[dict]) -> dict: return { "base_accession": base_accession, - "taxId": first_row.get("taxId", ""), + "taxId": cell(first_row, "taxId", "taxid"), "first_version_accession": first_row["accession"], "first_version_number": versions[0], - "first_version_date": first_row.get("releaseDate", ""), + "first_version_date": cell(first_row, "releaseDate"), "current_version_accession": current_row["accession"], "current_version_number": versions[-1], - "current_version_date": current_row.get("releaseDate", ""), + "current_version_date": cell(current_row, "releaseDate"), "total_versions": len(rows), "superseded_versions": superseded_count, - "first_ebp_metric_accession": first_ebp_row["accession"] if first_ebp_row else "", - "first_ebp_metric_version": parse_accession(first_ebp_row["accession"])[1] if first_ebp_row else "", - "first_ebp_metric_date": first_ebp_row.get("releaseDate", "") if first_ebp_row else "", + "first_ebp_metric_accession": ( + first_ebp_row["accession"] if first_ebp_row else "" + ), + "first_ebp_metric_version": ( + parse_accession(first_ebp_row["accession"])[1] if first_ebp_row else "" + ), + "first_ebp_metric_date": ( + cell(first_ebp_row, "releaseDate") if first_ebp_row else "" + ), "version_gaps": version_gaps, } @@ -182,7 +200,7 @@ def generate_assembly_summary( ) return - print(f"\n[2/3] Generating summaries...") + print("\n[2/3] Generating summaries...") by_base: dict[str, list[dict]] = defaultdict(list) for row in rows: acc = row.get("accession", "") @@ -197,7 +215,7 @@ def generate_assembly_summary( print(f" Processed {len(summaries)} unique base accessions") - print(f"\n[3/3] Writing output...") + print("\n[3/3] Writing output...") with open(output_tsv, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=SUMMARY_FIELDNAMES, delimiter="\t") writer.writeheader() diff --git a/flows/parsers/parse_backfill_historical_versions.py b/flows/parsers/parse_backfill_historical_versions.py index 79a85ef..ba56231 100644 --- a/flows/parsers/parse_backfill_historical_versions.py +++ b/flows/parsers/parse_backfill_historical_versions.py @@ -195,28 +195,37 @@ def resolve_output_path(config: Config, work_dir: str) -> str: return output_path -def load_existing_accessions(output_path: str) -> set: - """Collect the accessions already present in the historical TSV. +def read_existing_output(output_path: str) -> tuple[list, set]: + """Read the header and the accessions already in the historical TSV. - Only the accession column is read, so this stays cheap on a file holding - thousands of rows. + Only the header line and the accession column are kept, so this stays + cheap on a file holding tens of thousands of rows. + + The header matters as much as the accessions: the Phase 1 daily parser + rewrites this file with the union of the columns across every row, so once + it has appended a superseded row copied from the current TSV the file + carries columns the historical config never declared. Appending against + the config header would then write rows a few columns short. Args: output_path (str): Path to assembly_historical.tsv. Returns: - set: Accessions already written; empty when the file does not exist. + tuple: (header, accessions), both empty when the file does not exist + or holds nothing yet. """ if not os.path.exists(output_path) or os.path.getsize(output_path) == 0: - return set() + return [], set() accessions = set() with open_tsv(output_path) as f: - for row in csv.DictReader(f, delimiter=DELIMITER): + reader = csv.DictReader(f, delimiter=DELIMITER) + for row in reader: accession = get_accession(row) if accession: accessions.add(accession) - return accessions + header = list(reader.fieldnames or []) + return header, accessions def write_or_append_parsed(parsed: dict, config: Config) -> int: @@ -233,6 +242,10 @@ def write_or_append_parsed(parsed: dict, config: Config) -> int: find_all_assembly_versions re-parses every version below the current one, and a completed checkpoint deliberately restarts from index 0. + Rows are appended under the header already on disk rather than the config + header, so a file the Phase 1 parser has widened keeps its columns lined + up. A row missing one of those columns is written empty for it. + Args: parsed (dict): Mapping of accession -> parsed row. config (Config): Loaded YAML configuration. @@ -241,7 +254,7 @@ def write_or_append_parsed(parsed: dict, config: Config) -> int: int: Number of rows actually written. """ output_path = config.meta["file_name"] - existing = load_existing_accessions(output_path) + header, existing = read_existing_output(output_path) if not existing: write_to_tsv(parsed, config) @@ -254,7 +267,7 @@ def write_or_append_parsed(parsed: dict, config: Config) -> int: if skipped: print(f" Skipping {skipped} versions already in {output_path}") if new_rows: - append_to_tsv(config.headers, new_rows, config.meta) + append_to_tsv(header or config.headers, new_rows, config.meta) return len(new_rows) diff --git a/tests/test_assembly_summary.py b/tests/test_assembly_summary.py new file mode 100644 index 0000000..613ae48 --- /dev/null +++ b/tests/test_assembly_summary.py @@ -0,0 +1,190 @@ +"""Tests for generate_assembly_summary.py (Phase 2.2). + +Covers the per-base-accession aggregation: version ordering and gaps, the +superseded count, the first row to meet the EBP metric, and the treatment of +the literal string "None" that upstream writes for an absent value. + +The last of those is a cross-phase invariant rather than a local detail: +Phase 2 and Phase 3 both decide which rows carry an EBP metric, and a row the +summary counts but the milestones do not would leave the two outputs +disagreeing about the same assembly. +""" + +import csv +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +os.environ["SKIP_PREFECT"] = "true" + +from flows.lib.compute_taxon_milestones import _has_metric # noqa: E402 +from flows.lib.generate_assembly_summary import ( # noqa: E402 + OUTPUT_TSV, + SUMMARY_FIELDNAMES, + _has_ebp_metric, + find_version_gaps, + generate_assembly_summary, + generate_summary_for_base, + load_assemblies, +) + +TAXID = 9001 + + +def row(accession, version_status="current", ebp="", release=None, taxid=TAXID): + """Build a minimal assembly row as the summary reads it.""" + version = int(accession.split(".")[1]) + return { + "genbankAccession": accession, + "accession": accession, + "versionStatus": version_status, + "taxId": str(taxid), + "releaseDate": f"20{10 + version:02d}-01-01" if release is None else release, + "ebpStandardDate": ebp, + } + + +def write_tsv(path, rows): + """Write rows to a tab-separated file.""" + fieldnames = list(rows[0].keys()) + with open(path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t") + writer.writeheader() + writer.writerows(rows) + + +def read_tsv(path): + """Read a tab-separated file into a list of dicts.""" + with open(path, encoding="utf-8") as f: + return list(csv.DictReader(f, delimiter="\t")) + + +class TestVersionGaps: + def test_no_gaps_in_a_complete_run(self): + assert find_version_gaps([1, 2, 3]) == "" + + def test_reports_every_missing_version(self): + assert find_version_gaps([1, 4]) == "2,3" + + def test_a_series_starting_above_one_is_all_gap(self): + assert find_version_gaps([3]) == "1,2" + + def test_no_versions_is_not_a_gap(self): + assert find_version_gaps([]) == "" + + +class TestSummaryForBase: + def test_orders_by_version_not_by_row_order(self): + summary = generate_summary_for_base( + "GCA_1", + [row("GCA_1.3"), row("GCA_1.1", "superseded"), row("GCA_1.2", "superseded")], + ) + assert summary["first_version_accession"] == "GCA_1.1" + assert summary["current_version_accession"] == "GCA_1.3" + assert summary["total_versions"] == 3 + assert summary["superseded_versions"] == 2 + + def test_first_ebp_metric_is_the_earliest_qualifying_version(self): + summary = generate_summary_for_base( + "GCA_1", + [ + row("GCA_1.1", "superseded"), + row("GCA_1.2", "superseded", ebp="2016-01-01"), + row("GCA_1.3", ebp="2018-01-01"), + ], + ) + assert summary["first_ebp_metric_accession"] == "GCA_1.2" + assert summary["first_ebp_metric_version"] == 2 + + def test_no_ebp_metric_leaves_the_columns_empty(self): + summary = generate_summary_for_base("GCA_1", [row("GCA_1.1")]) + assert summary["first_ebp_metric_accession"] == "" + assert summary["first_ebp_metric_version"] == "" + + def test_a_row_with_no_version_status_counts_as_current(self): + summary = generate_summary_for_base( + "GCA_1", [{"accession": "GCA_1.1", "genbankAccession": "GCA_1.1"}] + ) + assert summary["superseded_versions"] == 0 + + +class TestNoneSentinel: + """Upstream writes the string "None" wherever a value was absent.""" + + def test_ebp_metric_of_none_is_not_a_metric(self): + assert _has_ebp_metric(row("GCA_1.1", ebp="None")) is False + + def test_phase_2_and_phase_3_agree_on_the_metric(self): + for value in ("", "None", "2016-01-01"): + assembly = row("GCA_1.1", ebp=value) + assert _has_ebp_metric(assembly) == _has_metric(assembly), value + + def test_a_none_release_date_is_summarised_as_empty(self): + summary = generate_summary_for_base( + "GCA_1", [row("GCA_1.1", release="None")] + ) + assert summary["first_version_date"] == "" + assert summary["current_version_date"] == "" + + def test_a_none_taxid_is_summarised_as_empty(self): + rows = [row("GCA_1.1")] + rows[0]["taxId"] = "None" + assert generate_summary_for_base("GCA_1", rows)["taxId"] == "" + + +class TestLoadAssemblies: + def test_reads_both_files_and_normalises_the_accession(self, tmp_path): + write_tsv(tmp_path / "assembly_current.tsv", [row("GCA_1.2")]) + write_tsv( + tmp_path / "assembly_historical.tsv", [row("GCA_1.1", "superseded")] + ) + rows = load_assemblies( + str(tmp_path / "assembly_current.tsv"), + str(tmp_path / "assembly_historical.tsv"), + ) + assert [r["accession"] for r in rows] == ["GCA_1.2", "GCA_1.1"] + + def test_a_missing_file_is_a_warning_not_a_failure(self, tmp_path): + write_tsv(tmp_path / "assembly_current.tsv", [row("GCA_1.1")]) + rows = load_assemblies( + str(tmp_path / "assembly_current.tsv"), + str(tmp_path / "assembly_historical.tsv"), + ) + assert len(rows) == 1 + + +class TestFlow: + @pytest.fixture + def work_dir(self, tmp_path): + write_tsv( + tmp_path / "assembly_current.tsv", + [row("GCA_1.3", ebp="2018-01-01"), row("GCA_2.1")], + ) + write_tsv( + tmp_path / "assembly_historical.tsv", + [row("GCA_1.1", "superseded")], + ) + generate_assembly_summary(work_dir=str(tmp_path)) + return tmp_path + + def test_writes_one_row_per_base_accession(self, work_dir): + rows = read_tsv(work_dir / OUTPUT_TSV) + assert {r["base_accession"] for r in rows} == {"GCA_1", "GCA_2"} + + def test_header_matches_the_schema(self, work_dir): + with open(work_dir / OUTPUT_TSV, encoding="utf-8") as f: + header = f.readline().rstrip("\n").split("\t") + assert header == SUMMARY_FIELDNAMES + + def test_the_missing_middle_version_is_reported_as_a_gap(self, work_dir): + rows = {r["base_accession"]: r for r in read_tsv(work_dir / OUTPUT_TSV)} + assert rows["GCA_1"]["version_gaps"] == "2" + assert rows["GCA_2"]["version_gaps"] == "" + + def test_no_data_writes_no_output(self, tmp_path): + generate_assembly_summary(work_dir=str(tmp_path)) + assert not (tmp_path / OUTPUT_TSV).exists() diff --git a/tests/test_assembly_versions.py b/tests/test_assembly_versions.py index 6b2d2ee..f77218b 100644 --- a/tests/test_assembly_versions.py +++ b/tests/test_assembly_versions.py @@ -20,6 +20,7 @@ import pytest TAB = "\t" +NEWLINE = "\n" HISTORICAL_YAML = ( Path(__file__).parent.parent / "configs" / "assembly_historical.types.yaml" @@ -52,7 +53,7 @@ resolve_current_tsv_paths, ) from flows.parsers.parse_backfill_historical_versions import ( # noqa: E402 - load_existing_accessions, + read_existing_output, resolve_output_path, write_or_append_parsed, ) @@ -921,15 +922,41 @@ def test_reparsed_versions_are_not_duplicated(self, tmp_path): assert written == 0 assert len(read_tsv(tmp_path / "assembly_historical.tsv")) == 1 - def test_load_existing_accessions_reads_the_accession_column(self, tmp_path): + def test_read_existing_output_reads_the_header_and_accessions(self, tmp_path): output = tmp_path / "assembly_historical.tsv" write_tsv(output, [self._row("GCA_000412225.1"), self._row("GCA_000412225.2")]) - assert load_existing_accessions(str(output)) == { - "GCA_000412225.1", "GCA_000412225.2", - } + header, accessions = read_existing_output(str(output)) + assert accessions == {"GCA_000412225.1", "GCA_000412225.2"} + assert header[0] == "genbankAccession" + + def test_read_existing_output_on_an_absent_file(self, tmp_path): + assert read_existing_output(str(tmp_path / "nope.tsv")) == ([], set()) + + def test_appends_under_a_header_phase_1_has_widened(self, tmp_path): + # Phase 1 rewrites this file with the union of every row column, so a + # gap fill must append under what is on disk, not the config header. + config = self._config(tmp_path) + output = tmp_path / "assembly_historical.tsv" + write_or_append_parsed({"GCA_000412225.1": self._row("GCA_000412225.1")}, config) - def test_load_existing_accessions_on_an_absent_file(self, tmp_path): - assert load_existing_accessions(str(tmp_path / "nope.tsv")) == set() + rows = read_tsv(output) + widened = list(rows[0].keys()) + ["genusTaxId"] + rows[0]["genusTaxId"] = "8001" + with open(output, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=widened, delimiter=TAB) + writer.writeheader() + writer.writerows(rows) + + write_or_append_parsed({"GCA_000222935.1": self._row("GCA_000222935.1")}, config) + + with open(output, encoding="utf-8") as f: + lines = [line.rstrip(NEWLINE).split(TAB) + for line in f if line.strip()] + # Every row spans the full widened header, so no column can slide. + assert {len(line) for line in lines} == {len(widened)} + appended = read_tsv(output)[-1] + assert appended["genbankAccession"] == "GCA_000222935.1" + assert appended["genusTaxId"] == "" def test_flow_routes_its_write_through_the_helper(self): source = Path(backfill_module.__file__).read_text(encoding="utf-8") From 7f33f62662d27eb95f356d05cfe040b74ac4268a Mon Sep 17 00:00:00 2001 From: Fang Chen Date: Sun, 6 Sep 2026 07:15:16 -0700 Subject: [PATCH 4/5] Decide the historical rewrite on the header, not the accessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit read_existing_output only collects non-empty accessions, so a historical TSV whose rows all lack one came back with an empty set and took the write_to_tsv branch, which opens the file with mode "w" — the very rows that made the file non-empty were discarded. The header is the exact signal instead: it is empty only when the file is missing or zero-length. Keying the branch off it sends such a file down the append path, where its rows survive. That also makes the header non-empty on every append, so the config-header fallback is now dead and goes away. --- .../parse_backfill_historical_versions.py | 12 ++++++++--- tests/test_assembly_versions.py | 21 +++++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/flows/parsers/parse_backfill_historical_versions.py b/flows/parsers/parse_backfill_historical_versions.py index ba56231..8073aa9 100644 --- a/flows/parsers/parse_backfill_historical_versions.py +++ b/flows/parsers/parse_backfill_historical_versions.py @@ -201,6 +201,10 @@ def read_existing_output(output_path: str) -> tuple[list, set]: Only the header line and the accession column are kept, so this stays cheap on a file holding tens of thousands of rows. + The header doubles as the "this file already holds data" signal: it is + empty only when the file is missing or zero-length, whereas the accession + set is also empty for a file whose rows all lack an accession. + The header matters as much as the accessions: the Phase 1 daily parser rewrites this file with the union of the columns across every row, so once it has appended a superseded row copied from the current TSV the file @@ -236,7 +240,9 @@ def write_or_append_parsed(parsed: dict, config: Config) -> int: gap-fill path re-invokes this parser daily against a populated assembly_historical.tsv. Write the whole file only when it does not exist yet — that branch produces the header line — and otherwise append the rows - that are not already present. + that are not already present. The rewrite branch keys off the header + rather than the accession set, so a file whose rows all lack an accession + is appended to rather than overwritten. Deduplication is required because appending is not idempotent: find_all_assembly_versions re-parses every version below the current one, @@ -256,7 +262,7 @@ def write_or_append_parsed(parsed: dict, config: Config) -> int: output_path = config.meta["file_name"] header, existing = read_existing_output(output_path) - if not existing: + if not header: write_to_tsv(parsed, config) return len(parsed) @@ -267,7 +273,7 @@ def write_or_append_parsed(parsed: dict, config: Config) -> int: if skipped: print(f" Skipping {skipped} versions already in {output_path}") if new_rows: - append_to_tsv(header or config.headers, new_rows, config.meta) + append_to_tsv(header, new_rows, config.meta) return len(new_rows) diff --git a/tests/test_assembly_versions.py b/tests/test_assembly_versions.py index f77218b..ed6ddd7 100644 --- a/tests/test_assembly_versions.py +++ b/tests/test_assembly_versions.py @@ -932,6 +932,27 @@ def test_read_existing_output_reads_the_header_and_accessions(self, tmp_path): def test_read_existing_output_on_an_absent_file(self, tmp_path): assert read_existing_output(str(tmp_path / "nope.tsv")) == ([], set()) + def test_rows_without_an_accession_are_not_overwritten(self, tmp_path): + # read_existing_output only collects non-empty accessions, so a file + # whose rows all lack one yields an empty accession set. The rewrite + # branch must key off the header instead, or those rows are lost. + config = self._config(tmp_path) + output = tmp_path / "assembly_historical.tsv" + write_tsv(output, [{ + "genbankAccession": "", + "assemblyID": "GCA_000412225_1", + "versionStatus": "superseded", + }]) + + written = write_or_append_parsed( + {"GCA_000222935.1": self._row("GCA_000222935.1")}, config + ) + rows = read_tsv(output) + assert written == 1 + assert len(rows) == 2 + assert rows[0]["assemblyID"] == "GCA_000412225_1" + assert rows[1]["genbankAccession"] == "GCA_000222935.1" + def test_appends_under_a_header_phase_1_has_widened(self, tmp_path): # Phase 1 rewrites this file with the union of every row column, so a # gap fill must append under what is on disk, not the config header. From 0a231b7960957590ef767f49600a4eed14b50148 Mon Sep 17 00:00:00 2001 From: Fang Chen Date: Sun, 6 Sep 2026 07:16:36 -0700 Subject: [PATCH 5/5] Keep the current TSV's schema out of the historical TSV MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Phase 1 row is copied wholesale from the current TSV, whose schema is upstream's rather than this pipeline's, and append_superseded_to_tsv unioned the columns across every row. Every current-only column therefore landed in assembly_historical.tsv despite the YAML never declaring it — upstream's assemblyId spelling among them, sitting next to the assemblyID that build_superseded_row sets, so the file carried two assembly-ID columns disagreeing about the same assembly. Three parts: - canonicalize_columns folds each alias spelling onto the column the YAML declares, and build_superseded_row copies through it. Reads still tolerate every alias; a row about to be written now carries one column per field. - When headers are supplied, they bound the schema as well as ordering it: a row may contribute the declared columns plus whatever is already on disk, and nothing else. Columns already in the file are still never dropped, which is what F3 was about. - load_historical_headers resolves assembly_historical.types.yaml from work_dir, then from the repo's configs/, and both entry points pass the result. The headers parameter existed but nothing ever supplied it, so the bound would have applied to nothing. Every failure to resolve degrades to the order already on disk: a missing schema is not a reason to fail a daily run. --- flows/lib/assembly_versions_utils.py | 65 ++++++++ flows/parsers/parse_assembly_versions.py | 70 +++++++- .../parse_backfill_historical_versions.py | 10 +- tests/test_assembly_versions.py | 153 ++++++++++++++++++ 4 files changed, 286 insertions(+), 12 deletions(-) diff --git a/flows/lib/assembly_versions_utils.py b/flows/lib/assembly_versions_utils.py index 44a6798..d0b59ed 100644 --- a/flows/lib/assembly_versions_utils.py +++ b/flows/lib/assembly_versions_utils.py @@ -47,9 +47,23 @@ ASSEMBLY_ID_ALIASES = (COL_ASSEMBLY_ID, "assemblyId", "assembly_id") VERSION_STATUS_ALIASES = (COL_VERSION_STATUS, "version_status") +# Every aliased field, canonical spelling first. canonicalize_columns walks +# this to collapse a row onto the spellings the YAML declares. +ALIASES_BY_CANONICAL = ( + ACCESSION_ALIASES, + ASSEMBLY_ID_ALIASES, + VERSION_STATUS_ALIASES, +) + # Filename of the historical TSV, and the outputs derived from it. Used to # exclude the pipeline's own products when discovering the current TSV. HISTORICAL_TSV_NAME = "assembly_historical.tsv" +HISTORICAL_YAML_NAME = "assembly_historical.types.yaml" +# configs/ sits beside flows/ in the repo, and holds the historical schema. +REPO_CONFIG_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), + "configs", +) CURRENT_TSV_DEFAULT = "assembly_current.tsv" DERIVED_TSV_NAMES = frozenset({ HISTORICAL_TSV_NAME, @@ -126,6 +140,36 @@ def get_version_status(row: dict) -> str: return _first_present(row, VERSION_STATUS_ALIASES) +def canonicalize_columns(row: dict) -> dict: + """Return a copy of ``row`` with each alias spelling folded onto its column. + + Reads tolerate the alias spellings, but a row about to be *written* to the + historical TSV must carry one column per field. A row copied from the + current TSV otherwise arrives holding upstream's ``assemblyId`` alongside + the ``assemblyID`` the caller sets, and the merged output then has two + assembly-ID columns disagreeing about the same assembly — only one of + which the historical YAML declares. + + Only fields the row actually has are touched, so this never invents an + empty column for a field the row never carried. + + Args: + row (dict): A TSV row, possibly using alias spellings. + + Returns: + dict: A copy holding only the canonical spelling of each aliased field. + """ + canonical_row = dict(row) + for aliases in ALIASES_BY_CANONICAL: + if not any(alias in row for alias in aliases): + continue + value = _first_present(row, aliases) + for alias in aliases[1:]: + canonical_row.pop(alias, None) + canonical_row[aliases[0]] = value + return canonical_row + + def is_gzipped(path: str) -> bool: """Report whether a TSV is gzip-compressed. @@ -242,6 +286,27 @@ def resolve_historical_tsv_path(work_dir: str) -> str: return os.path.join(work_dir, HISTORICAL_TSV_NAME) +def resolve_historical_yaml_path(work_dir: str) -> Optional[str]: + """Locate assembly_historical.types.yaml, or None when it is not found. + + The pipeline copies a parser's YAML into work_dir, so look there first and + fall back to the copy in the repo's configs/ directory. Returning None + rather than raising keeps the schema an optimisation: the caller falls + back to the column order already on disk. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + str or None: Path to the historical types YAML. + """ + candidates = [ + os.path.join(work_dir, HISTORICAL_YAML_NAME), + os.path.join(REPO_CONFIG_DIR, HISTORICAL_YAML_NAME), + ] + return next((path for path in candidates if os.path.exists(path)), None) + + def parse_version(accession: str) -> int: """Extract the version number from a dotted accession string. diff --git a/flows/parsers/parse_assembly_versions.py b/flows/parsers/parse_assembly_versions.py index 081b9b3..10715ca 100644 --- a/flows/parsers/parse_assembly_versions.py +++ b/flows/parsers/parse_assembly_versions.py @@ -22,12 +22,15 @@ COL_SUPERSEDED_BY_VERSION, COL_SUPERSEDED_DATE, COL_VERSION_STATUS, + HISTORICAL_YAML_NAME, + canonicalize_columns, get_accession, get_assembly_id, open_tsv, parse_accession, resolve_current_tsv_paths, resolve_historical_tsv_path, + resolve_historical_yaml_path, ) from flows.lib.conditional_import import flow from flows.lib.shared_args import INPUT_PATH, YAML_PATH @@ -117,14 +120,17 @@ def build_superseded_row( Emits the canonical camelCase column names declared in assembly_historical.types.yaml, so Phase 1 rows are indistinguishable - from Phase 0 rows in the historical TSV. + from Phase 0 rows in the historical TSV. The copied row is canonicalized + first: it comes from the current TSV, which spells the assembly ID + ``assemblyId``, and leaving that alongside the ``assemblyID`` set here + would put two disagreeing assembly-ID columns into the output. Returns: dict: Updated row with versionStatus, assemblyID and the three supersession columns set. """ base_acc, _ = parse_accession(get_accession(previous_row)) - row = previous_row.copy() + row = canonicalize_columns(previous_row) row[COL_VERSION_STATUS] = "superseded" row[COL_ASSEMBLY_ID] = f"{base_acc}_{previous_version}" row[COL_SUPERSEDED_BY] = new_accession @@ -220,7 +226,9 @@ def identify_newly_superseded( def merge_fieldnames( - rows: list[dict], preferred_order: Optional[list[str]] = None + rows: list[dict], + preferred_order: Optional[list[str]] = None, + allowed: Optional[set] = None, ) -> list[str]: """Build an output header covering every key present in any row. @@ -233,6 +241,9 @@ def merge_fieldnames( rows (list): All row dicts destined for the output file. preferred_order (list, optional): Column order to honour where possible. + allowed (set, optional): Columns permitted in the output. Keys + outside it are dropped. Defaults to no restriction, which keeps + the union whole. Returns: list: Ordered fieldnames covering the union of all row keys. @@ -240,7 +251,8 @@ def merge_fieldnames( union: dict[str, None] = {} for row in rows: for key in row: - union.setdefault(key, None) + if allowed is None or key in allowed: + union.setdefault(key, None) order = list(preferred_order or []) ordered = [col for col in order if col in union] @@ -260,12 +272,20 @@ def append_superseded_to_tsv( The output header is the union of the columns across all rows, so merging Phase 0 and Phase 1 row sets cannot silently drop a column. + When headers are supplied they also bound the schema: a new row is written + under the declared columns plus whatever is already on disk, and nothing + else. Phase 1 rows are copied wholesale from the current TSV, whose + schema is upstream's rather than this one, so without that bound every + current-only column leaks into the historical TSV. Columns already in the + file stay in it either way — F3 is about never dropping those. + Args: newly_superseded (list): Row dicts from identify_newly_superseded. historical_tsv (str): Path to assembly_historical.tsv. - headers (list, optional): Preferred column order, e.g. - config.headers from assembly_historical.types.yaml. Defaults to - the order already present in the file on disk. + headers (list, optional): The declared schema, i.e. config.headers + from assembly_historical.types.yaml. Sets both the column order + and the bound above. Defaults to the order already present in the + file on disk, unbounded. """ if not newly_superseded: print(" No newly superseded versions to add.") @@ -285,8 +305,9 @@ def append_superseded_to_tsv( for row in newly_superseded: existing[get_assembly_id(row)] = row + allowed = set(headers) | set(file_order) if headers else None fieldnames = merge_fieldnames( - list(existing.values()), headers or file_order + list(existing.values()), headers or file_order, allowed=allowed ) with open(historical_tsv, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( @@ -430,6 +451,35 @@ def load_assembly_versions_config(working_yaml: Optional[str]) -> Optional[Confi return None +def load_historical_headers(work_dir: str) -> Optional[list[str]]: + """Load the declared historical schema, returning None when unavailable. + + The headers set the column order of assembly_historical.tsv and bound + which columns a Phase 1 row may contribute, so that a row copied from the + current TSV cannot widen the file with upstream-only columns. Every + failure degrades to None — the file is then written in the order already + on disk, exactly as before — because a missing schema is not a reason to + fail a daily run. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + list or None: Declared column order from the historical types YAML. + """ + yaml_path = resolve_historical_yaml_path(work_dir) + if not yaml_path: + print(f" Note: {HISTORICAL_YAML_NAME} not found;") + print(" using the column order already on disk.") + return None + try: + return list(load_config(config_file=yaml_path).headers) + except Exception as e: + print(f" Warning: could not load {yaml_path} ({e});") + print(" using the column order already on disk.") + return None + + def parse_assembly_versions_wrapper( working_yaml: str, work_dir: str, @@ -468,6 +518,7 @@ def parse_assembly_versions_wrapper( new_jsonl=paths[0], previous_tsv=previous_tsv, historical_tsv=historical_tsv, + historical_headers=load_historical_headers(work_dir), ) if results["missing_versions_count"] > 0: @@ -499,6 +550,9 @@ def plugin() -> Parser: new_jsonl=args.input_path, previous_tsv=previous_tsv, historical_tsv=historical_tsv, + historical_headers=load_historical_headers( + os.path.dirname(os.path.abspath(historical_tsv)) + ), ) print(f"Summary: superseded={results['newly_superseded_count']}, " f"missing={results['missing_versions_count']}") diff --git a/flows/parsers/parse_backfill_historical_versions.py b/flows/parsers/parse_backfill_historical_versions.py index 8073aa9..779cb3f 100644 --- a/flows/parsers/parse_backfill_historical_versions.py +++ b/flows/parsers/parse_backfill_historical_versions.py @@ -206,10 +206,12 @@ def read_existing_output(output_path: str) -> tuple[list, set]: set is also empty for a file whose rows all lack an accession. The header matters as much as the accessions: the Phase 1 daily parser - rewrites this file with the union of the columns across every row, so once - it has appended a superseded row copied from the current TSV the file - carries columns the historical config never declared. Appending against - the config header would then write rows a few columns short. + rewrites this file with the union of the columns across every row, so the + order on disk is its output, not this config's. Phase 1 now bounds that + union by the declared schema where it can resolve the YAML, but a file + widened by an earlier run — or by a run that could not — still carries + columns the config never declared. Appending against the config header + would then write rows a few columns short. Args: output_path (str): Path to assembly_historical.tsv. diff --git a/tests/test_assembly_versions.py b/tests/test_assembly_versions.py index ed6ddd7..11472e7 100644 --- a/tests/test_assembly_versions.py +++ b/tests/test_assembly_versions.py @@ -41,16 +41,19 @@ build_superseded_row, derive_assembly_version_paths, identify_newly_superseded, + load_historical_headers, load_previous_parsed_by_base, merge_fieldnames, parse_assembly_versions, ) from flows.lib.assembly_versions_utils import ( # noqa: E402 + canonicalize_columns, get_accession, get_assembly_id, get_version_status, open_tsv, resolve_current_tsv_paths, + resolve_historical_yaml_path, ) from flows.parsers.parse_backfill_historical_versions import ( # noqa: E402 read_existing_output, @@ -839,6 +842,156 @@ def test_written_header_is_a_subset_of_the_yaml_schema(self, tmp_path): assert written == list(headers) +# --------------------------------------------------------------------------- +# TestCurrentSchemaDoesNotLeakIntoHistorical +# --------------------------------------------------------------------------- + +class TestCurrentSchemaDoesNotLeakIntoHistorical: + """A Phase 1 row is copied from the current TSV, whose schema is not ours. + + Left unbounded, every current-only column — upstream's `assemblyId` spelling + among them — lands in assembly_historical.tsv despite the YAML never + declaring it. + """ + + @staticmethod + def _yaml_headers(): + return list(load_config(config_file=str(HISTORICAL_YAML)).headers) + + @staticmethod + def _current_row(): + """A row as the current TSV spells it, extra columns and all.""" + return { + "genbankAccession": "GCA_000222935.1", + "assemblyId": "GCA_000222935_1", + "versionStatus": "current", + "releaseDate": "2011-06-01", + "sourceDatabase": "SOURCE_DATABASE_GENBANK", + } + + def test_canonicalize_columns_drops_the_alias_spelling(self): + row = canonicalize_columns( + {"assemblyId": "GCA_000222935_1", "version_status": "current"} + ) + assert row == { + "assemblyID": "GCA_000222935_1", "versionStatus": "current", + } + + def test_canonicalize_columns_prefers_the_canonical_value(self): + row = canonicalize_columns( + {"assemblyID": "canonical", "assemblyId": "stale"} + ) + assert row["assemblyID"] == "canonical" + assert "assemblyId" not in row + + def test_canonicalize_columns_invents_nothing(self): + assert canonicalize_columns({"releaseDate": "2011-06-01"}) == { + "releaseDate": "2011-06-01" + } + + def test_superseded_row_carries_one_assembly_id_column(self): + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + assert row["assemblyID"] == "GCA_000222935_1" + assert "assemblyId" not in row + + def test_undeclared_columns_are_not_written(self, tmp_path): + tsv = tmp_path / "historical.tsv" + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + append_superseded_to_tsv([row], str(tsv), headers=self._yaml_headers()) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert set(written) <= set(self._yaml_headers()) + assert "sourceDatabase" not in written + assert read_tsv(tsv)[0]["assemblyID"] == "GCA_000222935_1" + + def test_columns_already_on_disk_are_kept(self, tmp_path): + # The schema bounds what a new row may add; it must not evict a column + # an earlier run already put in the file (F3). + tsv = tmp_path / "historical.tsv" + write_tsv(tsv, [{ + "genbankAccession": "GCA_000412225.1", + "assemblyID": "GCA_000412225_1", + "legacyColumn": "keep me", + }]) + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + append_superseded_to_tsv([row], str(tsv), headers=self._yaml_headers()) + result = read_tsv(tsv) + by_accession = {r["genbankAccession"]: r for r in result} + assert by_accession["GCA_000412225.1"]["legacyColumn"] == "keep me" + assert "sourceDatabase" not in by_accession["GCA_000222935.1"] + + def test_without_headers_the_union_is_still_taken(self, tmp_path): + # No schema resolved means no bound — the pre-existing behaviour. + tsv = tmp_path / "historical.tsv" + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + append_superseded_to_tsv([row], str(tsv)) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert "sourceDatabase" in written + + def test_merge_fieldnames_honours_allowed(self): + assert merge_fieldnames( + [{"a": 1, "b": 2}], ["a", "b"], allowed={"a"} + ) == ["a"] + + +# --------------------------------------------------------------------------- +# TestHistoricalHeadersAreResolved +# --------------------------------------------------------------------------- + +class TestHistoricalHeadersAreResolved: + """The entry points must actually supply the schema, or it bounds nothing.""" + + def test_repo_config_is_found_from_an_unrelated_work_dir(self, tmp_path): + assert resolve_historical_yaml_path(str(tmp_path)) == str(HISTORICAL_YAML) + + def test_a_copy_in_work_dir_wins(self, tmp_path): + local = tmp_path / "assembly_historical.types.yaml" + local.write_text(HISTORICAL_YAML.read_text(encoding="utf-8"), encoding="utf-8") + assert resolve_historical_yaml_path(str(tmp_path)) == str(local) + + def test_headers_load_from_the_repo_config(self, tmp_path): + headers = load_historical_headers(str(tmp_path)) + assert headers == list(load_config(config_file=str(HISTORICAL_YAML)).headers) + + def test_an_unloadable_yaml_degrades_to_none(self, tmp_path): + broken = tmp_path / "assembly_historical.types.yaml" + broken.write_text("this: [is not: valid", encoding="utf-8") + assert load_historical_headers(str(tmp_path)) is None + + def test_a_missing_config_degrades_to_none(self, tmp_path, monkeypatch): + monkeypatch.setattr( + incremental_module, "resolve_historical_yaml_path", lambda _: None + ) + assert load_historical_headers(str(tmp_path)) is None + + def test_the_wrapper_passes_the_headers_through(self, tmp_path, monkeypatch): + (tmp_path / "report.jsonl").write_text("", encoding="utf-8") + captured = {} + + def fake_parse(**kwargs): + captured.update(kwargs) + return {"missing_versions_count": 0, "missing_versions": []} + + monkeypatch.setattr( + incremental_module, "parse_assembly_versions", fake_parse + ) + incremental_module.parse_assembly_versions_wrapper( + working_yaml="", work_dir=str(tmp_path), append=False + ) + assert captured["historical_headers"] == list( + load_config(config_file=str(HISTORICAL_YAML)).headers + ) + + # --------------------------------------------------------------------------- # TestResolveOutputPath # ---------------------------------------------------------------------------