feat: remove http based file ingest

src/anyvlm/anyvar/base_client.py

@@ -3,10 +3,11 @@
 import abc
 from collections.abc import Iterable, Sequence
 
-from anyvar.core.objects import SupportedVrsVariation
 from anyvar.mapping.liftover import ReferenceAssembly
 from ga4gh.vrs.models import Allele
 
+from anyvlm.anyvar.types import SupportedVrsVariation
+
 
 class AnyVarClientError(Exception):
     """Generic client-related exception."""

src/anyvlm/anyvar/python_client.py

@@ -5,7 +5,6 @@
 
 from anyvar import AnyVar
 from anyvar.core.metadata import VariationMapping, VariationMappingType
-from anyvar.core.objects import SupportedVrsVariation
 from anyvar.mapping.liftover import ReferenceAssembly
 from anyvar.restapi.schema import SupportedVariationType
 from anyvar.storage.base import Storage
@@ -14,6 +13,7 @@
 from ga4gh.vrs.models import Allele
 
 from anyvlm.anyvar.base_client import BaseAnyVarClient
+from anyvlm.anyvar.types import SupportedVrsVariation
 
 _logger = logging.getLogger(__name__)
 

src/anyvlm/anyvar/types.py

@@ -0,0 +1,13 @@
+"""Compatibility aliases for AnyVar/VRS object types.
+
+These names preserve AnyVLM's broader type intent even when upstream AnyVar stops
+exporting its convenience aliases.
+"""
+
+from ga4gh.vrs.models import Allele
+
+try:
+    from anyvar.core.objects import SupportedVrsObject, SupportedVrsVariation
+except ImportError:
+    SupportedVrsObject = Allele
+    SupportedVrsVariation = Allele

src/anyvlm/cli.py

@@ -1,19 +1,29 @@
 """CLI for interacting with AnyVLM instance"""
 
 import logging
-from http import HTTPStatus
 from pathlib import Path
 from timeit import default_timer as timer
 
 import click
-import requests
 from anyvar.mapping.liftover import ReferenceAssembly
 
 import anyvlm
+from anyvlm.anyvar.base_client import BaseAnyVarClient
 from anyvlm.config import Settings, get_config
+from anyvlm.functions.ingest_vcf import ingest_vcf as ingest_vcf_function
+from anyvlm.main import create_anyvar_client, create_anyvlm_storage
+from anyvlm.storage import Storage
+
+# Create alias for easier mocking in tests
+ingest_vcf = ingest_vcf_function
 
 _logger = logging.getLogger(__name__)
 
+# Constants
+MAX_FILE_SIZE = 5 * 1024 * 1024 * 1024  # 5GB
+UPLOAD_CHUNK_SIZE = 1024 * 1024  # 1MB
+REQUIRED_INFO_FIELDS = {"AC", "AN", "AC_Het", "AC_Hom", "AC_Hemi"}
+
 
 @click.version_option(anyvlm.__version__)
 @click.group()
@@ -22,10 +32,10 @@ def _cli() -> None:
     logging.basicConfig(filename="anyvlm.log", level=logging.INFO)
 
 
-@_cli.command()
+@_cli.command(name="ingest-vcf")
 @click.option(
     "--file",
-    "vcf_path",
+    "vcf_file_path",
     type=click.Path(exists=True, dir_okay=False, path_type=Path),
     required=True,
     help="Path to a gzip-compressed VCF file (.vcf.gz)",
@@ -39,44 +49,31 @@ def _cli() -> None:
     callback=lambda _, __, value: ReferenceAssembly(value),
     help="Reference genome assembly",
 )
-def ingest_vcf(vcf_path: Path, assembly: ReferenceAssembly) -> None:
+def ingest_vcf_cli_wrapper(vcf_file_path: Path, assembly: ReferenceAssembly) -> None:
     """Deposit variants and allele frequencies from VCF into AnyVLM instance
 
     $ anyvlm ingest-vcf --file path/to/file.vcf.gz --assembly grch38
     """
     start: float = timer()
-
     _logger.info(
         "Starting VCF ingestion: file='%s', assembly='%s'",
-        str(vcf_path),
+        str(vcf_file_path),
         assembly.value,
     )
 
     config: Settings = get_config()
-    endpoint: str = f"{config.service_uri}/ingest_vcf"
-
-    params = {"assembly": assembly.value}
-
-    with vcf_path.open("rb") as fh:
-        files = {"file": (vcf_path.name, fh, "application/gzip")}
-
-        try:
-            response: requests.Response = requests.post(
-                endpoint,
-                files=files,
-                params=params,
-                timeout=3600,  # 1 hour
-            )
-        except requests.RequestException as e:
-            _logger.exception("HTTP POST request to AnyVLM '/ingest_vcf' failed")
-            raise click.ClickException(str(e)) from e
-
-    if response.status_code != HTTPStatus.OK:
-        _logger.error("Request failed with status code %s", response.status_code)
-        raise click.ClickException(
-            f"Request failed with status code: {response.status_code}"
-        )
+    anyvar_client: BaseAnyVarClient = create_anyvar_client(
+        connection_string=config.anyvar_uri
+    )
+    anyvlm_storage: Storage = create_anyvlm_storage(uri=config.storage_uri)
+    ingest_vcf_function(
+        vcf_path=vcf_file_path,
+        av=anyvar_client,
+        storage=anyvlm_storage,
+        assembly=assembly,
+    )
 
     end: float = timer()
     duration: float = end - start
     _logger.info("Ingestion complete in %s", f"{duration:.3f} seconds")
