Stores & Pipeline¶

Adapters for cohort-scale variant extraction and parallel annotation pipelines.

ZarrVariantStore¶

Cohort store backed by a vcf2zarr Zarr v3 archive. Implements the VariantAnnotationStore port.

Key design points¶

variant_position values are 1-based (VCF native, as written by vcf2zarr without --zero-based).
open() loads the full variant_contig array once and builds per-contig start/end offsets via numpy.searchsorted, then discards the array.
Per-chromosome position vectors are pre-loaded into _chrom_pos_cache for O(1) mask computation inside fetch().
fetch() is thread-safe: it performs only read operations on pre-loaded numpy arrays and issues independent fancy-index reads against the Zarr store (Zarr's LocalStore supports concurrent reads).

Usage¶

from iris.adapters.driven.stores import ZarrVariantStore

with ZarrVariantStore(path="/data/wes_cohort.zarr") as store:
    variants = store.fetch(region)

Genotype processors¶

ZarrVariantStore accepts an optional processor (a GenotypeProcessor implementation) and processor_context dict. When set, the processor is called once per fetch block per ALT allele with the raw genotype arrays (n_variants, n_samples, ploidy), and its returned dict is merged into each VariantAnnotation.metadata.data. Built-in genotype counts (n_hom, n_het, …) are always computed regardless of whether a processor is supplied. See Genomic analysis for concrete implementations (AncestryStratifiedAFProcessor, PerIndividualAltCountProcessor).

iris.adapters.driven.stores.zarr ¶

ZarrVariantStore — vcf2zarr cohort store adapter.

Supports user-defined GenotypeProcessor callables injected at construction time, enabling cohort-level statistics, stratified allele frequencies, and regression models without subclassing.

ZarrVariantStore ¶

Bases: VariantAnnotationStore

Cohort-scale variant store backed by a vcf2zarr Zarr v3 archive.

Designed for variant_position values that are 1-based (VCF native, as written by vcf2zarr without the --zero-based flag).

The store must be opened before use — either explicitly via :meth:open or through the context-manager protocol::

with ZarrVariantStore(path=ZARR_PATH) as store:
    variants = store.fetch(region)

After open():

Contig offsets are built once from the full variant_contig array using numpy.searchsorted.
Per-chromosome position arrays are pre-loaded and cached for O(1) mask computation in :meth:fetch.

:meth:fetch is thread-safe: it performs only read operations on the pre-loaded numpy arrays and issues independent fancy-index reads against the underlying Zarr store (Zarr's LocalStore supports concurrent reads).

ATTRIBUTE	DESCRIPTION
`path`	Path to the Zarr store directory or `.zarr` file. TYPE: `Path`
`assembly`	Genome assembly label stored in created :class:`~iris.domain.genomics.objects.GenomicPosition` objects. TYPE: `str`
`source`	Population / source label used in the generated :class:`~iris.domain.genomics.objects.AlleleFrequency` objects. TYPE: `str`

processor `class-attribute` `instance-attribute` ¶

processor = None

Optional user-defined processor called for each variant block.

When set, the processor receives the raw genotype arrays (n_variants, n_samples, ploidy) and returns a dict of computed values that are merged into VariantAnnotation.metadata.data. The built-in genotype counts (n_hom, n_het, etc.) are always computed regardless of whether a processor is supplied.

See :class:~iris.domain.genomics.ports.driven.genotype_processor.GenotypeProcessor.

processor_context `class-attribute` `instance-attribute` ¶

processor_context = field(factory=dict)

Arbitrary context dict forwarded to the processor on every call.

Use this to pass sample covariates, phenotypes, or configuration that the processor needs but that are not part of the genotype arrays.

sample_ids `property` ¶

sample_ids

Sample IDs in the same order as the store's sample dimension.

Empty until :meth:open has been called. Used to align external per-sample covariates (ancestry, phenotype, ...) to the gt/ gt_mask arrays a :class:GenotypeProcessor receives — the processor itself is never given this ordering directly, so callers that need it (e.g. building a valid_mask) read it from here before constructing the processor.

open ¶

open()

Open the Zarr store and build in-memory indices.

Loads the full variant_contig array (~35 MB for a 79k cohort) once to build per-contig start / end offsets, then pre-loads each per-chromosome position array into _chrom_pos_cache. The large variant_contig array is discarded after the index is built.

RETURNS	DESCRIPTION
`ZarrVariantStore`	Self, so this can be used fluently: `store.open()`.

RAISES	DESCRIPTION
`ImportError`	If `zarr` or `numpy` are not installed.

Source code in src/iris/adapters/driven/stores/zarr.py

def open(self) -> ZarrVariantStore:
    """Open the Zarr store and build in-memory indices.

    Loads the full ``variant_contig`` array (~35 MB for a 79k cohort)
    once to build per-contig start / end offsets, then pre-loads each
    per-chromosome position array into ``_chrom_pos_cache``.  The large
    ``variant_contig`` array is discarded after the index is built.

    Returns:
        Self, so this can be used fluently: ``store.open()``.

    Raises:
        ImportError: If ``zarr`` or ``numpy`` are not installed.
    """
    try:
        import numpy as np  # pyright: ignore[reportMissingImports]
        import zarr  # pyright: ignore[reportMissingImports]
    except ImportError as exc:
        raise ImportError(
            "zarr and numpy are required: uv pip install 'iris[genomics]'"
        ) from exc

    self._store = zarr.open(str(self.path), mode="r")
    contig_id = self._store["contig_id"][:]
    self._contig_names = [
        HUMAN_CHROMOSOMES.resolve(str(c)) for c in contig_id if str(c) in HUMAN_CHROMOSOMES
    ]
    self._contig_name_to_idx = {c: i for i, c in enumerate(self._contig_names)}

    # Build start/end offsets for each contig index using searchsorted.
    v_contig_all = self._store["variant_contig"][:]
    n_contigs = len(self._contig_names)
    idx = np.arange(n_contigs)
    self._contig_start = np.searchsorted(v_contig_all, idx, side="left").astype(int)
    self._contig_end = np.searchsorted(v_contig_all, idx, side="right").astype(int)
    del v_contig_all

    # Pre-load per-chromosome position vectors.
    for chrom, ci in self._contig_name_to_idx.items():
        s, e = int(self._contig_start[ci]), int(self._contig_end[ci])
        if s >= e:
            continue
        pos = self._store["variant_position"][s:e]
        self._chrom_pos_cache[chrom] = (pos, s)

    self._sample_ids = list(self._store["sample_id"][:]) if "sample_id" in self._store else []
    self._n_samples = len(self._sample_ids)
    return self

close ¶

close()

Release in-memory caches and close the Zarr store handle.

Source code in src/iris/adapters/driven/stores/zarr.py

def close(self) -> None:
    """Release in-memory caches and close the Zarr store handle."""
    self._store = None
    self._chrom_pos_cache = {}
    self._sample_ids = []
    self._contig_start = None
    self._contig_end = None

fetch ¶

fetch(region)

Return all variants whose positions fall within region.

region.start and region.end must both be non-None and are interpreted as 1-based inclusive coordinates matching the variant_position encoding in the store.

The method opens the store lazily if :meth:open has not been called. It is safe to call concurrently from multiple threads.

If you only need self.processor's side effects (e.g. AltCountProcessor's per-sample accumulator, read back via summary_stats()) and never look at the returned VariantAnnotation objects, use :meth:scan instead — building this list is most of fetch()'s cost with a wide multi-stratum processor, and :meth:scan skips it entirely.

PARAMETER	DESCRIPTION
`region`	1-based inclusive genomic window. TYPE: `GenomicPosition`

RETURNS	DESCRIPTION
`list[VariantAnnotation]`	Possibly-empty list of
`list[VariantAnnotation]`	class:`~iris.domain.genomics.entities.VariantAnnotation`.

RAISES	DESCRIPTION
`ValueError`	If `region.start` or `region.end` is `None`.
`ImportError`	If `zarr` or `numpy` are not installed.

Source code in src/iris/adapters/driven/stores/zarr.py

def fetch(self, region: GenomicPosition) -> list[VariantAnnotation]:
    """Return all variants whose positions fall within *region*.

    ``region.start`` and ``region.end`` must both be non-``None`` and are
    interpreted as **1-based inclusive** coordinates matching the
    ``variant_position`` encoding in the store.

    The method opens the store lazily if :meth:`open` has not been called.
    It is safe to call concurrently from multiple threads.

    If you only need ``self.processor``'s side effects (e.g.
    ``AltCountProcessor``'s per-sample accumulator, read back via
    ``summary_stats()``) and never look at the returned
    ``VariantAnnotation`` objects, use :meth:`scan` instead — building
    this list is most of ``fetch()``'s cost with a wide multi-stratum
    processor, and :meth:`scan` skips it entirely.

    Args:
        region: 1-based inclusive genomic window.

    Returns:
        Possibly-empty list of
        :class:`~iris.domain.genomics.entities.VariantAnnotation`.

    Raises:
        ValueError: If ``region.start`` or ``region.end`` is ``None``.
        ImportError: If ``zarr`` or ``numpy`` are not installed.
    """
    block = self._load_block(region)
    if block is None:
        return []
    chrom, positions, v_allele, v_id, gt, gt_mask, local_indices, processor_block_results = (
        block
    )

    results: list[VariantAnnotation] = []

    for local_i in range(len(local_indices)):
        alleles_row = v_allele[local_i]
        ref = str(alleles_row[0])
        alts = [str(a) for a in alleles_row[1:] if str(a) not in ("", ".")]
        if not alts:
            continue

        pos = int(positions[local_i])  # already 1-based

        vid_raw = str(v_id[local_i]) if v_id is not None else None
        if vid_raw in ("", "."):
            vid_raw = None

        # Slice genotype arrays for this one variant.
        gt_v = gt[local_i : local_i + 1]  # shape (1, n_samples, 2)
        gt_mask_v = gt_mask[local_i : local_i + 1] if gt_mask is not None else None

        for alt_idx, alt in enumerate(alts, start=1):
            n_hom_ref_arr, n_het_arr, n_hom_alt_arr, n_missing_arr = (
                _genotype_counts_vectorized(gt_v, gt_mask_v, alt_idx)
            )
            gc = GenotypeCount(
                n_hom_ref=int(n_hom_ref_arr[0]),
                n_het=int(n_het_arr[0]),
                n_hom_alt=int(n_hom_alt_arr[0]),
                n_missing=int(n_missing_arr[0]),
                allele=alt,
            )
            af = gc.to_allele_frequency(self.source)

            vid = vid_raw or f"{chrom}-{pos}-{ref}-{alt}"
            dbsnp_id = vid_raw if vid_raw and vid_raw.startswith("rs") else None

            # Base metadata — always present
            meta: dict[str, Any] = {
                "n_hom": int(n_hom_alt_arr[0]),
                "n_het": int(n_het_arr[0]),
                "n_hom_ref": int(n_hom_ref_arr[0]),
                "n_missing": int(n_missing_arr[0]),
                "n_samples": self._n_samples,
                "ac": gc.ac,
                "an": gc.an,
                "af": (gc.ac / gc.an) if gc.an else None,
            }

            # Merge processor results for this variant and alt index
            if alt_idx in processor_block_results:
                for field, arr in processor_block_results[alt_idx].items():
                    if hasattr(arr, "__len__") and hasattr(arr, "__getitem__"):
                        # Array — extract the value for this variant
                        val = arr[local_i]
                        meta[field] = val.tolist() if hasattr(val, "tolist") else float(val)
                    else:
                        # Scalar — apply to all variants
                        meta[field] = arr

            results.append(
                VariantAnnotation(
                    variant=Variant(
                        variant_type=_infer_type(ref, alt),
                        nomenclature=VariantNomenclature.from_key(
                            (chrom, pos, ref, alt),
                            dbsnp_id=dbsnp_id,
                            extra_ids=[vid] if vid else None,
                        ),
                        alleles=Allele(reference=ref, alternate=(alt,)),
                        position=GenomicPosition(
                            chromosome=chrom,
                            assembly=self.assembly,
                            start=pos,
                            end=pos + len(ref) - 1,
                        ),
                    ),
                    allele_frequencies=(af,),
                    metadata=Metadata(source=self.source, data=meta),
                )
            )

    return results

scan ¶

scan(region)

Run self.processor over region without materializing VariantAnnotation.

Use this instead of :meth:fetch when you only need the processor's side effects — e.g. AltCountProcessor accumulates its per-sample counters (read back via summary_stats() once every region has been scanned) regardless of what a caller does with its per-block return value. Building the Variant/Allele/GenomicPosition/ Metadata object graph in :meth:fetch is most of its cost when the processor is configured with many strata; :meth:scan skips it entirely while still calling self.processor with the exact same arrays, in the exact same order, as :meth:fetch would.

It is safe to call concurrently from multiple threads, same as :meth:fetch.

PARAMETER	DESCRIPTION
`region`	1-based inclusive genomic window. TYPE: `GenomicPosition`

RETURNS	DESCRIPTION
`int`	Number of (variant, ALT allele) pairs found in region — useful
`int`	for progress logging, not otherwise meaningful.

RAISES	DESCRIPTION
`ValueError`	If `region.start` or `region.end` is `None`.
`ImportError`	If `zarr` or `numpy` are not installed.

Source code in src/iris/adapters/driven/stores/zarr.py

def scan(self, region: GenomicPosition) -> int:
    """Run ``self.processor`` over *region* without materializing ``VariantAnnotation``.

    Use this instead of :meth:`fetch` when you only need the processor's
    side effects — e.g. ``AltCountProcessor`` accumulates its per-sample
    counters (read back via ``summary_stats()`` once every region has
    been scanned) regardless of what a caller does with its per-block
    return value. Building the ``Variant``/``Allele``/``GenomicPosition``/
    ``Metadata`` object graph in :meth:`fetch` is most of its cost when
    the processor is configured with many strata; :meth:`scan` skips it
    entirely while still calling ``self.processor`` with the exact same
    arrays, in the exact same order, as :meth:`fetch` would.

    It is safe to call concurrently from multiple threads, same as
    :meth:`fetch`.

    Args:
        region: 1-based inclusive genomic window.

    Returns:
        Number of (variant, ALT allele) pairs found in *region* — useful
        for progress logging, not otherwise meaningful.

    Raises:
        ValueError: If ``region.start`` or ``region.end`` is ``None``.
        ImportError: If ``zarr`` or ``numpy`` are not installed.
    """
    block = self._load_block(region)
    if block is None:
        return 0

    n_pairs = 0
    for local_i in range(len(block.local_indices)):
        alts = [a for a in block.v_allele[local_i][1:] if str(a) not in ("", ".")]
        n_pairs += len(alts)
    return n_pairs

UvarVariantStore¶

Region-scale store backed by the per-chromosome uvar Parquet database — an exhaustive precomputed annotation source (~8.9B rows, ~0.55 TB across GRCh38). Implements the VariantAnnotationStore port via fetch(), plus two lower-level bulk-query methods (query_region_columns, query_positions_columns) for callers that need more than the fixed default column set.

Full data model, category-by-category schema, and live query output: see Genomic Data → uvar.

Key design points¶

Both access modes are region-bounded — region.start/region.end must be set; there is no "fetch the whole file" mode given file sizes up to ~46 GB per chromosome.
Relies entirely on pyarrow.dataset predicate pushdown (row-group statistics pruning), not manual indexing — files are pre-sorted by position.
Column projection works at top-level struct granularity only — e.g. columns=["gnomad_genome"] returns the whole nested struct, not individual leaf fields.
require_population_signal=True pushes an "observed gnomAD frequency OR ClinVar entry" filter into the same scan — necessary because uvar is exhaustive (every theoretical SNV has a row) and a gene-sized region otherwise returns hundreds of thousands of never-observed substitutions.
aggregate="pos_bin_1m" groups by the schema's own 1 Mb bin column for genome/chromosome-scale summaries instead of row-level data.

Usage¶

from iris.adapters.driven.stores.uvar import UvarVariantStore, COLUMN_GROUPS

store = UvarVariantStore(data_dir="/mnt/cephfs/hot_nvme/uvar")
annotations = store.fetch(region)  # small (gene-sized) regions

cols = [*COLUMN_GROUPS["population_variation"], *COLUMN_GROUPS["clinical_significance"]]
table = store.query_region_columns(region, columns=cols, require_population_signal=True)

iris.adapters.driven.stores.uvar ¶

UvarVariantStore — Parquet-backed adapter for the uvar universal SNV database.

uvar is a per-chromosome Parquet database with exhaustive coverage of every possible SNV in GRCh38 (3 rows per position, one per alternate allele), 416 columns of precomputed annotation (population frequencies, ClinVar, in-silico predictors, conservation, regulatory scores, ...). Files are large (up to ~46 GB per chromosome) and sorted by position, so this adapter relies entirely on pyarrow.dataset predicate pushdown (row-group statistics pruning) rather than any manual indexing.

Two access modes, both region-bounded (region.start/region.end must both be set — there is no "fetch the whole file" mode given the file sizes):

:meth:fetch — full port conformance (VariantAnnotationStore), maps a small fixed default column set into :class:VariantAnnotation. Intended for small regions (gene-level); never requests all 416 columns.
:meth:query_region_columns — low-level bulk path for the region-features use cases. Column projection works at TOP-LEVEL struct/scalar granularity only (e.g. requesting "gnomad_genome" returns the whole nested struct; there is no per-leaf-field pushdown). aggregate="pos_bin_1m" groups by the schema's existing 1 Mb bin column for genome/chromosome-level summaries instead of returning row-level data.

COLUMN_GROUPS `module-attribute` ¶

COLUMN_GROUPS = {
    name: (load_str_tuple(_TOML, "column_groups", name))
    for name in (
        "population_variation",
        "clinical_significance",
        "insilico",
    )
}

Maps a dashboard "source" filter name to the top-level uvar columns it needs.

UvarVariantStore ¶

Bases: VariantAnnotationStore

Region-scale variant store backed by per-chromosome uvar Parquet files.

ATTRIBUTE	DESCRIPTION
`data_dir`	Directory containing `chr{1..22,X,Y}.p` Parquet files. TYPE: `Path`
`file_pattern`	Filename template, formatted with the canonical chromosome name (e.g. `"21"`, `"X"`) — note the files use a `chr` prefix (`chr21.p`) even though the in-file `chromosome` column values do not. TYPE: `str`
`assembly`	Genome assembly label stored in produced GenomicPosition objects. TYPE: `str`
`source`	Label stored on produced VariantAnnotation.metadata. TYPE: `str`

query_region_columns ¶

query_region_columns(
    region,
    *,
    columns,
    aggregate=None,
    require_population_signal=False,
)

Return uvar rows within region, projected to columns (plus position/chromosome).

PARAMETER	DESCRIPTION
`region`	1-based inclusive genomic window. Both start and end must be set — there is no unbounded query mode. TYPE: `GenomicPosition`
`columns`	Top-level uvar column names to project (see COLUMN_GROUPS for the dashboard-source groupings). Struct columns are returned whole (no per-leaf-field projection). TYPE: `Sequence[str]`
`aggregate`	When `"pos_bin_1m"`, groups rows by the schema's existing 1 Mb bin column and returns a per-bin variant count instead of row-level data — used for genome/chromosome-level zoom, where returning every row would be impractical. TYPE: `str \| None` DEFAULT: `None`
`require_population_signal`	uvar is exhaustive (every theoretical SNV, 3 rows/position) — for gene/cytoband-sized regions this is hundreds of thousands to millions of rows, almost all never-observed. When True, pushes down a row-group/predicate filter for an observed gnomAD frequency or a ClinVar classification (mirrors `use_cases.gosling_common._has_population_signal`, which otherwise has to apply this same gate in Python after the full exhaustive table is materialized). dbSNP membership is deliberately not used as a signal — see `_has_population_signal` for why. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Table`	A pyarrow.Table. Empty (zero columns) if the chromosome has no
`Table`	backing file.

RAISES	DESCRIPTION
`ValueError`	If region.start or region.end is None.

Source code in src/iris/adapters/driven/stores/uvar.py

def query_region_columns(
    self,
    region: GenomicPosition,
    *,
    columns: Sequence[str],
    aggregate: str | None = None,
    require_population_signal: bool = False,
) -> pa.Table:
    """Return uvar rows within region, projected to columns (plus position/chromosome).

    Args:
        region: 1-based inclusive genomic window. Both start and end
            must be set — there is no unbounded query mode.
        columns: Top-level uvar column names to project (see
            COLUMN_GROUPS for the dashboard-source groupings). Struct
            columns are returned whole (no per-leaf-field projection).
        aggregate: When ``"pos_bin_1m"``, groups rows by the schema's
            existing 1 Mb bin column and returns a per-bin variant count
            instead of row-level data — used for genome/chromosome-level
            zoom, where returning every row would be impractical.
        require_population_signal: uvar is exhaustive (every theoretical
            SNV, 3 rows/position) — for gene/cytoband-sized regions this
            is hundreds of thousands to millions of rows, almost all
            never-observed. When True, pushes down a row-group/predicate
            filter for an observed gnomAD frequency or a ClinVar
            classification (mirrors
            ``use_cases.gosling_common._has_population_signal``, which
            otherwise has to apply this same gate in Python *after* the
            full exhaustive table is materialized). dbSNP membership is
            deliberately not used as a signal — see
            ``_has_population_signal`` for why.

    Returns:
        A pyarrow.Table. Empty (zero columns) if the chromosome has no
        backing file.

    Raises:
        ValueError: If region.start or region.end is None.
    """
    if region.start is None or region.end is None:
        raise ValueError("region.start and region.end must be non-None for UvarVariantStore")

    dataset = self._dataset_for(region.chromosome)
    if dataset is None:
        return pa.table({})

    filt = (pc.field("position") >= region.start) & (pc.field("position") <= region.end)
    if require_population_signal:
        filt = filt & (
            pc.field("gnomad_genome", "af").is_valid()
            | (
                pc.list_value_length(  # pyright: ignore[reportAttributeAccessIssue]
                    pc.field("clinvar", "clnsig")
                )
                > 0
            )
        )
    scan_columns = list(dict.fromkeys((*columns, "position", "chromosome")))
    if aggregate == "pos_bin_1m" and "pos_bin_1m" not in scan_columns:
        scan_columns.append("pos_bin_1m")

    table = dataset.to_table(filter=filt, columns=scan_columns)

    if aggregate == "pos_bin_1m":
        table = table.group_by("pos_bin_1m").aggregate([("position", "count")])

    return table

query_positions_columns ¶

query_positions_columns(chrom, positions, *, columns)

Point-lookup rows for an explicit, possibly scattered set of positions.

Uses an isin predicate rather than :meth:query_region_columns's >=/<= range — still row-group-prunable via pyarrow's statistics, but the right tool when the targets are arbitrary positions from a variant list (e.g. annotate-by-key) rather than one contiguous window. Used by UvarAnnotator.

PARAMETER	DESCRIPTION
`chrom`	Chromosome, any spelling `HUMAN_CHROMOSOMES` resolves. TYPE: `str`
`positions`	1-based positions to fetch. TYPE: `Sequence[int]`
`columns`	Top-level uvar column names to project. TYPE: `Sequence[str]`

RETURNS	DESCRIPTION
`Table`	A pyarrow.Table. Empty (zero columns) if the chromosome has no
`Table`	backing file or positions is empty.

Source code in src/iris/adapters/driven/stores/uvar.py

def query_positions_columns(
    self, chrom: str, positions: Sequence[int], *, columns: Sequence[str]
) -> pa.Table:
    """Point-lookup rows for an explicit, possibly scattered set of positions.

    Uses an ``isin`` predicate rather than :meth:`query_region_columns`'s
    ``>=``/``<=`` range — still row-group-prunable via pyarrow's statistics,
    but the right tool when the targets are arbitrary positions from a variant
    list (e.g. annotate-by-key) rather than one contiguous window. Used by
    ``UvarAnnotator``.

    Args:
        chrom: Chromosome, any spelling ``HUMAN_CHROMOSOMES`` resolves.
        positions: 1-based positions to fetch.
        columns: Top-level uvar column names to project.

    Returns:
        A pyarrow.Table. Empty (zero columns) if the chromosome has no
        backing file or *positions* is empty.
    """
    if not positions:
        return pa.table({})
    dataset = self._dataset_for(chrom)
    if dataset is None:
        return pa.table({})

    filt = pc.field("position").isin(pa.array(positions, type=pa.int64()))
    scan_columns = list(dict.fromkeys((*columns, "position", "chromosome")))
    return dataset.to_table(filter=filt, columns=scan_columns)

fetch ¶

fetch(region)

Return VariantAnnotation objects within region, using a fixed default column set.

Intended for small (gene-level) regions — always requests _DEFAULT_FETCH_COLUMNS, never the full 416-column schema. Larger regions should use :meth:query_region_columns directly with a narrower, purpose-specific column set.

PARAMETER	DESCRIPTION
`region`	1-based inclusive genomic window. TYPE: `GenomicPosition`

RETURNS	DESCRIPTION
`list[VariantAnnotation]`	Possibly-empty list of VariantAnnotation.

RAISES	DESCRIPTION
`ValueError`	If region.start or region.end is None.

Source code in src/iris/adapters/driven/stores/uvar.py

def fetch(self, region: GenomicPosition) -> list[VariantAnnotation]:
    """Return VariantAnnotation objects within region, using a fixed default column set.

    Intended for small (gene-level) regions — always requests
    ``_DEFAULT_FETCH_COLUMNS``, never the full 416-column schema. Larger
    regions should use :meth:`query_region_columns` directly with a
    narrower, purpose-specific column set.

    Args:
        region: 1-based inclusive genomic window.

    Returns:
        Possibly-empty list of VariantAnnotation.

    Raises:
        ValueError: If region.start or region.end is None.
    """
    table = self.query_region_columns(region, columns=_DEFAULT_FETCH_COLUMNS)
    results = (
        _row_to_variant_annotation(row, assembly=self.assembly, source=self.source)
        for row in table.to_pylist()
    )
    return [r for r in results if r is not None]

VariantExtractionPipeline¶

Orchestrates parallel variant extraction followed by sequential annotation.

Two-phase execution¶

Extraction — the store's fetch() is called once per region using a ThreadPoolExecutor. Because Zarr stores release the GIL for array reads, multi-threading yields near-linear speedups on I/O-bound workloads.
Annotation — collected VariantAnnotation objects are passed sequentially through each annotator. Annotators run in serial to avoid saturating tabix file handles or in-memory lookup tables.

Usage¶

from iris.adapters.driven.pipeline import VariantExtractionPipeline
from iris.adapters.driven.stores import ZarrVariantStore
from iris.adapters.driven.annotators.clinvar import ClinvarAnnotator

annotators = (ClinvarAnnotator(path=CLINVAR_TSV),)
with ZarrVariantStore(path=ZARR_PATH) as store:
    pipeline = VariantExtractionPipeline(
        store, annotators, workers=28, show_progress=True
    )
    variants = pipeline.run(regions=regions_from_bed)

iris.adapters.driven.pipeline ¶

VariantExtractionPipeline — parallel extraction and sequential annotation.

VariantExtractionPipeline ¶

Orchestrates parallel variant extraction followed by sequential annotation.

The pipeline executes in two phases:

Extraction — :attr:store is queried once per region using a :class:~concurrent.futures.ThreadPoolExecutor with up to :attr:workers threads. Because Zarr stores release the GIL for array reads, multi-threading yields near-linear speedups.
Annotation — the collected :class:~iris.domain.genomics.entities.VariantAnnotation objects are passed through each annotator in :attr:annotators sequentially. Annotators run in serial to avoid saturating the I/O devices they read from (tabix files, in-memory lookup tables, etc.).

Example::

annotators = (ClinvarAnnotator(path=CLINVAR_TSV),)
with ZarrVariantStore(path=ZARR_PATH) as store:
    pipeline = VariantExtractionPipeline(store, annotators, workers=28)
    variants = pipeline.run(regions=regions_from_bed)

ATTRIBUTE	DESCRIPTION
`store`	Opened :class:`~iris.domain.genomics.ports.driven.variant_store.VariantAnnotationStore` to fetch variants from. TYPE: `VariantAnnotationStore`
`annotators`	Zero or more :class:`~iris.domain.genomics.ports.driven.annotators.VariantAnnotator` instances applied in order after extraction. TYPE: `tuple[VariantAnnotator, ...]`
`workers`	Number of threads for the extraction phase. Use `1` for deterministic sequential execution. TYPE: `int`
`show_progress`	When `True`, emit a `tqdm` progress bar over the region tasks. Requires `tqdm` to be installed; falls back to plain logging when absent. TYPE: `bool`

run ¶

run(*, regions=None, variants=None)

Run the pipeline and return fully annotated variants.

Exactly one of regions or variants must be supplied.

regions — each region is submitted as one extraction task.
variants — each variant is converted to a point region (pos, pos) and then de-duplicated by region to avoid redundant store reads. Variants without a valid position are skipped.

PARAMETER	DESCRIPTION
`regions`	Genomic windows to extract variants from. TYPE: `Sequence[GenomicPosition] \| None` DEFAULT: `None`
`variants`	Specific variants whose positions define point regions. TYPE: `Sequence[Variant] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[VariantAnnotation]`	Fully annotated list of
`list[VariantAnnotation]`	class:`~iris.domain.genomics.entities.VariantAnnotation`.

RAISES	DESCRIPTION
`ValueError`	If both or neither of regions and variants are supplied.

Source code in src/iris/adapters/driven/pipeline.py

def run(
    self,
    *,
    regions: Sequence[GenomicPosition] | None = None,
    variants: Sequence[Variant] | None = None,
) -> list[VariantAnnotation]:
    """Run the pipeline and return fully annotated variants.

    Exactly one of *regions* or *variants* must be supplied.

    * **regions** — each region is submitted as one extraction task.
    * **variants** — each variant is converted to a point region
      ``(pos, pos)`` and then de-duplicated by region to avoid redundant
      store reads.  Variants without a valid position are skipped.

    Args:
        regions: Genomic windows to extract variants from.
        variants: Specific variants whose positions define point regions.

    Returns:
        Fully annotated list of
        :class:`~iris.domain.genomics.entities.VariantAnnotation`.

    Raises:
        ValueError: If both or neither of *regions* and *variants* are
            supplied.
    """
    if (regions is None) == (variants is None):
        raise ValueError("Exactly one of 'regions' or 'variants' must be provided.")

    tasks = self._resolve_tasks(regions=regions, variants=variants)
    logger.info(
        "pipeline start — %s tasks · workers=%d · %s",
        f"{len(tasks):,}",
        self.workers,
        _fmt_rss(_rss_gb()),
    )
    extracted = self._extract_parallel(tasks)
    logger.info(
        "extraction done — %s variants · %s", f"{len(extracted):,}", _fmt_rss(_rss_gb())
    )
    return self._annotate_sequential(extracted)

Stores & Pipeline¶

ZarrVariantStore¶

Key design points¶

Usage¶

Genotype processors¶

iris.adapters.driven.stores.zarr ¶

ZarrVariantStore ¶

processor class-attribute instance-attribute ¶

processor_context class-attribute instance-attribute ¶

sample_ids property ¶

open ¶

close ¶

fetch ¶

scan ¶

UvarVariantStore¶

Key design points¶

Usage¶

iris.adapters.driven.stores.uvar ¶

COLUMN_GROUPS module-attribute ¶

UvarVariantStore ¶

query_region_columns ¶

query_positions_columns ¶

fetch ¶

VariantExtractionPipeline¶

Two-phase execution¶

Usage¶

iris.adapters.driven.pipeline ¶

VariantExtractionPipeline ¶

run ¶

processor `class-attribute` `instance-attribute` ¶

processor_context `class-attribute` `instance-attribute` ¶

sample_ids `property` ¶

COLUMN_GROUPS `module-attribute` ¶