WIP: some polish on how the standards are stored and loaded.

Tests still need to be updated, as do the default standards YAML.

Author: Richard Liang

src/easyhla/easyhla.py

@@ -21,7 +21,6 @@
 from .utils import (
     BIN2NUC,
     HLA_LOCUS,
-    GroupedAllele,
     StoredHLAStandards,
     count_strict_mismatches,
     nuc2bin,
@@ -76,7 +75,7 @@ def __init__(
     @classmethod
     def use_config(
         cls,
-        standards_path: Optional[str],
+        standards_path: Optional[str] = None,
         frequencies_path: Optional[str] = None,
     ) -> "EasyHLA":
         """
@@ -112,13 +111,8 @@ def read_hla_standards(standards_io: TextIOBase) -> LoadedStandards:
             "B": {},
             "C": {},
         }
-        stored_grouped_alleles: dict[HLA_LOCUS, list[GroupedAllele]] = {
-            "A": stored_stds.A,
-            "B": stored_stds.B,
-            "C": stored_stds.C,
-        }
         for locus in ("A", "B", "C"):
-            for grouped_allele in stored_grouped_alleles[locus]:
+            for grouped_allele in stored_stds.standards[locus]:
                 hla_stds[locus][grouped_allele.name] = HLAStandard(
                     allele=grouped_allele.name,
                     two=nuc2bin(grouped_allele.exon2),

src/easyhla/models.py

@@ -1,8 +1,7 @@
 import re
 from collections.abc import Iterable
-from datetime import datetime
 from operator import itemgetter
-from typing import Optional, TypedDict
+from typing import Optional
 
 import numpy as np
 from pydantic import BaseModel, ConfigDict

src/easyhla/update_alleles.py

@@ -2,12 +2,13 @@
 
 import argparse
 import hashlib
+import json
 import logging
 import os
 import time
 from datetime import datetime
 from io import StringIO
-from typing import Final
+from typing import Final, Optional, TypedDict
 
 import Bio
 import requests
@@ -81,9 +82,13 @@
 
 # Find all releases (and their corresponding tags) of the HLA data at
 # https://github.com/ANHIG/IMGTHLA/releases
-REPO_PATH: Final[str] = os.environ.get(
-    "EASYHLA_REPO_PATH",
-    "https://raw.githubusercontent.com/ANHIG/IMGTHLA",
+REPO_OWNER: Final[str] = os.environ.get(
+    "EASYHLA_REPO_OWNER",
+    "ANHIG",
+)
+REPO_NAME: Final[str] = os.environ.get(
+    "EASYHLA_REPO_NAME",
+    "IMGTHLA",
 )
 HLA_ALLELES_FILENAME: Final[str] = os.environ.get(
     "EASYHLA_REPO_ALLELES_FILENAME",
@@ -95,21 +100,109 @@ class RetrieveAllelesError(Exception):
     pass
 
 
+class RetrieveCommitHashError(Exception):
+    pass
+
+
 def get_alleles_file(
     tag: str,
-    base_url: str = REPO_PATH,
+    repo_owner: str = REPO_OWNER,
+    repo_name: str = REPO_NAME,
     alleles_filename: str = HLA_ALLELES_FILENAME,
 ) -> str:
     """
     Retrieve the HLA alleles file from the specified tag.
     """
-    url: str = f"{base_url}/{tag}/{alleles_filename}"
-    response: requests.Response = requests.get(url)
+    url: str = (
+        f"https://api.github.com/repos/{repo_owner}/{repo_name}/"
+        f"contents/{alleles_filename}?ref={tag}"
+    )
+    response: requests.Response = requests.get(
+        url,
+        headers={
+            "Accept": "application/vnd.github.raw+json",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+    )
     if response.status_code != requests.codes.ok:
         raise RetrieveAllelesError()
     return response.text
 
 
+class CommitInfo(TypedDict):
+    sha: str
+    url: str
+
+
+class TagInfo(TypedDict):
+    name: str
+    commit: CommitInfo
+    zipball_url: str
+    tarball_url: str
+    node_id: str
+
+
+def get_commit_hash(
+    tag_name: str,
+    repo_owner: str = REPO_OWNER,
+    repo_name: str = REPO_NAME,
+) -> Optional[str]:
+    """
+    Retrieve the commit hash of the specified tag.
+    """
+    url: str = f"https://api.github.com/repos/{repo_owner}/{repo_name}/tags"
+    response: requests.Response = requests.get(
+        url,
+        headers={
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+    )
+    if response.status_code != requests.codes.ok:
+        raise RetrieveCommitHashError()
+
+    tags: list[TagInfo] = json.loads(response.text)
+    for tag in tags:
+        if tag["name"] == tag_name:
+            return tag["commit"]["sha"]
+
+    return None
+
+
+def get_from_git(tag: str) -> tuple[str, datetime, str]:
+    alleles_str: str
+    retrieval_datetime: datetime
+    for i in range(5):
+        try:
+            retrieval_datetime = datetime.now()
+            alleles_str = get_alleles_file(tag)
+        except RetrieveAllelesError:
+            if i < 4:
+                logger.info("Failed to retrieve alleles; retrying in 20 seconds....")
+                time.sleep(20)
+            else:
+                raise
+        else:
+            break
+
+    commit_hash: str
+    for i in range(5):
+        try:
+            commit_hash = get_commit_hash(tag)
+        except RetrieveCommitHashError:
+            if i < 4:
+                logger.info(
+                    "Failed to retrieve the commit hash; retrying in 20 seconds...."
+                )
+                time.sleep(20)
+            else:
+                raise
+        else:
+            break
+
+    return alleles_str, retrieval_datetime, commit_hash
+
+
 def main():
     parser: argparse.ArgumentParser = argparse.ArgumentParser(
         "Retrieve HLA alleles from IPD-IMGT/HLA."
@@ -178,18 +271,12 @@ def main():
     logger.info(f"Retrieving alleles from tag {args.tag}....")
     alleles_str: str
     retrieval_datetime: datetime
-    for i in range(5):
-        try:
-            retrieval_datetime = datetime.now()
-            alleles_str = get_alleles_file(args.tag)
-        except RetrieveAllelesError:
-            if i < 4:
-                logger.info("Failed to retrieve alleles; retrying in 20 seconds....")
-                time.sleep(20)
-            else:
-                raise
-        else:
-            break
+    commit_hash: str
+    alleles_str, retrieval_datetime, commit_hash = get_from_git(args.tag)
+    logger.info(
+        f"Alleles (version {args.tag}, commit hash {commit_hash}) retrieved at "
+        f"{retrieval_datetime}."
+    )
 
     if args.dump_full_fasta_to != "":
         logger.info(f"Dumping the full FASTA file to {args.dump_full_fasta_to}.")
@@ -214,8 +301,9 @@ def main():
     logger.info("Identifying identical HLA alleles....")
     standards_for_saving: StoredHLAStandards = StoredHLAStandards(
         tag=args.tag,
+        commit_hash=commit_hash,
         last_updated=retrieval_datetime,
-        **{
+        standards={
             locus: group_identical_alleles(raw_standards[locus])
             for locus in ("A", "B", "C")
         },

src/easyhla/utils.py

@@ -1,14 +1,15 @@
+import hashlib
 import logging
 import re
 from collections import defaultdict
 from collections.abc import Iterable, Sequence
 from datetime import datetime
 from operator import attrgetter
-from typing import Final, Literal, Optional
+from typing import Final, Literal, Optional, Self
 
 import numpy as np
 from Bio.SeqIO import SeqRecord
-from pydantic import BaseModel, computed_field
+from pydantic import BaseModel, computed_field, model_validator
 
 # A lookup table of translations from ambiguous nucleotides to unambiguous
 # nucleotides.
@@ -490,9 +491,44 @@ def group_identical_alleles(
     return sorted(grouped_alleles, key=attrgetter("name"))
 
 
+def compute_stored_standard_checksum(
+    tag: str,
+    commit_hash: str,
+    last_updated: datetime,
+    alleles: dict[HLA_LOCUS, list[GroupedAllele]],
+) -> str:
+    """
+    Compute a checksum for the stored data.
+    """
+    stored_string: str = f"{tag}\n{commit_hash}\n{last_updated}\n"
+    for locus in ("A", "B", "C"):
+        for ga in alleles[locus]:
+            stored_string += f"{ga.name},{ga.exon2},{ga.exon3},{';'.join(ga.alleles)}\n"
+
+    # Compute the checksum.
+    sha256_calc = hashlib.sha256()
+    sha256_calc.update(stored_string.encode())
+    return sha256_calc.hexdigest()
+
+
 class StoredHLAStandards(BaseModel):
     tag: str
+    commit_hash: str
     last_updated: datetime
-    A: list[GroupedAllele]
-    B: list[GroupedAllele]
-    C: list[GroupedAllele]
+    standards: dict[HLA_LOCUS, list[GroupedAllele]]
+    checksum: Optional[str] = None
+
+    @model_validator(mode="after")
+    def compute_compare_checksum(self) -> Self:
+        checksum: str = compute_stored_standard_checksum(
+            self.tag,
+            self.commit_hash,
+            self.last_updated,
+            self.standards,
+        )
+
+        if self.checksum is None:
+            self.checksum = checksum
+        else:
+            if self.checksum != checksum:
+                raise ValueError("Checksum mismatch")