Genomics

Endpoints de anotación de variantes y consulta de stores y perfiles.

POST /genomics/annotate

Recibe una lista de VariantAnnotation serializados, los pasa por el pipeline de anotación seleccionado y devuelve las variantes anotadas.

Request body:

{
  "variants": [ { "...VariantAnnotation serializado..." } ],
  "profile":  "clinical",
  "add":      ["cadd"],
  "exclude":  [],
  "workers":  4
}

Response:

{
  "variants":       [ { "...VariantAnnotation anotado..." } ],
  "total":          1234,
  "stores_applied": ["clinvar", "mcps_af", "loftee"]
}

Ejemplo completo

import httpx
from iris.adapters.driven.serializers.variant import converter
from iris.domain.genomics.entities import VariantAnnotation

variants: list[VariantAnnotation] = [...]

response = httpx.post(
    "http://localhost:8000/genomics/annotate",
    json={
        "variants": [converter.unstructure(va) for va in variants],
        "profile": "clinical",
        "workers": 8,
    },
)
result = response.json()
annotated = [converter.structure(v, VariantAnnotation) for v in result["variants"]]

---

GET /genomics/stores

Devuelve todos los stores registrados con su estado de disponibilidad.

Response:

[
  { "name": "clinvar",  "kind": "annotator+region", "available": true  },
  { "name": "mcps_af",  "kind": "annotator+region", "available": true  },
  { "name": "loftee",   "kind": "annotator",         "available": false },
  { "name": "gencode",  "kind": "gene",              "available": true  }
]

---

GET /genomics/profiles

Devuelve los perfiles definidos en profiles.toml.

Response:

[
  { "name": "clinical",  "stores": ["clinvar", "mcps_af", "loftee"] },
  { "name": "research",  "stores": ["clinvar", "mcps_af", "cadd", "dbnsfp", "loftee"] },
  { "name": "minimal",   "stores": ["clinvar", "mcps_af"] },
  { "name": "gene_view", "stores": ["clinvar", "loftee"] }
]

---

Referencia

iris.applications.api.routers.genomics

Genomics API router — variant annotation pipeline.

annotate async

annotate(body)

Annotate a list of VariantAnnotation objects through a configurable store pipeline.

The pipeline is resolved as: profile stores (in profile order) + add stores − exclude stores

PARAMETER	DESCRIPTION
body	JSON object with keys: - variants: list of serialized VariantAnnotation dicts - profile: named profile (clinical, research, minimal) - add: extra stores appended after the profile - exclude: stores removed from the final pipeline - workers: parallel tabix workers (default 4) TYPE: dict[str, Any]

Source code in src/iris/applications/api/routers/genomics.py

@router.post("/annotate", response_class=ORJSONResponse)
async def annotate(
    body: Annotated[
        dict[str, Any],
        Body(
            openapi_extra={
                "example": {
                    "variants": [
                        {"variant": {"nomenclature": {"ids": {"names": ["1:925952:G:A"]}}}}
                    ],  # noqa: E501
                    "profile": "clinical",
                    "add": [],
                    "exclude": [],
                    "workers": 4,
                }
            }
        ),
    ],
) -> ORJSONResponse:
    """Annotate a list of VariantAnnotation objects through a configurable store pipeline.

    The pipeline is resolved as:
    ``profile stores (in profile order) + add stores − exclude stores``

    Args:
        body: JSON object with keys:
            - ``variants``: list of serialized VariantAnnotation dicts
            - ``profile``: named profile (``clinical``, ``research``, ``minimal``)
            - ``add``: extra stores appended after the profile
            - ``exclude``: stores removed from the final pipeline
            - ``workers``: parallel tabix workers (default 4)
    """
    variants_raw: list[dict] = body.get("variants", [])
    profile: str | None = body.get("profile")
    add: list[str] = body.get("add", [])
    exclude: list[str] = body.get("exclude", [])
    workers: int = int(body.get("workers", 4))

    variants: list[VariantAnnotation] = [
        converter.structure(v, VariantAnnotation) for v in variants_raw
    ]

    store_names = resolve_stores(profile, add, exclude, get_profiles())
    for name in store_names:
        annotator = require_annotator(name)
        variants = annotator.annotate(variants, workers=workers)

    return ORJSONResponse(
        content={
            "variants": [converter.unstructure(va) for va in variants],
            "total": len(variants),
            "stores_applied": store_names,
        }
    )

stores async

stores()

Return all registered stores with their availability status.

Source code in src/iris/applications/api/routers/genomics.py

@router.get("/stores", response_class=ORJSONResponse)
async def stores() -> ORJSONResponse:
    """Return all registered stores with their availability status."""
    return ORJSONResponse(
        content=[
            {"name": s.name, "kind": s.kind, "available": s.available} for s in list_store_info()
        ]
    )

profiles async

profiles()

Return all annotation profiles with their store lists.

Source code in src/iris/applications/api/routers/genomics.py

@router.get("/profiles", response_class=ORJSONResponse)
async def profiles() -> ORJSONResponse:
    """Return all annotation profiles with their store lists."""
    return ORJSONResponse(
        content=[
            {"name": name, "stores": store_list} for name, store_list in get_profiles().items()
        ]
    )

iris.applications.api.stores

Store registries and profile resolution for the iris API.

Three registries