+    print("✅ Ingestion complete")  # noqa: T201

src/anyvlm/functions/ingest_vcf.py

@@ -1,14 +1,17 @@
 """Get a VCF, register its contained variants, and add cohort frequency data to storage"""
 
 import logging
-from collections import namedtuple
 from collections.abc import Iterator
+from enum import StrEnum
+from logging import Logger
 from pathlib import Path
+from typing import NamedTuple
 
 import pysam
 from anyvar.mapping.liftover import ReferenceAssembly
 from ga4gh.core.models import iriReference
 from ga4gh.va_spec.base import StudyGroup
+from pysam.libcbcf import VariantRecordInfo
 
 from anyvlm.anyvar.base_client import BaseAnyVarClient
 from anyvlm.storage.base_storage import Storage
@@ -18,16 +21,47 @@
     QualityMeasures,
 )
 
-_logger = logging.getLogger(__name__)
+_logger: Logger = logging.getLogger(__name__)
 
 
-AfData = namedtuple("AfData", ("ac", "an", "ac_het", "ac_hom", "ac_hemi", "filters"))
+class VcfInfoField(StrEnum):
+    """Represents required info fields"""
+
+    AC = "AC"
+    AN = "AN"
+    AC_HET = "AC_Het"
+    AC_HOM = "AC_Hom"
+    AC_HEMI = "AC_Hemi"
+
+
+REQUIRED_INFO_FIELDS: frozenset[VcfInfoField] = frozenset(VcfInfoField)
+
+
+class AfData(NamedTuple):
+    """Represents Af data"""
+
+    ac: int
+    an: int
+    ac_het: int
+    ac_hom: int
+    ac_hemi: int
+    filters: object
 
 
 class VcfAfColumnsError(Exception):
     """Raise for missing VCF INFO columns that are required for AF ingestion"""
 
 
+def _validate_vcf_header(vcf: pysam.VariantFile) -> None:
+    """Validate that a VCF header includes the required INFO fields."""
+    found_fields: set[str] = set[str](vcf.header.info.keys())
+    missing: frozenset[VcfInfoField] = REQUIRED_INFO_FIELDS - found_fields
+    if missing:
+        raise VcfAfColumnsError(
+            f"VCF ingestion failed: missing required INFO fields: {', '.join(sorted(missing))}"
+        )
+
+
 def _yield_expression_af_batches(
     vcf: pysam.VariantFile, batch_size: int = 1000
 ) -> Iterator[list[tuple[str, AfData]]]:
@@ -47,9 +81,9 @@ def _yield_expression_af_batches(
             if record.ref is None or "*" in record.ref or "*" in alt:
                 _logger.info("Skipping missing allele at %s", record)
                 continue
-            expression = f"{record.chrom}-{record.pos}-{record.ref}-{alt}"
+            expression: str = f"{record.chrom}-{record.pos}-{record.ref}-{alt}"
             try:
-                af = AfData(
+                af: AfData = AfData(
                     ac=record.info["AC"][i],
                     an=record.info["AN"],
                     ac_het=record.info["AC_Het"][i],
@@ -58,8 +92,8 @@ def _yield_expression_af_batches(
                     filters=record.filter.keys(),
                 )
             except KeyError as e:
-                info = record.info
-                msg = f"One or more required INFO column is missing: {'AC' in info}, {'AN' in info}, {'AC_Het' in info}, {'AC_Hom' in info}, {'AC_Hemi' in info}"
+                info: VariantRecordInfo = record.info
+                msg: str = f"One or more required INFO column is missing: {VcfInfoField.AC in info}, {VcfInfoField.AN in info}, {VcfInfoField.AC_HET in info}, {VcfInfoField.AC_HOM in info}, {VcfInfoField.AC_HEMI in info}"
                 _logger.exception(msg)
                 raise VcfAfColumnsError(msg) from e
             if af.an == 0:
@@ -101,16 +135,27 @@ def ingest_vcf(
     :param av: AnyVar client
     :param storage: AnyVLM storage instance
     :param assembly: reference assembly used by VCF
+    :raise ValueError: if VCF is unreadable or missing required INFO fields
     :raise VcfAfColumnsError: if VCF is missing required columns
     """
     pysam.set_verbosity(0)  # silences warning re: lack of an index for the vcf file
-    vcf = pysam.VariantFile(filename=vcf_path.absolute().as_uri(), mode="r")
+
+    try:
+        vcf = pysam.VariantFile(filename=vcf_path.absolute().as_uri(), mode="r")
+    except ValueError as e:
+        error_message = (
+            "VCF ingestion failed: Not a valid VCF file (missing format declaration)"
+        )
+        _logger.exception(msg=error_message)
+        raise ValueError(error_message) from e
+
+    _validate_vcf_header(vcf)
 
     for batch in _yield_expression_af_batches(vcf):
         expressions, afs = zip(*batch, strict=True)
         variant_ids = av.put_allele_expressions(expressions, assembly)
 
-        cafs = []
+        cafs: list[AnyVlmCohortAlleleFrequencyResult] = []
         for variant_id, af in zip(variant_ids, afs, strict=True):
             if variant_id is None:
                 continue