Assembly version schema and path contracts (Phase 4, 1 of 3)

.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

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

flows/lib/assembly_versions_utils.py

@@ -1,18 +1,311 @@
-"""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 csv
+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+$")
 
+# 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"
+
+# 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")
+# "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")
+
+# 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,
+    "assembly_version_summary.tsv",
+    "taxon_milestone_summary.tsv",
+})
+
+
+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 = 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.
+
+    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 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.
+
+    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 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.
+
+    The name alone is not enough: ``snapshot_previous_output`` copies a
+    gzipped current TSV verbatim to ``<name>.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 ``<current>.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 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.
@@ -40,6 +333,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.