feat: build vlm response (issue #13)

src/anyvlm/config.py

@@ -15,7 +15,7 @@ class Settings(BaseSettings):
     """
 
     model_config = SettingsConfigDict(
-        env_prefix="vlm_",
+        env_prefix="anyvlm_",
         env_file=".env",
         env_file_encoding="utf-8",
         extra="ignore",

src/anyvlm/functions/build_vlm_response.py

@@ -0,0 +1,18 @@
+"""Craft a VlmResponse object from a list of CohortAlleleFrequencyStudyResults"""
+
+from ga4gh.va_spec.base.core import CohortAlleleFrequencyStudyResult
+
+from anyvlm.schemas.vlm import (
+    VlmResponse,
+)
+
+
+def build_vlm_response_from_caf_data(
+    caf_data: list[CohortAlleleFrequencyStudyResult],
+) -> VlmResponse:
+    """Craft a VlmResponse object from a list of CohortAlleleFrequencyStudyResults.
+
+    :param caf_data: A list of `CohortAlleleFrequencyStudyResult` objects that will be used to build the VlmResponse
+    :return: A `VlmResponse` object.
+    """
+    raise NotImplementedError  # TODO: Implement this during/after Issue #16

src/anyvlm/functions/get_caf.py

@@ -3,17 +3,30 @@
 from ga4gh.va_spec.base.core import CohortAlleleFrequencyStudyResult
 
 from anyvlm.anyvar.base_client import BaseAnyVarClient
+from anyvlm.utils.types import (
+    ChromosomeName,
+    GenomicSequence,
+    GrcAssemblyId,
+    UscsAssemblyBuild,
+)
 
 
 def get_caf(
-    av: BaseAnyVarClient, accession_id: str, start: int, end: int
+    anyvar_client: BaseAnyVarClient,
+    assembly_id: GrcAssemblyId | UscsAssemblyBuild,
+    reference_name: ChromosomeName,
+    start: int,
+    reference_bases: GenomicSequence,
+    alternate_bases: GenomicSequence,
 ) -> list[CohortAlleleFrequencyStudyResult]:
     """Retrieve Cohort Allele Frequency data for all known variants matching provided search params
 
-    :param av: AnyVar client
-    :param accession_id: ID for sequence to search upon
+    :param anyvar_client: AnyVar client
+    :param assembly_id: The reference assembly to utilize - must be one of: "GRCh37", "GRCh38", "hg38", "hg19"
+    :param reference_name: The chromosome to search on, with an optional "chr" prefix - e.g., "1", "chr22", "X", "chrY", etc.
     :param start: start of range search
-    :param end: end of range to search
+    :param reference_bases: Genomic bases ('T', 'AC', etc.)
+    :param alternate_bases: Genomic bases ('T', 'AC', etc.)
     :return: list of CAFs contained in search interval
     """
-    raise NotImplementedError
+    raise NotImplementedError  # TODO: Implement this. See Issue #16.

src/anyvlm/main.py

@@ -2,7 +2,6 @@
 
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from enum import Enum
 
 from fastapi import FastAPI
 
@@ -15,6 +14,9 @@
     ServiceOrganization,
     ServiceType,
 )
+from anyvlm.utils.types import (
+    EndpointTag,
+)
 
 
 def create_anyvar_client(
@@ -58,18 +60,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator:
 )
 
 
-class _Tag(str, Enum):
-    """Define tag names for endpoints."""
-
-    META = "Meta"
-    SEARCH = "Search"
-
-
 @app.get(
     "/service-info",
     summary="Get basic service information",
     description="Retrieve service metadata, such as versioning and contact info. Structured in conformance with the [GA4GH service info API specification](https://www.ga4gh.org/product/service-info/)",
-    tags=[_Tag.META],
+    tags=[EndpointTag.META],
 )
 def service_info() -> ServiceInfo:
     """Provide service info per GA4GH Service Info spec"""

src/anyvlm/restapi/vlm.py

@@ -1,6 +1,25 @@
 """Define route(s) for the variant-level matching (VLM) protocol"""
 
 from pathlib import Path
+from typing import Annotated
+
+from fastapi import Query, Request
+from ga4gh.va_spec.base.core import CohortAlleleFrequencyStudyResult
+
+from anyvlm.anyvar.base_client import BaseAnyVarClient
+from anyvlm.functions.build_vlm_response import build_vlm_response_from_caf_data
+from anyvlm.functions.get_caf import get_caf
+from anyvlm.main import app
+from anyvlm.schemas.vlm import (
+    VlmResponse,
+)
+from anyvlm.utils.types import (
+    ChromosomeName,
+    EndpointTag,
+    GenomicSequence,
+    GrcAssemblyId,
+    UscsAssemblyBuild,
+)
 
 
 def ingest_vcf(vcf_path: Path) -> None:
@@ -9,3 +28,43 @@ def ingest_vcf(vcf_path: Path) -> None:
     :param vcf_path: VCF file location
     """
     raise NotImplementedError
+
+
+@app.get(
+    "/variant_counts",
+    summary="Provides allele counts of a single sequence variant, broken down by zygosity",
+    description="Search for a single sequence variant and receive allele counts by zygosity, in accordance with the Variant-Level Matching protocol",
+    tags=[EndpointTag.SEARCH],
+)
+def variant_counts(
+    request: Request,
+    assemblyId: Annotated[  # noqa: N803
+        GrcAssemblyId | UscsAssemblyBuild,
+        Query(..., description="Genome reference assembly"),
+    ],
+    referenceName: Annotated[  # noqa: N803
+        ChromosomeName, Query(..., description="Chromosome with optional 'chr' prefix")
+    ],
+    start: Annotated[int, Query(..., description="Variant position")],
+    referenceBases: Annotated[  # noqa: N803
+        GenomicSequence, Query(..., description="Genomic bases ('T', 'AC', etc.)")
+    ],
+    alternateBases: Annotated[  # noqa: N803
+        GenomicSequence, Query(..., description="Genomic bases ('T', 'AC', etc.)")
+    ],
+) -> VlmResponse:
+    """Accept a Variant-Level Matching network request and return allele counts by zygosity.
+
+    :param request: FastAPI `Request` object
+    :param assemblyId: The genome reference assembly. Must be a GRC assembly identifier (e.g., "GRCh38) or a USCS assembly build (e.g., "hg38")
+    :param referenceName: The name of the reference chromosome, with optional 'chr' prefix
+    :param start: The start of the variant's position
+    :param referenceBases: Genomic bases ('T', 'AC', etc.)
+    :param alternateBases: Genomic bases ('T', 'AC', etc.)
+    :return: A VlmResponse object containing cohort allele frequency data. If no matches are found, endpoint will return a status code of 200 with an empty set of results.
+    """
+    anyvar_client: BaseAnyVarClient = request.app.state.anyvar_client
+    caf_data: list[CohortAlleleFrequencyStudyResult] = get_caf(
+        anyvar_client, assemblyId, referenceName, start, referenceBases, alternateBases
+    )
+    return build_vlm_response_from_caf_data(caf_data)

src/anyvlm/schemas/vlm.py

@@ -1 +1,154 @@
 """Schemas relating to VLM API."""
+
+from typing import ClassVar, Literal, Self
+
+from pydantic import BaseModel, Field, model_validator
+
+from anyvlm.utils.types import Zygosity
+
+# ruff: noqa: N815 (allows camelCase vars instead of snake_case to align with expected VLM protocol response)
+
+RESULT_ENTITY_TYPE = "genomicVariant"
+
+
+class HandoverType(BaseModel):
+    """The type of handover the parent `BeaconHandover` represents."""
+
+    id: str = Field(
+        default="gregor", description="Node-specific identifier"
+    )  # TODO: enable configuration of this field. See Issue #27.
+    label: str = Field(
+        default="GREGoR AnVIL browser", description="Node-specific label"
+    )  # TODO: enable configuration of this field. See Issue #27.
+
+
+class BeaconHandover(BaseModel):
+    """Describes how users can get more information about the results provided in the parent `VlmResponse`"""
+
+    handoverType: HandoverType = HandoverType()
+    url: str = Field(
+        default="https://anvil.terra.bio/#workspaces?filter=GREGoR",  # TODO: enable configuration of this field. See Issue #27.
+        description="A url which directs users to more detailed information about the results tabulated by the API (ideally human-readable)",
+    )
+
+
+class ReturnedSchema(BaseModel):
+    """Fixed [Beacon Schema](https://github.com/ga4gh-beacon/beacon-v2/blob/c6558bf2e6494df3905f7b2df66e903dfe509500/framework/json/common/beaconCommonComponents.json#L241)"""
+
+    entityType: str = Field(
+        default=RESULT_ENTITY_TYPE,
+        description=f"The type of entity this response describes. Must always be set to '{RESULT_ENTITY_TYPE}'",
+    )
+    schema_: str = Field(
+        default="ga4gh-beacon-variant-v2.0.0",
+        # Alias is required because 'schema' is reserved by Pydantic's BaseModel class,
+        # But VLM expects a field named 'schema'
+        alias="schema",
+    )
+
+    model_config = {"populate_by_name": True}
+
+
+class Meta(BaseModel):
+    """Relevant metadata about the results provided in the parent `VlmResponse`"""
+
+    apiVersion: str = Field(
+        default="v1.0",
+        description="The version of the VLM API that this response conforms to",
+    )
+    beaconId: str = Field(
+        default="org.gregor.beacon",  # TODO: enable configuration of this field. See Issue #27.
+        description="""
+            The Id of a Beacon. Usually a reversed domain string, but any URI is acceptable. The purpose of this attribute is,
+            in the context of a Beacon network, to disambiguate responses coming from different Beacons. See the beacon documentation
+            [here](https://github.com/ga4gh-beacon/beacon-v2/blob/c6558bf2e6494df3905f7b2df66e903dfe509500/framework/src/common/beaconCommonComponents.yaml#L26)
+        """,
+    )
+    returnedSchemas: list[ReturnedSchema] = [ReturnedSchema()]
+
+
+class ResponseSummary(BaseModel):
+    """A high-level summary of the results provided in the parent `VlmResponse"""
+
+    exists: bool = Field(
+        ..., description="Indicates whether the response contains any results."
+    )
+    numTotalResults: int = Field(
+        ..., description="The total number of results found for the given query"
+    )
+
+
+class ResultSet(BaseModel):
+    """A set of cohort allele frequency results. The zygosity of the ResultSet is identified in the `id` field"""
+
+    exists: Literal[True] = Field(
+        default=True,
+        description="Indicates whether this ResultSet exists. This must always be `True`, even if `resultsCount` = `0`",
+    )
+    id: str = Field(
+        ...,
+        description="id should be constructed of the `HandoverType.id` + the ResultSet's zygosity. See `validate_resultset_ids` validator in `VlmResponse` class.",
+        examples=["Geno2MP Homozygous", "MyGene2 Heterozygous"],
+    )
+    results: list = Field(
+        default=[],
+        min_length=0,
+        max_length=0,
+        description="This must always be set to an empty array",
+    )
+    resultsCount: int = Field(
+        ..., description="A count for the zygosity indicated by the ResultSet's `id`"
+    )
+    setType: str = Field(
+        default=RESULT_ENTITY_TYPE,
+        description=f"The type of entity relevant to these results. Must always be set to '{RESULT_ENTITY_TYPE}'",
+    )
+
+
+class ResponseField(BaseModel):
+    """A list of ResultSets"""
+
+    resultSets: list[ResultSet] = Field(
+        ..., description="A list of ResultSets for the given query."
+    )
+
+
+class VlmResponse(BaseModel):
+    """Define response structure for the variant_counts endpoint."""
+
+    beaconHandovers: list[BeaconHandover] = [BeaconHandover()]
+    meta: Meta = Meta()
+    responseSummary: ResponseSummary
+    response: ResponseField
+
+    resultset_id_error_message_base: ClassVar[str] = (
+        "Invalid ResultSet id - ids must be in form '<node_id> <zygosity>'"
+    )
+
+    @model_validator(mode="after")
+    def validate_resultset_ids(self) -> Self:
+        """Ensure each ResultSet.id is correctly constructed."""
+        handover_ids: list[str] = [
+            beaconHandover.handoverType.id for beaconHandover in self.beaconHandovers
+        ]
+
+        for result_set in self.response.resultSets:
+            node_id, zygosity = None, None
+            try:
+                node_id, zygosity = result_set.id.split(" ")
+            except ValueError as e:
+                error_message = f"{self.resultset_id_error_message_base}, but provided id of {result_set.id} contains invalid formatting"
+                raise ValueError(error_message) from e
+
+            if node_id not in handover_ids:
+                error_message = f"{self.resultset_id_error_message_base}, but provided node_id of {node_id} does not match any `handoverType.id` provided in `self.beaconHandovers`"
+                raise ValueError(error_message)
+
+            try:
+                Zygosity(zygosity)
+            except ValueError as e:
+                valid_zygosity_values = {zygosity.value for zygosity in Zygosity}
+                error_message = f"{self.resultset_id_error_message_base}, but provided zygosity of {zygosity} is not found in allowable value set of: {', '.join(valid_zygosity_values)}"
+                raise ValueError(error_message) from e
+
+        return self

src/anyvlm/utils/__init__.py

@@ -0,0 +1 @@
+"""Provide utilities."""