Download and cache HGNC aliases in one call

docs/guides/aliases.md

@@ -1,22 +1,44 @@
 # Look up gene name aliases
 
-Exact-name lookup uses the symbols in the selected annotation. To also search
-aliases and previous human symbols, download an [HGNC complete-set TSV
-snapshot](https://www.genenames.org/download/) and load it explicitly:
+Exact-name lookup uses the symbols in your annotation. To search aliases and
+previous human symbols too, load the [HGNC complete set](https://www.genenames.org/download/)
+with one call. PyEnsembl downloads it once and reuses the cached file:
 
 ```python
 from pyensembl import EnsemblRelease, GeneNameAliases
 
 data = EnsemblRelease(93, species="human")  # install this release first
-aliases = GeneNameAliases.from_hgnc("hgnc_complete_set.txt")
+aliases = GeneNameAliases.download_hgnc()
 genes = data.genes_by_name("p53", aliases=aliases)
+print([gene.name for gene in genes])
+data.close()
+```
+
+```text
+['TP53']
 ```
 
 Names are case-sensitive, including HGNC's lowercase `p53` alias. Alias data
-is read locally. Keep the snapshot with your analysis: current HGNC
-nomenclature can differ from an older Ensembl release. Only genes present in
+is human-only and comes from HGNC independently of Ensembl. Only genes present in
 the annotation are returned, with their original IDs. Ambiguous aliases return
 all matching genes, including exact-name matches; do not assume the first one
-is the intended locus. HGNC data is human-only. For other species or custom
-annotations, pass a mapping such as `aliases={"old_name": ["gene_id"]}`.
+is the intended locus.
 `gene_ids_of_gene_name(..., aliases=aliases)` returns just the IDs.
+
+## Reuse or select a snapshot
+
+Cached aliases work offline. They are stored under
+`aliases/homo_sapiens/hgnc` in the [PyEnsembl cache](cache.md#cache-location).
+Use `cache_directory_path=` to choose another directory, or
+`overwrite=True` to refresh the current complete set explicitly.
+
+For reproducible work, keep the snapshot with your analysis. Current HGNC names
+can differ from an older Ensembl release. Load a selected local snapshot with
+`GeneNameAliases.from_hgnc("hgnc_complete_set.txt")`, or pass an
+[archived HGNC TSV URL](https://www.genenames.org/download/)
+as `source_url=` to `download_hgnc()`. Plain and gzip TSV files are supported.
+
+## Other species or custom names
+
+Pass your own name-to-gene-ID mapping, for example
+`data.genes_by_name("old_name", aliases={"old_name": ["gene_id"]})`.

pyensembl/gene_name_aliases.py

@@ -3,9 +3,19 @@
 from collections.abc import Mapping
 import csv
 import gzip
+from hashlib import sha256
+from io import TextIOWrapper
 import os
 import re
 
+from .download_cache import DownloadCache, cache_root
+
+
+HGNC_COMPLETE_SET_URL = (
+    "https://storage.googleapis.com/public-download-files/"
+    "hgnc/tsv/tsv/hgnc_complete_set.txt"
+)
+
 
 def normalize_aliases(aliases):
     """Copy an alias mapping, treating one string ID as one value."""
@@ -54,6 +64,48 @@ def __iter__(self):
     def __len__(self):
         return len(self._aliases)
 
+    @classmethod
+    def download_hgnc(cls, *, cache_directory_path=None,
+                      source_url=HGNC_COMPLETE_SET_URL, overwrite=False):
+        """Download and load human HGNC aliases, reusing a cached snapshot.
+
+        Parameters
+        ----------
+        cache_directory_path : str or Path, optional
+            Storage base directory. Defaults to aliases/homo_sapiens/hgnc
+            under PyEnsembl's configured cache root. Each source URL has
+            a separate subdirectory.
+        source_url : str, optional
+            Complete-set TSV or gzip TSV URL. Defaults to HGNC's current
+            complete set; an archived snapshot or mirror can be selected.
+        overwrite : bool, optional
+            Download again even when this source is already cached.
+
+        Returns
+        -------
+        GeneNameAliases
+            Human alias mapping whose source is the cached file path.
+
+        Notes
+        -----
+        Existing cached files are read without network access. Refresh is
+        explicit; HGNC nomenclature is independent of the Ensembl release.
+        Use from_hgnc(path) to load an already selected local snapshot.
+        """
+        if not isinstance(source_url, str) or "://" not in source_url:
+            raise ValueError("HGNC source_url must be a URL; use from_hgnc(path) for a local file")
+        if cache_directory_path is None:
+            cache_directory_path = os.path.join(cache_root(), "aliases", "homo_sapiens", "hgnc")
+        source_key = sha256(source_url.encode("utf-8")).hexdigest()[:16]
+        cache = DownloadCache(
+            reference_name="homo_sapiens", annotation_name="hgnc",
+            cache_directory_path=os.path.join(os.fspath(cache_directory_path), source_key),
+        )
+        path = cache.download_or_copy_if_necessary(
+            source_url, download_if_missing=True, overwrite=overwrite,
+        )
+        return cls.from_hgnc(path)
+
     @classmethod
     def from_hgnc(cls, path):
         """Load an HGNC complete-set TSV snapshot (plain or gzip).
@@ -63,26 +115,27 @@ def from_hgnc(cls, path):
         HGNC nomenclature may be newer than the selected Ensembl annotation.
         """
         path = os.fspath(path)
-        opener = gzip.open if path.endswith(".gz") else open
         aliases = {}
         required = {"symbol", "alias_symbol", "prev_symbol", "ensembl_gene_id", "status"}
-        with opener(path, "rt", encoding="utf-8-sig", newline="") as stream:
-            reader = csv.DictReader(stream, delimiter="\t", strict=True)
-            if not required.issubset(reader.fieldnames or ()):
-                raise ValueError("HGNC TSV is missing required columns: %s" %
-                                 ", ".join(sorted(required - set(reader.fieldnames or ()))))
-            for row in reader:
-                if None in row or any(row[column] is None for column in required):
-                    raise ValueError("Malformed HGNC TSV row %d" % reader.line_num)
-                if row["status"] != "Approved" or not row["ensembl_gene_id"].strip():
-                    continue
-                ids = [value.strip() for value in row["ensembl_gene_id"].split("|")]
-                if any(not re.fullmatch(r"ENSG\d+(?:\.\d+)?", value) for value in ids):
-                    raise ValueError("Invalid Ensembl gene ID in HGNC row %d" % reader.line_num)
-                names = [row["symbol"], *row["alias_symbol"].split("|"),
-                         *row["prev_symbol"].split("|")]
-                for name in names:
-                    name = name.strip()
-                    if name:
-                        aliases.setdefault(name, set()).update(ids)
+        with open(path, "rb") as raw:
+            binary = gzip.GzipFile(fileobj=raw) if raw.peek(2)[:2] == b"\x1f\x8b" else raw
+            with TextIOWrapper(binary, encoding="utf-8-sig", newline="") as stream:
+                reader = csv.DictReader(stream, delimiter="\t", strict=True)
+                if not required.issubset(reader.fieldnames or ()):
+                    raise ValueError("HGNC TSV is missing required columns: %s" %
+                                     ", ".join(sorted(required - set(reader.fieldnames or ()))))
+                for row in reader:
+                    if None in row or any(row[column] is None for column in required):
+                        raise ValueError("Malformed HGNC TSV row %d" % reader.line_num)
+                    if row["status"] != "Approved" or not row["ensembl_gene_id"].strip():
+                        continue
+                    ids = [value.strip() for value in row["ensembl_gene_id"].split("|")]
+                    if any(not re.fullmatch(r"ENSG\d+(?:\.\d+)?", value) for value in ids):
+                        raise ValueError("Invalid Ensembl gene ID in HGNC row %d" % reader.line_num)
+                    names = [row["symbol"], *row["alias_symbol"].split("|"),
+                             *row["prev_symbol"].split("|")]
+                    for name in names:
+                        name = name.strip()
+                        if name:
+                            aliases.setdefault(name, set()).update(ids)
         return cls(aliases, species="homo_sapiens", source=os.path.abspath(path))

pyensembl/version.py

@@ -1,4 +1,4 @@
-__version__ = "2.22.3"
+__version__ = "2.23.0"
 
 def print_version():
     print(f"v{__version__}")

tests/test_gene_name_aliases.py

@@ -1,8 +1,13 @@
 """Alias lookup is explicit, ambiguity-preserving, and annotation-scoped."""
 
 import gzip
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+import threading
 
+import datacache
 import pytest
+from requests.exceptions import HTTPError
 
 from pyensembl import EnsemblRelease, GeneNameAliases, Genome
 
@@ -108,3 +113,115 @@ def test_invalid_hgnc_data_rejected(tmp_path, text):
     path.write_text(text)
     with pytest.raises(ValueError):
         GeneNameAliases.from_hgnc(path)
+
+
+@pytest.fixture
+def hgnc_server():
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def do_GET(self):
+            self.server.requests.append(self.path)
+            payload = self.server.payloads.get(self.path)
+            if payload is None:
+                self.send_error(404)
+                return
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    server.payloads = {"/hgnc.tsv": HGNC.encode()}
+    server.requests = []
+    server.url = "http://127.0.0.1:%d" % server.server_address[1]
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield server
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join()
+
+
+def test_download_hgnc_indexes_aliases_and_reuses_snapshot_offline(
+    genome, hgnc_server, tmp_path, monkeypatch,
+):
+    source_url = hgnc_server.url + "/hgnc.tsv"
+    options = dict(source_url=source_url, cache_directory_path=tmp_path / "aliases")
+    aliases = GeneNameAliases.download_hgnc(**options)
+    assert hgnc_server.requests == ["/hgnc.tsv"]
+    assert genome.gene_ids_of_gene_name("LFS1", aliases=aliases) == ["ENSG00000141510.7"]
+    assert Path(aliases.source).read_text() == HGNC
+    assert datacache.inspect_file(aliases.source).source_url == source_url
+
+    def no_network(*args, **kwargs):
+        raise AssertionError("Cached aliases must not download again")
+
+    monkeypatch.setattr(datacache, "fetch_file", no_network)
+    cached = GeneNameAliases.download_hgnc(**options)
+    assert cached.source == aliases.source
+    assert dict(cached) == dict(aliases)
+
+
+def test_download_hgnc_refresh_is_explicit(hgnc_server, tmp_path):
+    options = dict(source_url=hgnc_server.url + "/hgnc.tsv", cache_directory_path=tmp_path)
+    original = GeneNameAliases.download_hgnc(**options)
+    hgnc_server.payloads["/hgnc.tsv"] = HGNC.replace("LFS1", "NEW_ALIAS").encode()
+    cached = GeneNameAliases.download_hgnc(**options)
+    assert "LFS1" in cached and "NEW_ALIAS" not in cached
+    assert len(hgnc_server.requests) == 1
+    refreshed = GeneNameAliases.download_hgnc(**options, overwrite=True)
+    assert "NEW_ALIAS" in refreshed and "LFS1" not in refreshed
+    assert "LFS1" in original
+    assert len(hgnc_server.requests) == 2
+
+
+def test_download_hgnc_separates_urls_with_the_same_filename(hgnc_server, tmp_path):
+    hgnc_server.payloads["/archive/hgnc.tsv"] = HGNC.replace("LFS1", "OLD_ALIAS").encode()
+    current = GeneNameAliases.download_hgnc(
+        source_url=hgnc_server.url + "/hgnc.tsv", cache_directory_path=tmp_path,
+    )
+    archived = GeneNameAliases.download_hgnc(
+        source_url=hgnc_server.url + "/archive/hgnc.tsv", cache_directory_path=tmp_path,
+    )
+    assert current.source != archived.source
+    assert "LFS1" in current and "OLD_ALIAS" not in current
+    assert "OLD_ALIAS" in archived and "LFS1" not in archived
+    assert Path(current.source).read_text() == HGNC
+
+
+@pytest.mark.parametrize("url_path", ["/hgnc.tsv.gz", "/hgnc.tsv.gz?download=1", "/snapshot"])
+def test_download_hgnc_supports_gzip_snapshots(hgnc_server, tmp_path, url_path):
+    hgnc_server.payloads[url_path] = gzip.compress(HGNC.encode())
+    aliases = GeneNameAliases.download_hgnc(
+        source_url=hgnc_server.url + url_path, cache_directory_path=tmp_path,
+    )
+    assert aliases["LFS1"] == ("ENSG00000141510",)
+    assert Path(aliases.source).read_bytes()[:2] == b"\x1f\x8b"
+
+
+def test_download_hgnc_respects_configured_cache_root(hgnc_server, tmp_path, monkeypatch):
+    monkeypatch.setenv("PYENSEMBL_CACHE_DIR", str(tmp_path))
+    aliases = GeneNameAliases.download_hgnc(source_url=hgnc_server.url + "/hgnc.tsv")
+    base = tmp_path / "pyensembl" / "aliases" / "homo_sapiens" / "hgnc"
+    assert Path(aliases.source).is_relative_to(base)
+
+
+def test_failed_hgnc_refresh_preserves_cached_snapshot(hgnc_server, tmp_path):
+    options = dict(source_url=hgnc_server.url + "/hgnc.tsv", cache_directory_path=tmp_path)
+    aliases = GeneNameAliases.download_hgnc(**options)
+    del hgnc_server.payloads["/hgnc.tsv"]
+    with pytest.raises(HTTPError):
+        GeneNameAliases.download_hgnc(**options, overwrite=True)
+    assert Path(aliases.source).read_text() == HGNC
+    assert "LFS1" in GeneNameAliases.download_hgnc(**options)
+
+
+@pytest.mark.parametrize("source_url", [None, "", "/data/hgnc.tsv"])
+def test_download_hgnc_rejects_local_paths_and_missing_urls(source_url, tmp_path):
+    with pytest.raises(ValueError, match="source_url must be a URL"):
+        GeneNameAliases.download_hgnc(source_url=source_url, cache_directory_path=tmp_path)
+    assert not list(tmp_path.iterdir())