"""
ClinVar VCV/RCV database queries.
This module encodes some limited, but frequently used, queries to extract
clinvar data.
"""
from __future__ import annotations
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
import os
import csv
import sys
import sqlite3
import functools
import pandas as pd
import argparse
import logging
from abc import ABC, abstractmethod
from contextlib import contextmanager
from pathlib import Path
from typing import (
Literal,
Self,
NamedTuple,
)
from clinvar_build.errors import (
is_type,
)
from clinvar_build.constants import (
ASSEMBLY_GRCH37,
ASSEMBLY_GRCH38,
DEFAULT_ASSEMBLY,
RCV_ENRICHMENT_QUERY,
VCV_ENRICHMENT_QUERY,
CONCORDANCE_SUBQUERY,
QueryOutNames as QONames,
)
from clinvar_build.utils.parser_tools import configure_logging
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# constants
SECOND_DB_ERR = "No secondary database connection available"
EMPTY_TABLE = "The DataFrame is empty"
NO_JOIN_COL = "DataFrame must contain {0} column for enrichment"
NO_JOIN_VAL = "No non-null values in {0} column"
Assembly = Literal[ASSEMBLY_GRCH37, ASSEMBLY_GRCH38]
# initiating a logger
logger = logging.getLogger(__name__)
# Tables for which a missing-table warning has already been emitted this
# process. Module-level so all query instances share the dedup set.
# Tests must clear this between runs: queries._warned_missing_tables.clear()
_warned_missing_tables: set[str] = set()
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@contextmanager
def _atomic_write(output_path: str | Path):
"""
Context manager creates a temp file and moves this to output_path.
Parameters
----------
output_path : `str` or `Path`
Intended final file path.
Yields
------
tmp : `Path`
Hidden temp file path in the same directory. Caller writes to this.
Notes
-----
On success the temp file is renamed to ``output_path`` (atomic on the
same filesystem). On any exception (including ``KeyboardInterrupt``) the
temp file is deleted and the exception is re-raised, leaving no partial
output at the intended path.
"""
# original name
output_path = Path(output_path)
# create random temp file name
tmp = output_path.parent / f".{output_path.name}.{os.urandom(4).hex()}.tmp"
try:
# using yield here to pause the function to wait until a while look
# has finished (or failed).
yield tmp
os.replace(tmp, output_path)
except BaseException:
tmp.unlink(missing_ok=True)
raise
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _run_cli(func):
"""
Decorator to provide graceful exit handling for CLI methods.
Catches ``KeyboardInterrupt``, ``BrokenPipeError``, and unexpected
exceptions, logs an appropriate message, and calls ``sys.exit`` with a
conventional exit code.
Parameters
----------
func : callable
CLI main function to wrap.
Returns
-------
callable
Wrapped function with exit handling.
Notes
-----
Exit codes follow Unix convention: 130 for SIGINT (128+2), 0 for broken
pipe (normal consumer-side close), 1 for unexpected errors.
``BrokenPipeError`` stdout is redirected to ``/dev/null`` to suppress
Python's own flush-time error message at interpreter shutdown.
"""
# copy the correct function name and docus using wraps
@functools.wraps(func)
# The actual wrapping function
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except KeyboardInterrupt:
logger.warning("Interrupted.")
sys.exit(130)
except BrokenPipeError:
devnull = os.open(os.devnull, os.O_WRONLY)
os.dup2(devnull, sys.stdout.fileno())
sys.exit(0)
except Exception as e:
logger.error("Fatal error: %s", e)
sys.exit(1)
return wrapper
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _mc_fragments(query: "BaseQuery") -> tuple[str, str]:
"""
Return SELECT and JOIN fragments for the MolecularConsequence table.
Parameters
----------
query : `BaseQuery`
Query instance whose connection is checked for the table.
Returns
-------
select : `str`
SELECT fragment for variant_type and molecular_consequences.
join : `str`
JOIN clause, or empty string when the table is absent.
"""
fragments = query._optional_fragments(
table_name="MolecularConsequence",
select_present=(
"sa.variant_type,\n"
" mc.function AS molecular_consequences"
),
join_present=(
"LEFT JOIN MolecularConsequence AS mc "
"ON mc.simple_allele_id = sa.id"
),
select_absent=(
"sa.variant_type,\n"
" NULL AS molecular_consequences"
),
)
return fragments
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class GenomicLocation(NamedTuple):
"""
Genomic location for variant lookup.
Parameters
----------
chr : `str`
Chromosome (e.g., '19', 'X').
position : `int`
VCF position (1-based).
ref : `str`, default `None`
Reference allele.
alt : `str`, default `None`
Alternate allele.
assembly : {'GRCh37', 'GRCh38'}, default `GRCh38`
Genome assembly.
Examples
--------
>>> loc = GenomicLocation("19", 11089362)
>>> loc = GenomicLocation("19", 11089362, "A", "G")
>>> loc = GenomicLocation("19", 11089362, assembly="GRCh37")
"""
chr: str
position: int
ref: str | None = None
alt: str | None = None
assembly: Assembly = DEFAULT_ASSEMBLY
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class GenomicRange(NamedTuple):
"""
Genomic range for variant queries.
Parameters
----------
chr : 'str'
Chromosome (e.g., '19', 'X').
start : 'int'
Start position as basepair position.
end : 'int'
End position as basepair position.
assembly : {'GRCh37', 'GRCh38'}, default 'GRCh38'
Genome assembly.
Examples
--------
>>> r = GenomicRange("19", 11000000, 12000000)
>>> r = GenomicRange("19", 11000000, 12000000, assembly="GRCh37")
"""
chr: str
start: int
end: int
assembly: Assembly = DEFAULT_ASSEMBLY
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
# NOTE add AS in the SQL statement to clarify
[docs]
class BaseQuery(ABC):
"""
Abstract base class for ClinVar database queries.
Handles database connection, parameter building, and query
execution. Subclasses implement specific query logic.
Parameters
----------
db_path : `str` or `Path`
Path to the primary SQLite database file.
secondary_db_path : `str` or `Path`, optional
Path to secondary database for enrichment.
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# class attributes
# _primary_db: subclasses set this to indicate primary database type which
# should always be used.
_primary_db: Literal["vcv", "rcv"] = "vcv"
# default join queries
_rcv_enrichment_query: str = RCV_ENRICHMENT_QUERY
_vcv_enrichment_query: str = VCV_ENRICHMENT_QUERY
_join_column: str = "canonical_spdi"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __init__(
self,
db_path: str | Path,
secondary_db_path: str | Path | None = None,
):
"""Initialize querier."""
is_type(db_path, (str, Path))
is_type(secondary_db_path, (type(None), str, Path))
self.db_path = Path(db_path)
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.secondary_db_path = None
self.secondary_conn = None
if secondary_db_path:
self.secondary_db_path = Path(secondary_db_path)
self.secondary_conn = sqlite3.connect(self.secondary_db_path)
self.secondary_conn.row_factory = sqlite3.Row
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __repr__(self) -> str:
"""
Return unambiguous string representation.
"""
secondary = (
f", secondary={self.secondary_db_path}"
if self.secondary_db_path
else ""
)
return f"{type(self).__name__}({self.db_path}{secondary})"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __str__(self) -> str:
"""
Return human-readable string representation.
"""
return f"{type(self).__name__}({self.db_path.name})"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def close(self):
"""Close database connections."""
self.conn.close()
self.conn = None
if self.secondary_conn:
self.secondary_conn.close()
self.secondary_conn = None
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __enter__(self) -> Self:
"""Enter context manager.
Returns
-------
Self
The query instance.
"""
return self
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __exit__(
self,
*_: object,
) -> None:
"""Exit context manager and close connections.
Notes
-----
Exception parameters are part of the context manager protocol but
are not used in this base implementation. Connections are closed
regardless of whether an exception occurred, and any exception
propagates normally. Subclasses may override to add custom
exception handling such as transaction rollback.
"""
self.close()
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _execute(
self,
query: str,
query_params: list[str | int | float | bytes | None] | None = None,
secondary: bool = False,
output_path: str | Path | None = None,
) -> pd.DataFrame | int:
"""Execute query and return DataFrame or stream to disk.
Parameters
----------
query : `str`
SQL query string.
query_params : `list`, default `None`
Query parameters.
secondary : `bool`, default False
Execute on secondary database instead.
output_path : `str` or `Path`, default None
If provided, stream results to this file (TSV format) and
return row count. If None, return results as DataFrame.
Returns
-------
pd.DataFrame or int
Query results as DataFrame, or row count if streaming to
disk.
"""
# input
is_type(query, str)
is_type(query_params, (type(None), list))
is_type(secondary, bool)
is_type(output_path, (str, Path, type(None)))
# setting up the params and the connection
params = query_params or []
conn = self.secondary_conn if secondary else self.conn
# raise error if no connection
if conn is None:
db_type = "secondary" if secondary else "primary"
raise ValueError(f"No {db_type} database connection available")
# execute the query
cursor = conn.execute(query, params)
# return to disk
if output_path:
return self._stream_to_file(cursor, output_path)
# return pd.DataFrame
rows = [dict(row) for row in cursor.fetchall()]
return pd.DataFrame(rows)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _stream_to_file(
self,
cursor: sqlite3.Cursor,
output_path: str | Path,
delim: str = "\t",
mode: str = "w",
) -> int:
r"""
Stream cursor results to TSV file.
Parameters
----------
cursor : `sqlite3.Cursor`
Executed cursor to stream from.
output_path : `str` or `Path`
Output file path including the file name.
delim : `str`, default ``\t``
Field delimiter.
mode : `str`, default `w`
File open mode, use wz for gzip compression.
Returns
-------
int
Number of rows written.
"""
# confirm input
is_type(output_path, (str, Path))
is_type(delim, str)
is_type(mode, str)
# get the headers
headers = [desc[0] for desc in cursor.description]
# write to disk atomically via hidden temp file
with _atomic_write(output_path) as tmp:
with open(tmp, mode, newline="") as f:
writer = csv.writer(f, delimiter=delim)
writer.writerow(headers)
count = 0
for row in cursor:
writer.writerow(row)
count += 1
# return number or rows
return count
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _build_placeholders(self, items: list) -> str:
"""
Build comma-separated placeholders for parameterised query.
Parameters
----------
items : `list` [`Any`]
Query parameter values. Only the length is used.
Returns
-------
str
Placeholder string (e.g., '?,?,?').
"""
return ",".join("?" * len(items))
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _optional_fragments(
self,
table_name: str,
select_present: str,
join_present: str,
select_absent: str,
conn: sqlite3.Connection | None = None,
) -> tuple[str, str]:
"""
Return SELECT and JOIN fragments for an optional table.
Checks whether *table_name* exists in *conn*. When present, returns
the caller-supplied fragments; when absent, returns *select_absent*
with an empty JOIN and emits a deduplicated warning.
Parameters
----------
table_name : `str`
Name of the optional table to check.
select_present : `str`
SELECT fragment to use when the table exists.
join_present : `str`
JOIN clause to use when the table exists.
select_absent : `str`
SELECT fragment (typically NULL columns) when the table is absent.
conn : `sqlite3.Connection`, optional
Connection to check. Defaults to ``self.conn`` when ``None``.
Returns
-------
select : `str`
SELECT fragment.
join : `str`
JOIN clause, or empty string when the table is absent.
"""
target = conn if conn is not None else self.conn
cursor = target.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table_name,),
)
if cursor.fetchone():
return select_present, join_present
if table_name not in _warned_missing_tables:
logger.warning(
"Table %r absent — affected columns set to None.", table_name
)
_warned_missing_tables.add(table_name)
return select_absent, ""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _build_where(self, clauses: list[str]) -> str:
"""
Build WHERE clause from list of conditions.
Parameters
----------
clauses : `list` [`str`]
Individual WHERE conditions (e.g., ['g.symbol = ?']).
Returns
-------
str
Complete WHERE clause or empty string.
"""
if not clauses:
return ""
return "WHERE " + " AND ".join(clauses)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _enrich(
self,
table: pd.DataFrame,
source: Literal["rcv", "vcv"],
query: str | None = None,
how: str = "left",
) -> pd.DataFrame:
"""
Enrich query results with data from secondary database.
Parameters
----------
table : `pd.DataFrame`
Query results with canonical_spdi column.
source : {'rcv', 'vcv'}
Enrichment source: 'rcv' for RCV condition data,
'vcv' for VCV aggregate classification.
query : `str`, default None
Custom enrichment query. If None, uses default query for
the specified source. Custom queries must select the join
column specified by _join_column.
how : `str`, default `left`
Returns
-------
pd.DataFrame
Enriched dataframe.
"""
# types
is_type(table, pd.DataFrame)
is_type(source, str)
is_type(query, (type(None), str))
is_type(how, str)
# errors
if source not in ["rcv", "vcv"]:
raise ValueError(f"source must be 'rcv' or 'vcv', got {source}")
if self.secondary_conn is None:
raise ValueError(SECOND_DB_ERR)
if table.empty:
raise ValueError(EMPTY_TABLE)
if self._join_column not in table.columns:
raise ValueError(NO_JOIN_COL.format(self._join_column))
# get the unique ids
values = table[self._join_column].dropna().unique().tolist()
if not values:
raise ValueError(NO_JOIN_VAL.format(self._join_column))
# select default query
placeholders = self._build_placeholders(values)
if source == "vcv":
if query is not None:
formatted = query.format(spdi_placeholders=placeholders)
else:
mc_select, mc_join = self._optional_fragments(
table_name="MolecularConsequence",
select_present=(
"sa.variant_type,\n"
" mc.function AS molecular_consequences"
),
join_present=(
"LEFT JOIN MolecularConsequence AS mc "
"ON mc.simple_allele_id = sa.id"
),
select_absent=(
"sa.variant_type,\n"
" NULL AS molecular_consequences"
),
conn=self.secondary_conn,
)
formatted = self._vcv_enrichment_query.format(
mc_select=mc_select,
mc_join=mc_join,
spdi_placeholders=placeholders,
)
enrichment_table = self._execute(formatted, values, secondary=True)
else:
if query is None:
query = self._rcv_enrichment_query
query = query.format(spdi_placeholders=placeholders)
enrichment_table = self._execute(query, values, secondary=True)
if enrichment_table.empty:
raise ValueError(
f"{source.upper()} enrichment query returned no results"
)
# merge
return table.merge(enrichment_table, on=self._join_column, how=how)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _cast_location_types(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Cast chr column values to appropriate Python types.
DataFrames without a QONames.chrom column are returned unchanged.
Numeric string values are cast to ``int``; non-numeric strings remain
as ``str``; ``None`` values remain as ``None``.
Parameters
----------
df : `pd.DataFrame`
Query result dataframe, may or may not contain a QONames.chrom
column.
Returns
-------
df : `pd.DataFrame`
DataFrame with QONames.chrom values cast to ``int`` where numeric,
otherwise left as ``str`` or ``None``.
Notes
-----
Chromosome values from SQLite can be numeric (``'1'``–``'22'``) or
non-numeric (``'X'``, ``'Y'``, ``'MT'``). Casting numeric values to
``int`` avoids unintended string-sorting behaviour downstream.
"""
if QONames.chrom not in df.columns:
return df
def _cast(v: object) -> int | str | None:
if v is None or pd.isna(v):
return None
try:
return int(v)
except (ValueError, TypeError):
return str(v)
df = df.copy()
df[QONames.chrom] = df[QONames.chrom].apply(_cast)
return df
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _execute_and_enrich(
self,
query: str,
params: list,
enrich: bool = False,
output_path: str | Path | None = None,
grch38: bool = True,
grch37: bool = True,
) -> pd.DataFrame | int:
"""
Execute a query, optionally enrich, and optionally write to disk.
When enrichment is requested the result is always loaded into memory
first, enriched, then optionally written to disk. Without enrichment,
results are streamed directly to disk when ``output_path`` is given.
Parameters
----------
query : `str`
SQL query string.
params : `list`
Query parameters.
enrich : `bool`, default False
Enrich results with RCV condition data via secondary connection.
output_path : `str` or `Path`, default None
If provided, write results to this file (TSV) and return row count.
grch38 : `bool`, default True
Include GRCh38 assembly rows.
grch37 : `bool`, default True
Include GRCh37 assembly rows.
Returns
-------
pd.DataFrame or int
Enriched DataFrame, or row count when ``output_path`` is set.
"""
# --- enrich path (always in-memory) ---
if enrich and self.secondary_conn is not None:
result = self._execute(query, params)
result = self._cast_location_types(result)
result = self._enrich(result, source="rcv")
if output_path:
with _atomic_write(output_path) as tmp:
result.to_csv(tmp, sep="\t", index=False)
return len(result)
return result
# --- long path: streaming shortcut available ---
if output_path:
return self._execute(query, params, output_path=output_path)
result = self._execute(query, params)
return self._cast_location_types(result)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
@abstractmethod
def query(self, **kwargs) -> pd.DataFrame:
"""Execute query. Implemented by subclasses."""
pass
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class FilterQuery(BaseQuery):
"""
Filter variants by gene, condition, and classification.
This query class operates on the VCV (Variation Archive) database
as its primary source. When enrichment is enabled, condition data
is fetched from the RCV database and merged via canonical_spdi.
Returns one row per variant-assembly-condition combination. Each row
contains QONames.assembly, QONames.chrom, QONames.start, QONames.stop,
QONames.position_vcf, QONames.reference_allele_vcf, and
QONames.alternate_allele_vcf columns for that assembly. Because
QONames.cc_name, QONames.cc_db, QONames.cc_condition_id, and
QONames.rc_classification are one-to-many with respect to a variant, a
single variant can produce multiple rows.
Methods
-------
query(genes, condition_patterns, condition_ids, classifications, ...)
Filter variants matching specified criteria.
Examples
--------
>>> q = FilterQuery("clinvar_vcv.db", secondary_db_path="clinvar_rcv.db")
>>> df = q.query(
... genes=["LDLR"],
... classifications=["Pathogenic"],
... enrich=True,
... )
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# class attributes
_primary_db = "vcv"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def query(
self,
genes: list[str] | None = None,
condition_names: list[str] | None = None,
condition_identifiers: list[tuple[str, str]] | None = None,
classifications: list[str] | None = None,
grch38: bool = True,
grch37: bool = True,
enrich: bool = False,
output_path: str | Path | None = None,
) -> pd.DataFrame | int:
"""
Filter variants with optional enrichment.
Parameters
----------
genes : `list` [`str`], default None
Filter by gene symbols (e.g., ['LDLR', 'APOB']).
condition_names : `list` [`str`], default None
Filter by condition name using SQL LIKE patterns
(e.g., ['%cardiomyopathy%']) or exact matches.
condition_identifiers : `list` [`tuple`], default None
Filter by condition identifier as (db, id) tuples
(e.g., [('MedGen', 'C0007194'), ('OMIM', '115200')]).
classifications : `list` [`str`], default None
Filter by classification (e.g., ['Pathogenic']).
grch38 : `bool`, default True
Include GRCh38 assembly rows.
grch37 : `bool`, default True
Include GRCh37 assembly rows.
enrich : `bool`, default False
Enrich with data from secondary database.
output_path : `str` or `Path`, default None
If provided, stream results to file and return row count.
Returns
-------
pd.DataFrame or int
Filtered variants. One row per variant-assembly-condition
combination, or row count when streaming to disk.
Raises
------
ValueError
If both ``grch38`` and ``grch37`` are False.
"""
# check input
is_type(genes, (type(None), list))
is_type(condition_identifiers, (type(None), list))
is_type(condition_names, (type(None), list))
is_type(classifications, (type(None), list))
is_type(grch38, bool)
is_type(grch37, bool)
is_type(enrich, bool)
is_type(output_path, (type(None), Path, str))
if not grch38 and not grch37:
raise ValueError("At least one assembly must be selected")
if condition_names and condition_identifiers:
raise ValueError(
"Cannot use both condition_names and condition_ids; "
"use one or the other"
)
# build ordered list of requested assemblies (GRCh38 first)
requested_assemblies: list[str] = []
if grch38:
requested_assemblies.append(ASSEMBLY_GRCH38)
if grch37:
requested_assemblies.append(ASSEMBLY_GRCH37)
params = []
where_clauses = []
# Build gene filter
if genes:
where_clauses.append(
f"g.symbol IN ({self._build_placeholders(genes)})"
)
params.extend(genes)
# Build condition name pattern filter
# NOTE always LEFT JOIN condition tables so columns are always present
cond_join = """
LEFT JOIN RCVAccession ra ON ra.variation_archive_id = va.id
LEFT JOIN ClassifiedCondition cc ON cc.rcv_accession_id = ra.id
LEFT JOIN RCVClassification rc ON rc.rcv_accession_id = ra.id"""
if condition_names:
cond_clauses = " OR ".join(
"cc.name LIKE ?" for _ in condition_names
)
where_clauses.append(f"({cond_clauses})")
params.extend(condition_names)
# Build condition identifier filter
if condition_identifiers:
id_clauses = " OR ".join(
"(cc.db = ? AND cc.condition_id = ?)"
for _ in condition_identifiers
)
where_clauses.append(f"({id_clauses})")
for db, cid in condition_identifiers:
params.extend([db, cid])
# Build classification filter
if classifications:
where_clauses.append(
f"ac.description IN "
f"({self._build_placeholders(classifications)})"
)
params.extend(classifications)
# where statement
where = self._build_where(where_clauses)
# single JOIN parameterised on the requested assemblies
assembly_placeholders = self._build_placeholders(requested_assemblies)
loc_join = f"""
LEFT JOIN AlleleLocation al
ON al.simple_allele_id = sa.id
AND al.assembly IN ({assembly_placeholders})"""
# concordance subquery
conc_join = CONCORDANCE_SUBQUERY
# prepend assembly params — they come before WHERE clause params
params = list(requested_assemblies) + params
mc_select, mc_join = _mc_fragments(self)
query = f"""
SELECT DISTINCT
va.accession AS vcv_accession,
va.variation_id,
va.variation_name,
g.symbol AS gene_symbol,
ac.description AS ac_classification,
ac.review_status AS ac_review_status,
cc.name AS cc_name,
cc.db AS cc_db,
cc.condition_id AS cc_condition_id,
rc.description AS rc_classification,
sa.canonical_spdi,
al.assembly,
al.chr,
al.start,
al.stop,
al.position_vcf,
al.reference_allele_vcf,
al.alternate_allele_vcf,
conc.submitted_classifications,
conc.classification_count,
{mc_select}
FROM VariationArchive va
JOIN SimpleAllele sa ON sa.variation_archive_id = va.id
JOIN Gene g ON g.simple_allele_id = sa.id
LEFT JOIN AggregateClassification ac
ON ac.variation_archive_id = va.id
AND ac.classification_type = 'GermlineClassification'
{cond_join}
{loc_join}
{conc_join}
{mc_join}
{where}
ORDER BY g.symbol, va.accession, al.chr, al.position_vcf
"""
return self._execute_and_enrich(
query,
params,
enrich=enrich,
output_path=output_path,
grch38=grch38,
grch37=grch37,
)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class FetchVariantQuery(BaseQuery):
"""
Fetch specific variants by identifier, name, or position.
This query class operates on the VCV (Variation Archive) database
as its primary source. When enrichment is enabled, condition data
is fetched from the RCV database and merged via canonical_spdi.
Assembly behaviour
------------------
Returns one row per (variant × assembly). When ``location`` is given,
only the assembly specified in ``location.assembly`` (default GRCh38)
is returned.
Methods
-------
query(vcv_accessions, variation_ids, variant_names, ...)
Fetch variants matching specified identifiers or positions.
Examples
--------
>>> q = FetchVariantQuery("clinvar_vcv.db", "clinvar_rcv.db")
>>> # two rows per variant (GRCh38 + GRCh37)
>>> df = q.query(vcv_accessions=["VCV000017639"])
>>> # GRCh37 coordinates only
>>> df = q.query(location=GenomicLocation("19", 11089362), grch38=False)
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# class attributes
_primary_db = "vcv"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def query(
self,
vcv_accessions: list[str] | None = None,
variation_identifiers: list[int] | None = None,
variant_names: list[str] | None = None,
variant_name_pattern: str | None = None,
location: GenomicLocation | None = None,
spdi: str | None = None,
grch38: bool = True,
grch37: bool = True,
enrich: bool = False,
output_path: str | Path | None = None,
) -> pd.DataFrame | int:
"""
Fetch variants by identifier or position.
Parameters
----------
vcv_accessions : `list` [`str`], default None
VCV accession numbers (e.g., ['VCV000017639']).
variation_identifiers : `list` [`int`], default None
ClinVar variation IDs (e.g., [17639]).
variant_names : `list` [`str`], default None
Exact variant names (e.g., ['NM_000527.5:c.1A>G']).
variant_name_pattern : `str`, default None
Variant name pattern using SQL LIKE patterns.
location : `GenomicLocation`, default None
Genomic location as (chr, position, ref, alt). Only chr and
position are required. ``location.assembly`` (default 'GRCh38')
restricts results to that assembly.
spdi : `str`, default None
Canonical SPDI notation.
grch38 : `bool`, default True
Include GRCh38 assembly rows.
grch37 : `bool`, default True
Include GRCh37 assembly rows.
enrich : `bool`, default False
Enrich with RCV condition data.
output_path : `str` or `Path`, default None
If provided, stream results to file and return row count.
Returns
-------
pd.DataFrame or int
Matching variants. One row per variant-assembly combination,
or row count when streaming to disk.
Raises
------
ValueError
If both ``grch38`` and ``grch37`` are False, or if the
``location`` assembly conflicts with the assembly flags.
"""
# check input types
is_type(vcv_accessions, (type(None), list))
is_type(variation_identifiers, (type(None), list))
is_type(variant_names, (type(None), list))
is_type(variant_name_pattern, (type(None), str))
is_type(location, (type(None), GenomicLocation))
is_type(spdi, (type(None), str))
is_type(grch38, bool)
is_type(grch37, bool)
is_type(enrich, bool)
is_type(output_path, (type(None), Path, str))
# validate assembly selection
if not grch38 and not grch37:
raise ValueError("At least one assembly must be selected")
# validate location vs assembly conflict
if location is not None:
implied_assembly = location.assembly
if implied_assembly == ASSEMBLY_GRCH38 and not grch38:
raise ValueError(
f"Location filter targets {implied_assembly} but "
f"grch38={grch38} and grch37={grch37} exclude it. "
"Adjust assembly flags or remove location filter."
)
if implied_assembly == ASSEMBLY_GRCH37 and not grch37:
raise ValueError(
f"Location filter targets {implied_assembly} but "
f"grch38={grch38} and grch37={grch37} exclude it. "
"Adjust assembly flags or remove location filter."
)
# validate lookup params
lookup_params = [
vcv_accessions is not None,
variation_identifiers is not None,
variant_names is not None,
variant_name_pattern is not None,
location is not None,
spdi is not None,
]
if sum(lookup_params) > 1:
raise ValueError(
"Only one lookup method allowed: vcv_accessions, "
"variation_identifiers, variant_names, variant_name_pattern, "
"location, or spdi"
)
if sum(lookup_params) == 0:
raise ValueError("At least one lookup method required")
# build ordered list of requested assemblies (GRCh38 first)
requested_assemblies: list[str] = []
if grch38:
requested_assemblies.append(ASSEMBLY_GRCH38)
if grch37:
requested_assemblies.append(ASSEMBLY_GRCH37)
params: list = []
where_clauses: list[str] = []
# identifier filters
if vcv_accessions:
where_clauses.append(
f"va.accession IN ({self._build_placeholders(vcv_accessions)})"
)
params.extend(vcv_accessions)
if variation_identifiers:
where_clauses.append(
"va.variation_id IN "
f"({self._build_placeholders(variation_identifiers)})"
)
params.extend(variation_identifiers)
# name filters
if variant_names:
where_clauses.append(
"va.variation_name IN "
f"({self._build_placeholders(variant_names)})"
)
params.extend(variant_names)
if variant_name_pattern:
where_clauses.append("va.variation_name LIKE ?")
params.append(variant_name_pattern)
# spdi filter
if spdi:
where_clauses.append("sa.canonical_spdi = ?")
params.append(spdi)
# location filter — filter against the single `al` alias
if location is not None:
where_clauses.append("al.chr = ?")
params.append(location.chr)
where_clauses.append("al.position_vcf = ?")
params.append(location.position)
if location.ref is not None:
where_clauses.append("al.reference_allele_vcf = ?")
params.append(location.ref)
if location.alt is not None:
where_clauses.append("al.alternate_allele_vcf = ?")
params.append(location.alt)
# where statement
where = self._build_where(where_clauses)
# single parameterised JOIN on requested assemblies
assembly_placeholders = self._build_placeholders(requested_assemblies)
loc_join = f"""
LEFT JOIN AlleleLocation al
ON al.simple_allele_id = sa.id
AND al.assembly IN ({assembly_placeholders})"""
# concordance subquery
conc_join = CONCORDANCE_SUBQUERY
# prepend assembly params — they come before WHERE clause params
params = list(requested_assemblies) + params
mc_select, mc_join = _mc_fragments(self)
# The actual query
query = f"""
SELECT DISTINCT
va.accession AS vcv_accession,
va.variation_id,
va.variation_name,
g.symbol AS gene_symbol,
ac.description AS classification,
ac.review_status,
sa.canonical_spdi,
al.assembly,
al.chr,
al.start,
al.stop,
al.position_vcf,
al.reference_allele_vcf,
al.alternate_allele_vcf,
conc.submitted_classifications,
conc.classification_count,
{mc_select}
FROM VariationArchive va
JOIN SimpleAllele sa ON sa.variation_archive_id = va.id
JOIN Gene g ON g.simple_allele_id = sa.id
LEFT JOIN AggregateClassification ac
ON ac.variation_archive_id = va.id
AND ac.classification_type = 'GermlineClassification'
{loc_join}
{conc_join}
{mc_join}
{where}
ORDER BY al.chr, al.position_vcf
"""
return self._execute_and_enrich(
query,
params,
enrich=enrich,
output_path=output_path,
grch38=grch38,
grch37=grch37,
)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class FetchRangeQuery(BaseQuery):
"""
Fetch variants within genomic ranges.
This query class operates on the VCV (Variation Archive) database
as its primary source. When enrichment is enabled, condition data
is fetched from the RCV database and merged via canonical_spdi.
Assembly behaviour
------------------
Returns one row per (variant × assembly). Each ``GenomicRange`` carries
an assembly (default ``'GRCh38'``); only variants whose coordinates in
**that assembly** fall within the range are returned. Use ``grch38`` and
``grch37`` boolean params to restrict which assemblies are joined.
Methods
-------
query(ranges, genes, classifications, ...)
Fetch variants within specified genomic regions.
Examples
--------
Single range (GRCh38, the default):
>>> q = FetchRangeQuery("clinvar_vcv.db", "clinvar_rcv.db")
>>> df = q.query(ranges=[GenomicRange("19", 11000000, 12000000)])
GRCh37 range:
>>> df = q.query(ranges=[GenomicRange("19", 10850000, 11850000, "GRCh37")])
Both assemblies for the same locus:
>>> df = q.query(ranges=[
... GenomicRange("19", 11000000, 12000000, "GRCh38"),
... GenomicRange("19", 10850000, 11850000, "GRCh37"),
... ])
With filters and enrichment:
... ranges=[GenomicRange("19", 11000000, 12000000)],
... genes=["LDLR"],
... classifications=["Pathogenic"],
... enrich=True,
... )
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# class attributes
_primary_db = "vcv"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def query(
self,
ranges: list[GenomicRange],
genes: list[str] | None = None,
classifications: list[str] | None = None,
grch38: bool = True,
grch37: bool = True,
enrich: bool = False,
output_path: str | Path | None = None,
) -> pd.DataFrame | int:
"""
Fetch variants within genomic ranges.
Parameters
----------
ranges : `list` [`GenomicRange`]
Genomic ranges to query. Each ``GenomicRange`` carries an
assembly (default ``'GRCh38'``); only coordinates for that
assembly are matched. Pass ranges for multiple assemblies to
retrieve results across both GRCh37 and GRCh38.
genes : `list` [`str`], default None
Filter by gene symbols.
classifications : `list` [`str`], default None
Filter by classification.
grch38 : `bool`, default True
Include GRCh38 assembly rows in the JOIN.
grch37 : `bool`, default True
Include GRCh37 assembly rows in the JOIN.
enrich : `bool`, default `False`
Enrich with RCV condition data.
output_path : `str` or `Path`, default None
If provided, stream results to file and return row count.
Returns
-------
pd.DataFrame or int
Variants within specified ranges. One row per variant-assembly
combination, or row count when streaming to disk.
Raises
------
ValueError
If both ``grch38`` and ``grch37`` are False.
"""
# check input
is_type(ranges, list)
is_type(genes, (type(None), list))
is_type(classifications, (type(None), list))
is_type(grch38, bool)
is_type(grch37, bool)
is_type(enrich, bool)
is_type(output_path, (type(None), Path, str))
# validate assembly selection
if not grch38 and not grch37:
raise ValueError("At least one assembly must be selected")
# validate ranges
if not ranges:
raise ValueError("At least one range required")
# build ordered list of requested assemblies (GRCh38 first)
requested_assemblies: list[str] = []
if grch38:
requested_assemblies.append(ASSEMBLY_GRCH38)
if grch37:
requested_assemblies.append(ASSEMBLY_GRCH37)
# initiate lists
params: list = []
where_clauses = []
# build range clauses — per-range assembly filter uses the `al` alias
range_clauses = []
for r in ranges:
range_clauses.append(
"(al.chr = ? AND al.position_vcf BETWEEN ? AND ? "
"AND al.assembly = ?)"
)
params.extend([r.chr, r.start, r.end, r.assembly])
where_clauses.append(f"({' OR '.join(range_clauses)})")
# gene filter
if genes:
where_clauses.append(
f"g.symbol IN ({self._build_placeholders(genes)})"
)
params.extend(genes)
# classification filter
if classifications:
where_clauses.append(
f"ac.description IN "
f"({self._build_placeholders(classifications)})"
)
params.extend(classifications)
where = self._build_where(where_clauses)
# concordance subquery
conc_join = CONCORDANCE_SUBQUERY
# single parameterised JOIN on requested assemblies
assembly_placeholders = self._build_placeholders(requested_assemblies)
loc_join = f"""
LEFT JOIN AlleleLocation al
ON al.simple_allele_id = sa.id
AND al.assembly IN ({assembly_placeholders})"""
# prepend assembly params — they come before WHERE clause params
params = list(requested_assemblies) + params
mc_select, mc_join = _mc_fragments(self)
# The actual query
query = f"""
SELECT DISTINCT
va.accession AS vcv_accession,
va.variation_id,
va.variation_name,
g.symbol AS gene_symbol,
ac.description AS classification,
ac.review_status,
sa.canonical_spdi,
al.assembly,
al.chr,
al.start,
al.stop,
al.position_vcf,
al.reference_allele_vcf,
al.alternate_allele_vcf,
conc.submitted_classifications,
conc.classification_count,
{mc_select}
FROM VariationArchive va
JOIN SimpleAllele sa ON sa.variation_archive_id = va.id
JOIN Gene g ON g.simple_allele_id = sa.id
{loc_join}
LEFT JOIN AggregateClassification ac
ON ac.variation_archive_id = va.id
AND ac.classification_type = 'GermlineClassification'
{conc_join}
{mc_join}
{where}
ORDER BY al.chr, al.position_vcf
"""
return self._execute_and_enrich(
query,
params,
enrich=enrich,
output_path=output_path,
grch38=grch38,
grch37=grch37,
)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class ListConditionsQuery(BaseQuery):
"""
List condition/disease identifiers and metadata.
This query class operates on the RCV database as its primary
source.
Methods
-------
available_sources()
List available condition identifier sources.
query(source, include_frequency, min_frequency, ...)
Get unique condition identifiers.
Examples
--------
List available sources:
>>> q = ListConditionsQuery("clinvar_rcv.db")
>>> df = q.available_sources()
Query conditions from a specific source:
>>> df = q.query(source="MedGen")
Include variant frequency:
>>> df = q.query(
... source="MedGen",
... include_frequency=True,
... min_frequency=10,
... )
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# class attributes
_primary_db = "rcv"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def available_sources(self) -> pd.DataFrame:
"""
List available condition identifier sources from RCV database.
Returns
-------
pd.DataFrame or int
Sources with counts.
"""
query = """
SELECT
x.db AS source,
COUNT(DISTINCT x.xref_id) AS identifier_count,
COUNT(*) AS occurrence_count
FROM XRef x
WHERE x.entity_type = 'Trait'
GROUP BY x.db
ORDER BY occurrence_count DESC
"""
return self._execute(query)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def query(
self,
source: str | None = None,
include_frequency: bool = False,
min_frequency: int = 1,
output_path: str | Path | None = None,
) -> pd.DataFrame | int:
"""
Get unique condition identifiers from RCV database.
Parameters
----------
source : 'str', default None
Filter by source database (e.g., 'MedGen', 'OMIM'). If None,
returns all sources.
include_frequency : 'bool', default False
Include variant count per condition.
min_frequency : 'int'
Minimum variant count to include (default 1). Only applies when
include_frequency is True.
output_path : 'str' or 'Path', default None
If provided, stream results to file and return row count.
Returns
-------
pd.DataFrame or int
Unique conditions with identifiers, or row count if
streaming to disk.
"""
# check input
is_type(source, (type(None), str))
is_type(include_frequency, bool)
is_type(min_frequency, int)
is_type(output_path, (type(None), Path, str))
# validate
if min_frequency < 1:
raise ValueError(
f"min_frequency must be >= 1, got {min_frequency}"
)
# initiate lists
params = []
where_clauses = ["x.entity_type = 'Trait'"]
# subset on source
if source:
where_clauses.append("x.db = ?")
params.append(source)
where = self._build_where(where_clauses)
# counts
if include_frequency:
query = f"""
SELECT
x.db AS source,
x.xref_id AS identifier,
n.element_value AS condition_name,
t.type AS trait_type,
COUNT(DISTINCT ms.reference_assertion_id) AS variant_count
FROM XRef x
JOIN Trait t ON t.id = x.entity_id
JOIN TraitSet ts ON ts.id = t.trait_set_id
JOIN MeasureSet ms
ON ms.reference_assertion_id = ts.reference_assertion_id
LEFT JOIN Name n
ON n.entity_type = 'Trait'
AND n.entity_id = t.id
AND n.type = 'Preferred'
{where}
GROUP BY x.db, x.xref_id, n.element_value, t.type
HAVING variant_count >= ?
ORDER BY variant_count DESC
"""
params.append(min_frequency)
else:
query = f"""
SELECT DISTINCT
x.db AS source,
x.xref_id AS identifier,
n.element_value AS condition_name,
t.type AS trait_type
FROM XRef x
JOIN Trait t ON t.id = x.entity_id
LEFT JOIN Name n
ON n.entity_type = 'Trait'
AND n.entity_id = t.id
AND n.type = 'Preferred'
{where}
ORDER BY x.db, n.element_value
"""
# return
return self._execute(query, params, output_path=output_path)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
# the parser helpers
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _filter_parser() -> argparse.ArgumentParser:
"""Create argument parser for clinvar-filter command."""
parser = argparse.ArgumentParser(
description="Filter variants by gene/condition/classification"
)
parser.add_argument(
"vcv_db",
help="Path to VCV database",
)
parser.add_argument(
"--rcv-db",
help="Path to RCV database (for enrichment)",
)
parser.add_argument(
"-g", "--genes",
nargs="+",
help="Gene symbols (e.g., LDLR APOB)",
)
parser.add_argument(
"-n", "--condition-names",
nargs="+",
help="Condition name patterns (e.g., '%%cardiomyopathy%%')",
)
parser.add_argument(
"-i", "--condition-ids",
nargs="+",
help="Condition IDs as db:id (e.g., MedGen:C0007194)",
)
parser.add_argument(
"-c", "--classifications",
nargs="+",
help="Classifications (e.g., Pathogenic 'Likely pathogenic')",
)
parser.add_argument(
"--grch38",
action="store_true",
default=None,
help=(
"Include GRCh38 assembly rows. Passing --grch38 alone returns "
"only GRCh38; omit both flags to return both assemblies."
),
)
parser.add_argument(
"--grch37",
action="store_true",
default=None,
help=(
"Include GRCh37 assembly rows. Passing --grch37 alone returns "
"only GRCh37; omit both flags to return both assemblies."
),
)
parser.add_argument(
"-o", "--output",
required=True,
help="Output file (tsv)",
)
parser.add_argument(
"-v", "--verbose",
action="count",
default=0,
help="Increase verbosity (-v, -vv, -vvv)",
)
return parser
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _fetch_variant_parser() -> argparse.ArgumentParser:
"""Create argument parser for clinvar-fetch-variant command."""
parser = argparse.ArgumentParser(
description="Fetch variants by identifier or position"
)
parser.add_argument(
"vcv_db",
help="Path to VCV database",
)
parser.add_argument(
"--rcv-db",
help="Path to RCV database (for enrichment)",
)
parser.add_argument(
"-o", "--output",
required=True,
help="Output file (TSV)",
)
parser.add_argument(
"-v", "--verbose",
action="count",
default=0,
help="Increase verbosity (-v, -vv, -vvv)",
)
# lookup methods (mutually exclusive)
lookup_group = parser.add_argument_group("lookup method (one required)")
lookup = lookup_group.add_mutually_exclusive_group(required=True)
lookup.add_argument(
"--vcv",
nargs="+",
help="VCV accessions (e.g., VCV000017639)",
)
lookup.add_argument(
"-I", "--variation-ids",
nargs="+",
type=int,
help="Variation IDs",
)
lookup.add_argument(
"-n", "--names",
nargs="+",
help="Exact variant names",
)
lookup.add_argument(
"-N", "--name-pattern",
help="Variant name pattern using SQL LIKE patterns",
)
lookup.add_argument(
"-l", "--location",
help=(
"Genomic location as chr:pos, chr:pos:assembly, "
"chr:pos:ref:alt, or chr:pos:ref:alt:assembly"
),
)
lookup.add_argument(
"-s", "--spdi",
help="Canonical SPDI",
)
parser.add_argument(
"--grch38",
action="store_true",
default=None,
help=(
"Include GRCh38 assembly rows. Passing --grch38 alone returns "
"only GRCh38; omit both flags to return both assemblies."
),
)
parser.add_argument(
"--grch37",
action="store_true",
default=None,
help=(
"Include GRCh37 assembly rows. Passing --grch37 alone returns "
"only GRCh37; omit both flags to return both assemblies."
),
)
return parser
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _fetch_range_parser() -> argparse.ArgumentParser:
"""Create argument parser for clinvar-fetch-range command."""
parser = argparse.ArgumentParser(
description="Fetch variants within genomic ranges"
)
parser.add_argument(
"vcv_db",
help="Path to VCV database",
)
parser.add_argument(
"range_specs",
nargs="+",
help=(
"Ranges as chr:start-end or chr:start-end:assembly "
"(e.g., 19:11000000-12000000)"
),
)
parser.add_argument(
"--rcv-db",
help="Path to RCV database (for enrichment)",
)
parser.add_argument(
"-g", "--genes",
nargs="+",
help="Filter by gene symbols",
)
parser.add_argument(
"-c", "--classifications",
nargs="+",
help="Filter by classifications",
)
parser.add_argument(
"-o", "--output",
required=True,
help="Output file (TSV)",
)
parser.add_argument(
"--grch38",
action="store_true",
default=None,
help=(
"Include GRCh38 assembly rows. Passing --grch38 alone returns "
"only GRCh38; omit both flags to return both assemblies."
),
)
parser.add_argument(
"--grch37",
action="store_true",
default=None,
help=(
"Include GRCh37 assembly rows. Passing --grch37 alone returns "
"only GRCh37; omit both flags to return both assemblies."
),
)
parser.add_argument(
"-v", "--verbose",
action="count",
default=0,
help="Increase verbosity (-v, -vv, -vvv)",
)
return parser
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _list_conditions_parser() -> argparse.ArgumentParser:
"""Create argument parser for clinvar-list-conditions command."""
parser = argparse.ArgumentParser(
description="List condition identifiers from RCV database"
)
parser.add_argument(
"rcv_db",
help="Path to RCV database",
)
parser.add_argument(
"-s", "--source",
help="Filter by source (e.g., MedGen, OMIM)",
)
parser.add_argument(
"-f", "--include-frequency",
action="store_true",
help="Include variant counts",
)
parser.add_argument(
"-m", "--min-frequency",
type=int,
default=1,
help="Minimum variant count (default: 1)",
)
parser.add_argument(
"--list-sources",
action="store_true",
help="List available sources and exit",
)
parser.add_argument(
"-o", "--output",
required=True,
help="Output file (TSV)",
)
parser.add_argument(
"-v", "--verbose",
action="count",
default=0,
help="Increase verbosity (-v, -vv, -vvv)",
)
return parser
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
# main functions
[docs]
@_run_cli
def filter_main():
"""CLI entry point for FilterQuery."""
parser = _filter_parser()
args = parser.parse_args()
# configure logging
configure_logging(args.verbose)
# resolve assembly flags: passing only one flag excludes the other
grch38, grch37 = args.grch38, args.grch37
if grch38 is None and grch37 is None:
grch38, grch37 = True, True
elif grch38 is None:
grch38 = not grch37
elif grch37 is None:
grch37 = not grch38
# parse condition IDs
condition_ids = None
if args.condition_ids:
condition_ids = []
for v in args.condition_ids:
if ":" not in v:
parser.error(f"condition-id must be 'db:id' format: {v}")
db, cid = v.split(":", 1)
condition_ids.append((db, cid))
# enrichment
enrich = False
if args.rcv_db is not None:
enrich = True
# run the actual class
with FilterQuery(args.vcv_db, args.rcv_db) as q:
result = q.query(
genes=args.genes,
condition_names=args.condition_names,
condition_identifiers=condition_ids,
classifications=args.classifications,
grch38=grch38,
grch37=grch37,
enrich=enrich,
output_path=args.output,
)
logger.info(f"Wrote {result} rows to {args.output}")
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
@_run_cli
def fetch_variant_main():
"""CLI entry point for FetchVariantQuery."""
parser = _fetch_variant_parser()
args = parser.parse_args()
# configure logging
configure_logging(args.verbose)
# resolve assembly flags: passing only one flag excludes the other
grch38, grch37 = args.grch38, args.grch37
if grch38 is None and grch37 is None:
grch38, grch37 = True, True
elif grch38 is None:
grch38 = not grch37
elif grch37 is None:
grch37 = not grch38
# parse location
location = None
if args.location:
parts = args.location.split(":")
if len(parts) == 2:
# chr:pos
location = GenomicLocation(parts[0], int(parts[1]))
elif len(parts) == 3:
# chr:pos:assembly
location = GenomicLocation(
parts[0], int(parts[1]), assembly=parts[2]
)
elif len(parts) == 4:
# chr:pos:ref:alt
location = GenomicLocation(
parts[0], int(parts[1]), parts[2], parts[3]
)
elif len(parts) == 5:
# chr:pos:ref:alt:assembly
location = GenomicLocation(
parts[0], int(parts[1]), parts[2], parts[3], parts[4]
)
else:
parser.error(
"location must be chr:pos, chr:pos:assembly, "
"chr:pos:ref:alt, or chr:pos:ref:alt:assembly"
)
# enrichment
enrich = False
if args.rcv_db is not None:
enrich = True
with FetchVariantQuery(args.vcv_db, args.rcv_db) as q:
result = q.query(
vcv_accessions=args.vcv,
variation_identifiers=args.variation_ids,
variant_names=args.names,
variant_name_pattern=args.name_pattern,
location=location,
spdi=args.spdi,
grch38=grch38,
grch37=grch37,
enrich=enrich,
output_path=args.output,
)
logger.info(f"Wrote {result} rows to {args.output}")
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
@_run_cli
def fetch_range_main():
"""CLI entry point for FetchRangeQuery."""
parser = _fetch_range_parser()
args = parser.parse_args()
# configure logging
configure_logging(args.verbose)
# resolve assembly flags: passing only one flag excludes the other
grch38, grch37 = args.grch38, args.grch37
if grch38 is None and grch37 is None:
grch38, grch37 = True, True
elif grch38 is None:
grch38 = not grch37
elif grch37 is None:
grch37 = not grch38
# parse ranges
ranges = []
for r in args.range_specs:
if ":" not in r or "-" not in r:
parser.error(f"range must be 'chr:start-end' format: {r}")
chrom, rest = r.split(":", 1)
if ":" in rest:
coords, assembly = rest.rsplit(":", 1)
else:
coords = rest
assembly = DEFAULT_ASSEMBLY
start, end = coords.split("-", 1)
ranges.append(GenomicRange(chrom, int(start), int(end), assembly))
# enrichment
enrich = False
if args.rcv_db is not None:
enrich = True
with FetchRangeQuery(args.vcv_db, args.rcv_db) as q:
result = q.query(
ranges=ranges,
genes=args.genes,
classifications=args.classifications,
grch38=grch38,
grch37=grch37,
enrich=enrich,
output_path=args.output,
)
logger.info(f"Wrote {result} rows to {args.output}")
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
@_run_cli
def list_conditions_main():
"""CLI entry point for ListConditionsQuery."""
parser = _list_conditions_parser()
args = parser.parse_args()
# configure logging
configure_logging(args.verbose)
with ListConditionsQuery(args.rcv_db) as q:
if args.list_sources:
result = q.available_sources()
# list_sources returns DataFrame, write manually
with _atomic_write(args.output) as tmp:
result.to_csv(tmp, sep="\t", index=False)
logger.info(f"Wrote {len(result)} sources to {args.output}")
else:
result = q.query(
source=args.source,
include_frequency=args.include_frequency,
min_frequency=args.min_frequency,
output_path=args.output,
)
logger.info(f"Wrote {result} rows to {args.output}")