AnnotatorRegistry name → VariantAnnotator (annotate list[VariantAnnotation]) RegionRegistry name → McpsAFAnnotator (query_region → list[VariantAnnotation]) GeneRegistry name → GencodeGeneRepository (get_annotation → GeneAnnotation)

All registries are built lazily at first access and cached for the lifetime of the process. A store whose configured path is empty or missing is registered as available=False and raises HTTPException(503) when called.

StoreInfo dataclass

StoreInfo(name, kind, available, description='')

Metadata about a registered store.

resolve_stores

resolve_stores(profile, add, exclude, profiles)

Return the ordered list of store names for a request.

Final order: profile stores (profile order) + add stores, minus any names in exclude.

Source code in src/iris/applications/api/stores.py

def resolve_stores(
    profile: str | None, add: list[str], exclude: list[str], profiles: dict[str, list[str]]
) -> list[str]:
    """Return the ordered list of store names for a request.

    Final order: profile stores (profile order) + add stores,
    minus any names in exclude.
    """
    base: list[str] = []
    if profile is not None:
        if profile not in profiles:
            raise HTTPException(
                status_code=422,
                detail=f"Unknown profile {profile!r}. Available: {sorted(profiles)}",
            )
        base = list(profiles[profile])

    exclude_set = set(exclude)
    result: list[str] = [s for s in base if s not in exclude_set]
    for s in add:
        if s not in exclude_set and s not in result:
            result.append(s)
    return result

get_annotator_registry cached

get_annotator_registry()

Return the cached annotator registry (built once at first call).

Source code in src/iris/applications/api/stores.py

@lru_cache(maxsize=1)
def get_annotator_registry() -> dict[str, Any]:
    """Return the cached annotator registry (built once at first call)."""
    return _build_annotator_registry(get_settings())

get_region_registry cached

get_region_registry()

Return the cached region-query registry.

Source code in src/iris/applications/api/stores.py

@lru_cache(maxsize=1)
def get_region_registry() -> dict[str, Any]:
    """Return the cached region-query registry."""
    return _build_region_registry(get_settings())

get_gene_registry cached

get_gene_registry()

Return the cached gene annotation registry.

Source code in src/iris/applications/api/stores.py

@lru_cache(maxsize=1)
def get_gene_registry() -> dict[str, Any]:
    """Return the cached gene annotation registry."""
    return _build_gene_registry(get_settings())

get_profiles cached

get_profiles()

Return the cached profile → store-list mapping.

Source code in src/iris/applications/api/stores.py

@lru_cache(maxsize=1)
def get_profiles() -> dict[str, list[str]]:
    """Return the cached profile → store-list mapping."""
    return _load_profiles()

require_annotator

require_annotator(name)

Return the annotator for name, or raise 503/422.

Source code in src/iris/applications/api/stores.py

def require_annotator(name: str) -> Any:
    """Return the annotator for *name*, or raise 503/422."""
    reg = get_annotator_registry()
    if name not in reg:
        raise HTTPException(status_code=422, detail=f"Unknown store {name!r}")
    inst = reg[name]
    if inst is None:
        raise HTTPException(
            status_code=503,
            detail=f"Store {name!r} is not configured (check IRIS_{name.upper()}_* env vars)",
        )
    return inst

require_region_store

require_region_store(name)

Return the region store for name, or raise 503/422.

Source code in src/iris/applications/api/stores.py

def require_region_store(name: str) -> Any:
    """Return the region store for *name*, or raise 503/422."""
    reg = get_region_registry()
    if name not in reg:
        raise HTTPException(
            status_code=422, detail=f"Unknown region store {name!r}. Available: {sorted(reg)}"
        )
    inst = reg[name]
    if inst is None:
        raise HTTPException(status_code=503, detail=f"Region store {name!r} is not configured")
    return inst

require_gene_source

require_gene_source(name='gencode')

Return the gene source for name, or raise 503/422.

Source code in src/iris/applications/api/stores.py

def require_gene_source(name: str = "gencode") -> Any:
    """Return the gene source for *name*, or raise 503/422."""
    reg = get_gene_registry()
    if name not in reg:
        raise HTTPException(
            status_code=422, detail=f"Unknown gene source {name!r}. Available: {sorted(reg)}"
        )
    inst = reg[name]
    if inst is None:
        raise HTTPException(
            status_code=503,
            detail=f"Gene source {name!r} is not configured (set IRIS_GENCODE_GTF)",
        )
    return inst

list_store_info

list_store_info()

Return metadata for all registered stores across all registries.

Source code in src/iris/applications/api/stores.py

def list_store_info() -> list[StoreInfo]:
    """Return metadata for all registered stores across all registries."""
    infos: list[StoreInfo] = []
    for name, inst in get_annotator_registry().items():
        infos.append(StoreInfo(name=name, kind="annotator", available=inst is not None))
    for name, inst in get_region_registry().items():
        if not any(i.name == name and i.kind == "annotator" for i in infos):
            infos.append(StoreInfo(name=name, kind="region", available=inst is not None))
        else:
            # mcps_af is both annotator and region — mark it
            for i in infos:
                if i.name == name:
                    i.kind = "annotator+region"
    for name, inst in get_gene_registry().items():
        infos.append(StoreInfo(name=name, kind="gene", available=inst is not None))
    return infos