"""
varvis_connector VarvisClient class implementation module
Copyright (C) 2026 Labor Berlin – Charité Vivantes GmbH
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, version 3 of the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
:author: Markus Konrad <markus.konrad@laborberlin.com>
"""
from collections import Counter
import logging
import os
import random
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor, Future, as_completed
from contextlib import ExitStack
from dataclasses import dataclass, field
from json import JSONDecodeError
from pathlib import Path
from fnmatch import fnmatchcase
from urllib.parse import quote, urlsplit
from typing import Type, TypeVar, Callable, Any, BinaryIO, overload, Literal
from collections.abc import Collection, Iterator, Sequence
import requests
from pydantic import BaseModel, TypeAdapter, ValidationError
from requests import Response, Session
from requests.adapters import HTTPAdapter
from tqdm import tqdm
from urllib3.util.retry import Retry
from ._log import default_logger
from .errors import VarvisError, VarvisTimeoutError, VarvisTransportError
from ._snv_stream_parser import stream_snv_json_from_request_or_buffer, resolve_header_indices
from .models import (
SnvAnnotationData,
SnvAnnotationHeaderItem,
CnvTargetResults,
PendingCnvData,
QCCaseMetricsData,
CoverageData,
AnalysisItem,
CaseReport,
PersonReportItem,
FindByInputFileNameAnalysisItem,
AnalysisFileDownloadLinks,
PersonUpdateData,
PersonData,
VirtualPanelSummary,
VarvisGene,
VirtualPanelUpdateData,
VirtualPanelData,
)
DEFAULT_HTTP_ERROR_MESSAGES = {
400: "Bad request.",
404: "Not found.",
405: "Method not allowed.",
406: "Not acceptable.",
422: "Unprocessable entity.",
}
TModel = TypeVar("TModel", bound=BaseModel)
def _url_param(param: str) -> str:
"""Apply URL quoting to a URL parameter string."""
return quote(param, safe="")
def _validate_download_filename(file_name: str) -> str:
"""Validate a download filename for safe use as a single path component."""
# Windows is strange: certain file names were treated as devices in MS-DOS and now are still reserved names
device_names = {"con", "prn", "aux", "nul"}
device_names.update(f"com{number}" for number in range(1, 10))
device_names.update(f"lpt{number}" for number in range(1, 10))
device_names.update(f"{device}{digit}" for device in ("com", "lpt") for digit in "¹²³")
normalized_name = file_name.strip().rstrip(".")
device_name = normalized_name.split(".", maxsplit=1)[0].lower()
if (
not normalized_name
or file_name != normalized_name
or normalized_name in {".", ".."}
or any(char in file_name for char in ("\0", "/", "\\", ":"))
or device_name in device_names
):
raise ValueError(f"Invalid download filename: {file_name}")
return file_name
def _jsondata_from_response(resp: Response, data_from_key: str) -> list | dict | str | float | int | bool | None:
"""Convert API response to JSON and extract data from the specified key. Handle reported errors."""
jsondata = resp.json()
if not jsondata.get("success", False):
_raise_varvis_error(jsondata, "Varvis API request did not succeed.")
try:
extracted = jsondata[data_from_key]
if not isinstance(extracted, list | dict | str | float | int | bool | None):
raise VarvisError(f'Response data "{data_from_key}" has unexpected type: {type(extracted)}')
return extracted
except KeyError as e:
raise VarvisError(f"Response does not contain expected data key '{data_from_key}'") from e
def _parse_response_for_model(model_class: Type[TModel], resp: Response, data_from_key: str | None = None) -> TModel:
"""Parse response data using the specified model class. Handle data validation errors."""
data: str | dict
if data_from_key:
# some endpoints return the actual payload in a separate "response" key ...
result = _jsondata_from_response(resp, data_from_key)
if result is None:
raise VarvisError("Response data is None")
if not isinstance(result, dict):
raise VarvisError("Response data should be a dictionary")
cls_validation_method = getattr(model_class, "model_validate")
data = result
else:
# ... while others return the payload directly in the response body
data = resp.text
cls_validation_method = getattr(model_class, "model_validate_json")
try:
validated_model = cls_validation_method(data)
return validated_model
except ValidationError as e:
raise VarvisError(f"Response validation failed: {e}") from e
def _parse_response_for_model_list(
model_class: Type[TModel], resp: Response, data_from_key: str | None = None
) -> list[TModel]:
"""Validate a response list against a model class selected at runtime."""
if data_from_key:
# some endpoints return the actual payload in a separate "response" key ...
data = _jsondata_from_response(resp, data_from_key)
else:
# ... while others return the payload directly in the response body
data = resp.json()
if not isinstance(data, list):
raise VarvisError("Response data should be a list")
# construct the schema from the runtime model class
model_list_adapter = TypeAdapter[list[TModel]](list[model_class]) # pyright: ignore[reportInvalidTypeForm]
try:
return model_list_adapter.validate_python(data)
except ValidationError as e:
raise VarvisError(f"Response validation failed: {e}") from e
def _parse_response_for_primitive(
resp: Response, data_from_key: str, convert_result: Callable | None = None
) -> None | bool | int | float | str:
"""Parse response data as a primitive type (e.g. int, str, float). Handle data conversion errors."""
data = _jsondata_from_response(resp, data_from_key)
if convert_result:
try:
data = convert_result(data)
except ValueError as e:
raise VarvisError(f"Response conversion failed: {e}") from e
if not isinstance(data, None | bool | int | float | str):
raise VarvisError(f"Parsed response data item has unexpected type: {type(data)}")
return data
def _raise_varvis_error(error_data: dict[str, str | None], custom_error_msg: str = "") -> None:
"""
Helper function to raise a VarvisError with am optional custom error message and all information provided by
the API.
:param error_data: A dictionary containing the error data.
:param custom_error_msg: An optional custom error message.
"""
custom_error_msg = custom_error_msg + "\n" if custom_error_msg else ""
error_msg_id = error_data.get("errorMessageId", "<NOT GIVEN>")
error_expected = str(error_data.get("errorExpected", "<NOT GIVEN>"))
error_id = error_data.get("errorId", "<NOT GIVEN>")
additional_info = error_data.get("additionalInformation", "<NOT GIVEN>")
raise VarvisError(
f"{custom_error_msg}"
f"Error message ID: {error_msg_id}\n"
f"Error expected: {error_expected}\n"
f"Error ID: {error_id}\n"
f"Additional information: {additional_info}"
)
class _DownloadRetry(Retry):
"""Apply exponential backoff with full jitter before each download retry."""
def get_backoff_time(self) -> float:
"""Return a random delay bounded by the exponential wait for this retry."""
# the first retry uses the base factor, followed by a doubled delay on each retry
if not self.history:
return 0.0
max_wait = min(self.backoff_max, self.backoff_factor * 2 ** (len(self.history) - 1))
# non-cryptographic randomness is sufficient for spreading download retries
return random.uniform(0, max_wait) # nosec B311
@dataclass
class _SessionState:
"""Associate an authenticated session with its logout generation and recovery outcome."""
session: requests.Session
logout_generation: int
recovery: Future["_SessionState"] = field(default_factory=Future)
[docs]
@dataclass
class VarvisClient:
"""
Handles interactions with the Varvis API, including authentication and request
management.
The Varvis class facilitates login and logout operations as well as handling
authenticated communication with the API, using methods to manage session
state and API requests.
:ivar api_url: The base URL for the Varvis API.
:ivar username: The username used for Varvis authentication.
:ivar password: The password used for Varvis authentication.
:ivar https_proxy: Optional HTTPS proxy URL to use for API requests.
:ivar ssl_verify: Whether to verify SSL certificates for API requests.
:ivar connection_timeout: Legacy timeout in seconds for both connection and response reads.
:ivar connect_timeout: Optional connection timeout in seconds.
:ivar read_timeout: Optional response-read timeout in seconds.
:ivar backoff_factor_seconds: Base factor for API request backoff with full jitter and deterministic session
recovery backoff.
:ivar backoff_max_tries: Maximum attempts per API request and per shared session recovery.
:ivar download_chunk_size: Number of bytes read per chunk when downloading files.
:ivar download_pool_size: Number of cached download connection pools. Zero uses the parallel worker count.
:ivar download_pool_max_size: Maximum connections retained per download pool. Zero uses the parallel worker count.
:ivar download_max_retries: Maximum HTTP adapter retries per download request.
:ivar download_backoff_factor_seconds: Base factor for download retry backoff with full jitter.
:ivar logger: Logger instance for recording operations and errors.
:ivar _session: The HTTP session used for Varvis API communication,
initialized during login and cleared during logout.
:ivar _session_lock: The reentrant lock protecting authentication state and login operations.
:ivar _recovery_lock: The lock for joining recovery without waiting for login I/O.
:ivar _loggedin_csrf: The CSRF token used for API requests after login.
:ivar _session_state: The authenticated session and its shared recovery outcome.
:ivar _logout_generation: The counter invalidating existing requests on explicit logout.
:ivar _active_recoveries: The recovery owners whose waiters must be released on logout.
:ivar request_timeout: The read-only HTTP timeout, using phase-specific values when configured.
:ivar logged_in: The read-only authentication status, read under the session lock.
"""
api_url: str = field(metadata={"help": "Varvis API URL", "envvar": "VARVIS_URL"})
username: str = field(metadata={"help": "Varvis credentials username", "envvar": "VARVIS_USER"})
password: str = field(
metadata={
"help": "Varvis credentials password. WARNING: Passing this option may expose the password in process listings and shell history; prefer VARVIS_PASSWORD or the interactive prompt"
}
)
https_proxy: str | None = field(
default=None,
metadata={"help": "Optional HTTPS proxy URL", "envvar": "HTTPS_PROXY"},
)
ssl_verify: bool = field(
default=True,
metadata={
"help": "Disable SSL certificate verification for API requests",
"argname": "disable-ssl-verify",
},
)
connection_timeout: float = field(
default=10.0,
metadata={"help": "Timeout in seconds for API requests when no phase-specific timeout is set"},
)
connect_timeout: float | None = field(
default=None,
metadata={"help": "Optional connection timeout in seconds"},
)
read_timeout: float | None = field(
default=None,
metadata={"help": "Optional response-read timeout in seconds"},
)
backoff_factor_seconds: float = field(
default=0.5,
metadata={"help": "Base factor for exponential backoff between retries"},
)
backoff_max_tries: int = field(
default=5,
metadata={"help": "Maximum attempts per API request and per shared session recovery. Must be at least 1"},
)
download_chunk_size: int = field(
default=5 * 1024 * 1024,
metadata={"help": "Chunk size in bytes for file downloads. Default is 5MB"},
)
download_pool_size: int = field(
default=0,
metadata={"help": "Number of cached download connection pools. 0 uses the parallel download worker count"},
)
download_pool_max_size: int = field(
default=0,
metadata={"help": "Maximum connections retained per download pool. 0 uses the parallel download worker count"},
)
download_max_retries: int = field(
default=2,
metadata={"help": "Maximum HTTP adapter retries per download request. Default is 2"},
)
download_backoff_factor_seconds: float = field(
default=0.5,
metadata={"help": "Base factor in seconds for exponential download retry backoff with full jitter"},
)
logger: logging.Logger = logging.Logger("not_initialized", -1)
_session: requests.Session | None = None
_session_lock: threading.RLock = field(default_factory=threading.RLock, init=False, repr=False, compare=False)
_recovery_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False, compare=False)
_loggedin_csrf: str | None = None
_session_state: _SessionState | None = field(default=None, init=False, repr=False, compare=False)
_logout_generation: int = field(default=0, init=False, repr=False, compare=False)
_active_recoveries: list[_SessionState] = field(default_factory=list, init=False, repr=False, compare=False)
def __post_init__(self) -> None:
"""Normalize client settings and validate backoff and timeout limits."""
if not self.api_url:
raise ValueError("API URL must be provided")
if not self.username:
raise ValueError("Username must be provided")
if not self.password:
raise ValueError("Password must be provided")
if self.logger.level == -1:
self.logger = default_logger()
if not self.api_url.endswith("/"):
self.api_url += "/"
self.logger.info('VarvisClient initialized for API URL "%s"', self.api_url)
if self.https_proxy:
self.logger.info('Using HTTPS proxy "%s"', self.https_proxy)
if not self.ssl_verify:
self.logger.warning("SSL verification disabled")
if self.backoff_factor_seconds < 0:
raise ValueError("backoff_factor_seconds must be a non-negative number")
if self.download_backoff_factor_seconds < 0:
raise ValueError("download_backoff_factor_seconds must be a non-negative number")
if self.backoff_max_tries < 1:
raise ValueError("backoff_max_tries must be a strictly positive integer")
# validate all configured request timeouts
for timeout_name, timeout in {
"connection_timeout": self.connection_timeout,
"connect_timeout": self.connect_timeout,
"read_timeout": self.read_timeout,
}.items():
if timeout is not None and timeout <= 0:
raise ValueError(f"{timeout_name} must be a strictly positive number")
def __repr__(self) -> str:
return (
f"VarvisClient(api_url={self.api_url}, "
f"username={self.username}, "
f"password=***, "
f"https_proxy={self.https_proxy}, "
f"ssl_verify={self.ssl_verify}, "
f"connection_timeout={self.connection_timeout}, "
f"connect_timeout={self.connect_timeout}, "
f"read_timeout={self.read_timeout}, "
f"backoff_factor_seconds={self.backoff_factor_seconds}, "
f"backoff_max_tries={self.backoff_max_tries})"
)
def __str__(self) -> str:
return self.__repr__()
[docs]
@classmethod
def from_env(cls) -> "VarvisClient":
"""Create a client using authentication, timeout, retry, and download settings from the environment."""
# read credentials and convert optional environment settings to their constructor types
url, username, password = [os.getenv("VARVIS_" + s, "") for s in ("URL", "USER", "PASSWORD")]
kwargs = {}
for key in {
"https_proxy",
"ssl_verify",
"connection_timeout",
"connect_timeout",
"read_timeout",
"backoff_factor_seconds",
"download_backoff_factor_seconds",
"backoff_max_tries",
"download_chunk_size",
"download_pool_size",
"download_pool_max_size",
"download_max_retries",
}:
prefix = "VARVIS_" if key != "https_proxy" else ""
value: Any = os.getenv(prefix + key.upper())
if value is not None:
if key in {
"connection_timeout",
"connect_timeout",
"read_timeout",
"backoff_factor_seconds",
"download_backoff_factor_seconds",
}:
value = float(value)
elif key in {
"backoff_max_tries",
"download_chunk_size",
"download_pool_size",
"download_pool_max_size",
"download_max_retries",
}:
value = int(value)
elif key == "ssl_verify":
value = value.lower() in {"true", "1"} if value else True
kwargs[key] = value
return cls(url, username, password, **kwargs)
@property
def request_timeout(self) -> float | tuple[float, float]:
"""Return the Requests timeout using phase-specific values when configured."""
if self.connect_timeout is None and self.read_timeout is None:
return self.connection_timeout
# fill an unspecified phase with the legacy timeout
return (
self.connect_timeout if self.connect_timeout is not None else self.connection_timeout,
self.read_timeout if self.read_timeout is not None else self.connection_timeout,
)
@property
def logged_in(self) -> bool:
"""Read local session and CSRF status under the session lock without contacting the server."""
with self._session_lock:
return self._session is not None and self._loggedin_csrf is not None
[docs]
def login(self, raise_for_status: bool = True) -> bool:
"""
Performs the login operation for the user if not already logged in. It initializes
a session, retrieves the necessary cookies and CSRF token, and submits the user's
credentials to authenticate. After a successful login, the session is updated
with the new cookies and CSRF token required for subsequent authenticated requests.
Login is serialized with session acquisition and invalidation. Failed login attempts
clear the partially initialized session before releasing the lock. Successful login
creates a shared recovery outcome for requests using the authenticated session.
This method raises an error if the CSRF token is not found in the initial response
or if the login fails due to missing CSRF token in the login response.
:param raise_for_status: If true, raise an error if the login fails due to HTTP errors.
:raises VarvisError: If the login failed.
:raises HTTPError: Indicates a failure during the HTTP communication with the varvis API.
:return: True if the login was successful, False otherwise.
"""
with self._session_lock:
if self.logged_in:
self.logger.info("Already logged in -- not performing a new login")
return True
self.logger.info("Logging in")
# clear incomplete authentication before releasing the lock
try:
# set initial session
self._session = requests.Session()
self._session.verify = self.ssl_verify
if self.https_proxy:
self._session.proxies = {"https": self.https_proxy}
# get initial session cookie and CSRF code
self.logger.debug("Acquiring initial session")
initial_resp = self._session.head(self.api_url, stream=True, timeout=self.request_timeout)
if raise_for_status:
initial_resp.raise_for_status()
elif not initial_resp.ok:
self.logger.error("Could not acquire initial session -- aborting login")
return False
initial_cookies = initial_resp.cookies
initial_csrf = initial_resp.headers.get("X-CSRF-TOKEN", None)
if not initial_csrf:
if raise_for_status:
raise VarvisError("CSRF token not found in initial response")
else:
self.logger.error("CSRF token not found in initial response")
return False
self.logger.debug("Initial session OK")
# log in using the provided credentials, CSRF code and initial cookie
login_data = {
"username": self.username,
"password": self.password,
"_csrf": initial_csrf,
}
self.logger.debug("Acquiring login session")
login_resp = self._send_request(
"POST",
"login",
allow_retries=False,
raise_for_status=raise_for_status,
cookies=initial_cookies,
data=login_data,
)
# get the session cookie and CSRF code that can be used for the rest of the session
loggedin_cookies = login_resp.cookies
loggedin_csrf = login_resp.headers.get("X-CSRF-TOKEN", None)
if not loggedin_csrf:
# it seems the varvis API responds with HTTP 200 OK in case the login fails, but doesn't return a CSRF in that case
if raise_for_status:
raise VarvisError("Login failed (CSRF token not found in login response)")
else:
self.logger.error("Login failed (CSRF token not found in login response)")
return False
self.logger.debug("Login session OK")
if api_build_version := login_resp.headers.get("BUILD-VERSION", None):
self.logger.debug("Varvis API build version: %s", api_build_version)
# set cookies and CSRF for further requests
self._session.cookies.update(loggedin_cookies)
self._session.headers.update({"X-CSRF-TOKEN": loggedin_csrf})
self._loggedin_csrf = loggedin_csrf
# publish the authenticated record without blocking new recovery participants on login I/O
with self._recovery_lock:
self._session_state = _SessionState(self._session, self._logout_generation)
self.logger.info("Login successful")
finally:
if not self.logged_in:
self._reset_state_on_logout(clear_recovery=False)
return True
[docs]
def logout(self) -> None:
"""
Logs out the user by ending the current session and resetting authentication credentials.
This method ensures that the user is logged out by sending a logout request to the
appropriate endpoint. It clears the session data and invalidates the CSRF token.
Existing recovery work is invalidated even if authentication is already cleared.
Waiting workers receive a VarvisError, and recovery cannot log in again after
logout. A login already in progress finishes before logout acquires the lock.
In-flight HTTP requests are not interrupted. A later explicit login is allowed.
:raises HTTPError: Indicates a failure during the logout HTTP request.
"""
# invalidate old workers and wake recovery waiters before clearing authentication
with self._session_lock:
with self._recovery_lock:
self._logout_generation += 1
for state in [*self._active_recoveries, self._session_state]:
if state is not None and not state.recovery.done():
state.recovery.set_exception(VarvisError("Session recovery cancelled by logout"))
try:
if not self.logged_in:
self.logger.info("Not logged in -- not performing a logout")
return
# send logout while preventing another login from replacing the session
self.logger.info("Logging out")
logout_data = {"_csrf": self._loggedin_csrf}
self._send_request("POST", "logout", allow_retries=False, data=logout_data)
self.logger.info("Logout successful")
finally:
self._reset_state_on_logout()
[docs]
def get_snv_annotations(self, analysis_id: int) -> SnvAnnotationData:
"""
Retrieves the SNV (Single Nucleotide Variant) annotations for a given analysis
based on the specified analysis ID. This method sends an HTTP GET request
to the appropriate endpoint of the API to fetch the annotation data associated
with the analysis.
:param analysis_id: The unique identifier of the analysis for which SNV
annotations are being retrieved.
:return: An object of `SnvAnnotationData` containing the SNV annotations
for the specified analysis.
"""
self.logger.info("Getting SNV annotations for analysis %d", analysis_id)
resp = self._send_request(
"GET",
f"analysis/{analysis_id}/annotations",
handle_http_errors={400: "Analysis not found for the given ID."},
)
return _parse_response_for_model(SnvAnnotationData, resp)
@overload
def iter_snv_annotations(
self,
analysis_id: int,
*,
as_dict: Literal[False] = False,
header: Sequence[Any] | None = None,
target_genes: Collection[str] | None = None,
target_coordinates: Collection[tuple[str, int]] | None = None,
chunk_size: int = 65536,
allow_buffering: bool = False,
header_callback: Callable[[list[SnvAnnotationHeaderItem]], None] | None = None,
stream_to_tempfile: bool = False,
) -> Iterator[list[Any]]: ...
@overload
def iter_snv_annotations(
self,
analysis_id: int,
*,
as_dict: Literal[True],
header: Sequence[Any] | None = None,
target_genes: Collection[str] | None = None,
target_coordinates: Collection[tuple[str, int]] | None = None,
chunk_size: int = 65536,
allow_buffering: bool = False,
header_callback: Callable[[list[SnvAnnotationHeaderItem]], None] | None = None,
stream_to_tempfile: bool = False,
) -> Iterator[dict[str, Any]]: ...
@overload
def iter_snv_annotations(
self,
analysis_id: int,
*,
as_dict: bool = False,
header: Sequence[Any] | None = None,
target_genes: Collection[str] | None = None,
target_coordinates: Collection[tuple[str, int]] | None = None,
chunk_size: int = 65536,
allow_buffering: bool = False,
header_callback: Callable[[list[SnvAnnotationHeaderItem]], None] | None = None,
stream_to_tempfile: bool = False,
) -> Iterator[list[Any]] | Iterator[dict[str, Any]]: ...
[docs]
def iter_snv_annotations(
self,
analysis_id: int,
*,
as_dict: bool = False,
header: Sequence[Any] | None = None,
target_genes: Collection[str] | None = None,
target_coordinates: Collection[tuple[str, int]] | None = None,
chunk_size: int = 65536,
allow_buffering: bool = False,
header_callback: Callable[[list[SnvAnnotationHeaderItem]], None] | None = None,
stream_to_tempfile: bool = False,
) -> Iterator[list[Any]] | Iterator[dict[str, Any]]:
"""
Streams SNV annotations row-by-row for a given analysis to prevent memory exhaustion.
Streams the HTTP response directly by default; temporary-file mode caches the full response first.
Because the Varvis API serializes variant data before column headers, yielding dictionaries
or filtering by coordinates/genes requires either an explicit header sequence or setting
`allow_buffering=True` or `stream_to_tempfile=True`. Resources are closed on completion,
failure, or explicit generator closure. Close the generator explicitly when stopping early.
Temporary-file mode makes one HTTP request and completes the transfer before yielding
rows, scanning the cache for a header if none was supplied. It avoids row buffering
even when `allow_buffering=True`.
Recommended parameter combinations:
- For raw rows without filters, leave both `allow_buffering=False` and
`stream_to_tempfile=False`. Rows arrive during the transfer with low memory overhead
and no temporary-file I/O.
- For dictionaries or filters with a known `header`, also leave both flags False.
This streams immediately with low memory overhead. Fetching the header separately
with `get_snv_annotation_header()` requires an additional request and can transfer
nearly the entire payload before finding the trailing header.
- For dictionaries or filters without a known header on large analyses, use
`stream_to_tempfile=True` and `allow_buffering=False`. This uses one response and
keeps row memory bounded, but requires storage for the full response, extra file
reads and writes, and completion of the transfer and header scan before the first row.
- For smaller responses that fit comfortably in RAM, use `allow_buffering=True` and
`stream_to_tempfile=False` without a header. This avoids temporary-file I/O and uses
one response, but retains every row in memory and delays all output until the header
is parsed. Filtering does not reduce the memory needed for these buffered rows.
Setting both flags True is supported but provides no advantage over temporary-file
mode alone: the cached header takes precedence and no row buffering occurs. Likewise,
`allow_buffering=True` has no effect when `header` is supplied. Temporary-file mode
still caches the full response with a supplied header or raw-row output, so direct
streaming is usually preferable in those cases.
Temporary files prefer $XDG_CACHE_HOME/varvis-connector, defaulting to
~/.cache/varvis-connector, because user caches are usually disk-backed while system
temporary directories may use RAM. If cache creation fails, the system temporary
directory is used with a warning; filesystem backing is not checked. An unset, empty,
or relative XDG_CACHE_HOME selects ~/.cache. Cache write failures propagate without fallback.
:param analysis_id: The unique identifier of the analysis.
:param as_dict: If True, yields dictionaries keyed by column ID, then title, then col_<index>.
:param header: Optional column descriptors or names, taking precedence over the response header.
:param target_genes: Optional case-insensitive gene symbols to retain using local filtering.
Array-valued gene IDs are resolved through the Gene column's dataDictionary.
:param target_coordinates: Optional (chromosome, 1-based position) tuples to retain using local
filtering. Chromosomes may include a chr prefix. Coordinate cells may be scalars or
arrays whose first value is the chromosome or position. Both filters must match when supplied together.
:param chunk_size: Read chunk size in bytes for the HTTP stream and cached rows. Defaults to 64 KiB.
:param allow_buffering: If True without a supplied or cached header, buffers all raw rows in memory
until the trailing header is parsed, then yields filtered lists or dictionaries. Avoids
temporary-file I/O at the cost of memory proportional to the complete row set and delayed
output. Prefer this for small responses; it has no effect with a supplied header or
`stream_to_tempfile=True`.
:param header_callback: Optional callback invoked with the response header when the row-parsing pass
reaches it. Early generator closure may prevent invocation, including in temporary-file mode.
:param stream_to_tempfile: If True, caches the response in a temporary file and reads its header
if none was supplied before streaming rows from the same file. Prefer this with
`allow_buffering=False` for large responses requiring dictionaries or filters without a known
header. Bounds row memory usage but requires full-response storage and file I/O, and delays
output until the transfer and any required header scan finish.
:return: An iterator yielding individual variant rows as lists or dictionaries.
:raises ValueError: If dictionaries or filters require a header without buffering or temporary-file caching.
:raises VarvisError: If the Varvis API responds with an error or the payload terminates prematurely.
:raises OSError: If temporary-file creation fails in both locations or cache I/O fails.
"""
# validate arguments before acquiring resources
if as_dict and header is None and not allow_buffering and not stream_to_tempfile:
raise ValueError(
"as_dict=True requires 'header' to stream with zero buffering because 'data' precedes 'header' "
"in the Varvis API response. Provide 'header', use get_snv_annotation_header() first, or pass "
"allow_buffering=True."
)
if (target_genes or target_coordinates) and header is None and not allow_buffering and not stream_to_tempfile:
raise ValueError(
"Filtering by target_genes or target_coordinates requires 'header' to filter during streaming "
"with zero buffering. Provide 'header', use get_snv_annotation_header() first, or pass "
"allow_buffering=True."
)
self.logger.info("Streaming SNV annotations for analysis %d (as_dict=%s)", analysis_id, as_dict)
# close the response and optional cache on failure, completion, or generator close
with ExitStack() as stack:
resp = stack.enter_context(
self._send_request(
"GET",
f"analysis/{analysis_id}/annotations",
handle_http_errors={400: "Analysis not found for the given ID."},
stream=True,
)
)
input_stream: requests.Response | BinaryIO = resp
# cache the response and resolve its header before streaming rows
if stream_to_tempfile:
# prefer the user cache to avoid keeping large responses on ram-backed system temp storage
try:
cache_home = os.getenv("XDG_CACHE_HOME")
cache_dir = Path(cache_home) if cache_home else Path.home() / ".cache"
if not cache_dir.is_absolute():
cache_dir = Path.home() / ".cache"
cache_dir /= "varvis-connector"
cache_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
tempcache = stack.enter_context(tempfile.TemporaryFile(dir=cache_dir))
except (OSError, RuntimeError) as exc:
self.logger.warning(
"Could not create an SNV temporary file in the user cache: %s. "
"Falling back to the system temporary directory, which may be RAM-backed.",
exc,
)
tempcache = stack.enter_context(tempfile.TemporaryFile())
for chunk in resp.iter_content(chunk_size=chunk_size):
tempcache.write(chunk)
tempcache.seek(0)
if header is None:
header = self._get_snv_annotation_header_from_resp_or_filebuffer(tempcache)
tempcache.seek(0)
input_stream = tempcache
# yield rows while the required resources remain open
yield from self._stream_snv_annotations_generator(
resp_or_filebuffer=input_stream,
as_dict=as_dict,
header=header,
target_genes=target_genes,
target_coordinates=target_coordinates,
chunk_size=chunk_size,
allow_buffering=allow_buffering,
header_callback=header_callback,
stream_to_tempfile=stream_to_tempfile,
)
def _stream_snv_annotations_generator(
self,
resp_or_filebuffer: requests.Response | BinaryIO,
as_dict: bool,
header: Sequence[Any] | None,
target_genes: Collection[str] | None,
target_coordinates: Collection[tuple[str, int]] | None,
chunk_size: int,
allow_buffering: bool,
header_callback: Callable[[list[SnvAnnotationHeaderItem]], None] | None,
stream_to_tempfile: bool,
) -> Iterator[Any]:
"""
Internal generator yielding filtered SNV annotations incrementally from the response stream.
Handles upfront or buffered header resolution, applies case-insensitive gene symbol
and genomic coordinate filtering, and formats rows as raw lists or dictionaries.
Array-valued gene IDs are translated through the header dictionary; coordinate arrays
use their first value. Scalar annotations remain supported without changing yielded rows.
:param resp_or_filebuffer: Active streaming HTTP response or readable binary file.
:param as_dict: Whether to emit dictionary records mapped to column names.
:param header: Optional upfront header descriptors.
:param target_genes: Optional set of gene symbols to retain.
:param target_coordinates: Optional set of (chrom, pos) tuples to retain.
:param chunk_size: HTTP or binary-file read chunk size in bytes.
:param allow_buffering: Whether to buffer rows when header is trailing.
:param header_callback: Optional callback invoked when header is parsed.
:param stream_to_tempfile: Whether the caller cached the response and resolved its header before iteration.
:return: Generator yielding lists or dictionaries.
:raises ValueError: If target filters cannot resolve necessary column indices from header.
"""
# resolve filters and column keys if header is provided up front
genes_filter = {g.strip().upper() for g in target_genes} if target_genes else None
coords_filter = (
{(str(c[0]).strip().lower().removeprefix("chr"), int(c[1])) for c in target_coordinates}
if target_coordinates
else None
)
col_keys: list[str] | None = None
gene_idx: int | None = None
chr_idx: int | None = None
pos_idx: int | None = None
# translate dictionary-backed gene IDs after either upfront or buffered header discovery
def get_gene_symbols(columns: Sequence[Any], index: int | None) -> dict[str, str]:
"""Resolve gene IDs to normalized symbols from model or dictionary headers."""
if index is None:
return {}
item = columns[index]
dictionary = item.get("dataDictionary") if isinstance(item, dict) else getattr(item, "dataDictionary", None)
return {str(key): str(value).strip().upper() for key, value in (dictionary or {}).items()}
gene_symbols: dict[str, str] = {}
if header is not None:
col_keys = []
for i, item in enumerate(header):
if isinstance(item, SnvAnnotationHeaderItem):
col_keys.append(item.id or item.title or f"col_{i}")
elif isinstance(item, dict):
col_id = item.get("id") or item.get("title")
col_keys.append(str(col_id) if col_id is not None else f"col_{i}")
else:
col_keys.append(str(item))
indices = resolve_header_indices(header)
gene_idx = indices.get("gene")
chr_idx = indices.get("chr")
pos_idx = indices.get("pos")
gene_symbols = get_gene_symbols(header, gene_idx)
# validate that required filter columns exist in provided header
if genes_filter is not None and gene_idx is None:
raise ValueError("Could not resolve 'gene' column index from header for target_genes filter.")
if coords_filter is not None and (chr_idx is None or pos_idx is None):
raise ValueError(
"Could not resolve chromosome ('chr') or position ('pos') column index from header for target_coordinates filter."
)
# match scalar annotations and the array-valued cells returned by the playground
def matches_filters(row: list[Any], g_idx: int | None, c_idx: int | None, p_idx: int | None) -> bool:
"""Match normalized gene symbols and coordinates while preserving original row values."""
if genes_filter is not None:
if g_idx is None or g_idx >= len(row):
return False
values = row[g_idx] if isinstance(row[g_idx], list) else [row[g_idx]]
if not any(
gene_symbols.get(str(value), str(value).strip().upper()) in genes_filter for value in values
):
return False
if coords_filter is not None:
if c_idx is None or p_idx is None or c_idx >= len(row) or p_idx >= len(row):
return False
chromosome = row[c_idx]
position = row[p_idx]
if isinstance(chromosome, list):
chromosome = chromosome[0] if chromosome else None
if isinstance(position, list):
position = position[0] if position else None
if position is None:
return False
row_chr = str(chromosome).strip().lower().removeprefix("chr")
try:
row_pos = int(position)
except (ValueError, TypeError):
return False
if (row_chr, row_pos) not in coords_filter:
return False
return True
metadata: dict[str, Any] = {}
# stream with zero buffering when header is known or buffering not requested
if header is not None or not allow_buffering:
for row in stream_snv_json_from_request_or_buffer(
resp_or_filebuffer, chunk_size=chunk_size, metadata_out=metadata
):
if matches_filters(row, gene_idx, chr_idx, pos_idx):
if as_dict and col_keys is not None:
yield dict(zip(col_keys, row))
else:
yield row
# notify header callback if provided
if header_callback is not None and "header" in metadata:
validated_header = [SnvAnnotationHeaderItem.model_validate(item) for item in metadata["header"]]
header_callback(validated_header)
else:
# buffer rows when header needs to be discovered at stream end
buffered_rows: list[list[Any]] = []
for row in stream_snv_json_from_request_or_buffer(
resp_or_filebuffer, chunk_size=chunk_size, metadata_out=metadata
):
buffered_rows.append(row)
raw_header = metadata.get("header") or []
validated_header = [SnvAnnotationHeaderItem.model_validate(item) for item in raw_header]
if header_callback is not None:
header_callback(validated_header)
col_keys = [item.id or item.title or f"col_{i}" for i, item in enumerate(validated_header)]
indices = resolve_header_indices(validated_header)
gene_idx = indices.get("gene")
chr_idx = indices.get("chr")
pos_idx = indices.get("pos")
gene_symbols = get_gene_symbols(validated_header, gene_idx)
# validate that required filter columns exist in discovered header
if genes_filter is not None and gene_idx is None:
raise ValueError(
"Could not resolve 'gene' column index from discovered header for target_genes filter."
)
if coords_filter is not None and (chr_idx is None or pos_idx is None):
raise ValueError(
"Could not resolve chromosome ('chr') or position ('pos') column index from discovered header for target_coordinates filter."
)
for row in buffered_rows:
if matches_filters(row, gene_idx, chr_idx, pos_idx):
if as_dict:
yield dict(zip(col_keys, row))
else:
yield row
def _get_snv_annotation_header_from_resp_or_filebuffer(
self, resp_or_filebuffer: requests.Response | BinaryIO
) -> list[SnvAnnotationHeaderItem]:
"""
Read and validate the annotation header from the current stream position.
Rows before the header are discarded. The caller owns closing the HTTP response or
binary file and rewinding a cached file before a subsequent row-parsing pass.
:param resp_or_filebuffer: Active HTTP response or readable binary file.
:return: Validated annotation column descriptors.
:raises VarvisError: If the header is missing or the payload terminates prematurely.
"""
metadata: dict[str, Any] = {}
# consume variant data rows until header metadata is parsed
for _ in stream_snv_json_from_request_or_buffer(resp_or_filebuffer, metadata_out=metadata):
if "header" in metadata:
break
raw_header = metadata.get("header")
if raw_header is None:
raise VarvisError("Header missing in annotations response")
return [SnvAnnotationHeaderItem.model_validate(item) for item in raw_header]
[docs]
def get_cnv_target_results(
self,
person_lims_id: str,
analysis_ids: int | list[int],
virtual_panel_id: int | None = 1,
) -> CnvTargetResults:
"""
Retrieves the Copy Number Variant (CNV) target results for a specified person LIMS-ID and associated analyses.
A virtual panel can also optionally be specified to filter the results.
This method communicates with an external API to request and retrieve the CNV target results. The analysis IDs
can be provided either as a single integer or a list of integers. Results are validated to ensure they align
with the expected data structure, and an appropriate model is used for parsing.
:param person_lims_id: The LIMS ID of the person for whom the CNV target results are being queried.
:param analysis_ids: A single analysis ID or a list of analysis IDs to query CNV results for.
:param virtual_panel_id: (Optional) The ID of the virtual panel to apply in filtering the CNV target data.
If 1 (the default), the "all genes" panel is used (i.e. no filtering). If None, the Varvis documentation
states that "the lastly selected virtual panel for the given person is used or 'All Genes' if no virtual
panel was selected yet.", i.e. if None the behavior depends on the selection stored in the current user's
session.
:return: An instance of CnvTargetResults containing the parsed CNV target results data.
:raises ValueError: If `analysis_ids` is neither an integer nor a non-empty list of integers.
:raises requests.HTTPError: For errors occurring during the API request if they are unrelated to 422 status codes.
:raises VarvisError: If the API response indicates an invalid analysis ID.
"""
analysis_ids = [analysis_ids] if isinstance(analysis_ids, int) else analysis_ids
if not analysis_ids:
raise ValueError("Parameter `analysis_ids` must be either an integer a non-empty list of integers")
self.logger.info(
"Getting CNV target results for person LIMS-ID %s, analysis IDs %s%s",
person_lims_id,
", ".join(map(str, analysis_ids)),
", " + str(virtual_panel_id) if virtual_panel_id else "",
)
query_params = "&".join(f"analysisIds={a_id}" for a_id in analysis_ids)
if virtual_panel_id:
query_params += f"&virtualPanelId={virtual_panel_id}"
resp = self._send_request(
"GET",
f"results/{_url_param(person_lims_id)}/cnv?{query_params}",
handle_http_errors={
404: "Person with given LIMS-ID was not found.",
422: "Analysis ID is not associated with a CNV analysis or is not valid for the specified person LIMS-ID.",
},
)
return _parse_response_for_model(CnvTargetResults, resp)
[docs]
def get_internal_person_id(self, person_lims_id: str) -> int:
"""
Retrieve the internal Varvis ID associated with a given person's LIMS-ID by sending a
request to the Varvis API. This method fetches the internal ID required for
further operations on the person record.
:param person_lims_id: The LIMS-ID of the person whose internal ID is to
be retrieved.
:return: The internal ID of the person associated with the given LIMS-ID.
:raises VarvisError: If the Varvis API responds with an error, fails to provide
the expected data, or returns invalid or missing internal IDs.
:raises requests.HTTPError: For HTTP errors other than 404.
"""
self.logger.info('Getting internal ID for person LIMS-ID "%s"', person_lims_id)
resp = self._send_request(
"GET",
f"person/{_url_param(person_lims_id)}/id",
handle_http_errors={404: "Person not found for the given LIMS-ID."},
)
pers_id = _parse_response_for_primitive(resp, "response", convert_result=int)
if not isinstance(pers_id, int):
raise VarvisError("Parsed response item should be an integer")
return pers_id
[docs]
def get_pending_cnv_segments(
self,
*,
person_id: int | None = None,
person_lims_id: str | None = None,
analysis_ids: int | list[int] | None,
virtual_panel_id: int | None = 1,
) -> PendingCnvData:
"""
Retrieves pending CNV segments based for a given person (identified either by internal ID or LIMS ID) and
associated analysis IDs.
:param person_id: Internal ID of the person. Must be an integer or None and must be given if ``person_lims_id`` is None.
:param person_lims_id: LIMS ID of the person. Must be a string or None and must be given if ``person_id`` is None.
:param analysis_ids: Analysis IDs associated with the person. Can be an integer or a list of integers.
:param virtual_panel_id: Optional virtual panel ID, if applicable, for filtering the CNV segments.
If 1 (the default), the "all genes" panel is used (i.e. no filtering). If None, the Varvis documentation
states that "the lastly selected virtual panel for the given person is used or 'All Genes' if no virtual
panel was selected yet.", i.e. if None the behavior depends on the selection stored in the current user's
session.
:return: An instance of `PendingCnvData` containing the pending CNV segment
information.
:raises ValueError: Raised if both `person_id` and `person_lims_id` are missing
or if both are provided. Also raised when `analysis_ids` is improperly formatted
or empty.
:raises VarvisError: Raised if the response from the Varvis API does not contain
the expected data.
"""
if person_id is None and person_lims_id is None:
raise ValueError("Either `person_id` or `person_lims_id` must be provided")
if person_id is not None and person_lims_id is not None:
raise ValueError("Only one of `person_id` or `person_lims_id` can be provided")
if person_lims_id is not None:
self.logger.info("LIMS-ID is given; will retrieve internal person ID first")
person_id = self.get_internal_person_id(person_lims_id)
analysis_ids = [analysis_ids] if isinstance(analysis_ids, int) else analysis_ids
if not analysis_ids:
raise ValueError("Parameter `analysis_ids` must be either an integer a non-empty list of integers")
self.logger.info(
"Getting pending CNV segments for person internal ID %d, analysis IDs %s%s",
person_id,
", ".join(map(str, analysis_ids)),
", " + str(virtual_panel_id) if virtual_panel_id else "",
)
# note: the CNV target results endpoint uses analysisIds (with "i") as the parameter name, this endpoint uses
# analysesIds (with "e") -- crazy varvis API!
query_params = f"personId={person_id}&" + "&".join(f"analysesIds={a_id}" for a_id in analysis_ids)
if virtual_panel_id:
query_params += f"&virtualPanelId={virtual_panel_id}"
resp = self._send_request(
"GET",
f"pending-cnv?{query_params}",
handle_http_errors={400: "Person or analysis ID not found."},
)
return _parse_response_for_model(PendingCnvData, resp, "response")
[docs]
def get_qc_case_metrics(self, person_lims_id: str) -> QCCaseMetricsData:
"""
Fetches quality control (QC) case metrics for a given person based on their LIMS ID.
This method interacts with a remote API to retrieve and process QC data, ensuring
that it is structured into a usable format. If no metric results match the provided
LIMS ID, an error will be raised.
:param person_lims_id: The LIMS ID of the person whose QC case metrics are to
be retrieved.
:return: An instance of QCCaseMetricsData containing parsed and validated QC case
metrics.
:raises VarvisError: If no metric results data for the provided LIMS ID is found
or if the response validation fails.
"""
self.logger.info('Getting QC case metrics for person LIMS ID "%s"', person_lims_id)
resp = self._send_request(
"GET",
f"qualitycontrol/metrics/case/{_url_param(person_lims_id)}",
handle_http_errors={400: "Person with given LIMS-ID was not found."},
)
data = _jsondata_from_response(resp, "response")
if not isinstance(data, dict):
raise VarvisError("Parsed response data should be a dictionary")
try:
metric_results = data["metricResults"].pop(person_lims_id)
except KeyError:
raise VarvisError(f"No metric results data for person with LIMS ID {person_lims_id} found in response")
data["metricResults"] = metric_results
try:
return QCCaseMetricsData.model_validate(data)
except ValidationError as e:
raise VarvisError(f"Response validation failed: {e}") from e
[docs]
def get_coverage_data(self, person_lims_id: str, virtual_panel_id: int | None = 1) -> list[CoverageData]:
"""
Retrieves coverage data for a specified person LIMS ID and optional virtual panel ID.
This function fetches coverage data based on the provided person LIMS ID. If a
virtual panel ID is specified, it filters the coverage data accordingly. The
response is further validated and processed into a list of `CoverageData`
objects.
:param person_lims_id: A string representing the person LIMS ID for whom the
coverage data is to be retrieved.
:param virtual_panel_id: An optional integer representing the virtual panel
ID to filter the coverage data, if applicable.
If 1 (the default), the "all genes" panel is used (i.e. no filtering). If None, the Varvis documentation
states that "the lastly selected virtual panel for the given person is used or 'All Genes' if no virtual
panel was selected yet.", i.e. if None the behavior depends on the selection stored in the current user's
session.
:return: A list of `CoverageData` objects containing validated coverage
information.
:raises VarvisError: If the validation of any coverage data item fails.
"""
self.logger.info(
'Getting coverage data for person LIMS ID "%s"%s',
person_lims_id,
f" (virtual panel ID {virtual_panel_id})" if virtual_panel_id else "",
)
if virtual_panel_id:
query_params = f"?virtualPanelId={virtual_panel_id}"
else:
query_params = ""
resp = self._send_request(
"GET",
f"{_url_param(person_lims_id)}/coverage{query_params}",
handle_http_errors={400: "Person with given LIMS-ID was not found."},
)
return _parse_response_for_model_list(CoverageData, resp)
[docs]
def get_analyses(self, analysis_ids: int | list[int] | None = None) -> list[AnalysisItem]:
"""
Retrieves a list of analysis items from the varvis API. Optionally allows to filter by analysis IDs.
This method sends a GET request to the "analyses" endpoint of the
server and parses the response into a list of `AnalysisItem` objects.
Raises exceptions on request failure or if the response format
is invalid.
:param analysis_ids: Optional list of analysis IDs to filter the results by. If provided,
only analyses with matching IDs will be returned.
:return: A list of `AnalysisItem` objects retrieved from the varvis API.
"""
self.logger.info("Getting analyses")
if isinstance(analysis_ids, int):
analysis_ids = [analysis_ids]
if analysis_ids:
query_params = f"?analysisIds={','.join(map(str, analysis_ids))}"
else:
query_params = ""
resp = self._send_request("GET", f"analyses{query_params}")
return _parse_response_for_model_list(AnalysisItem, resp, "response")
[docs]
def get_person(self, person_lims_id: str) -> PersonData:
"""
Fetches person data for the given LIMS-ID.
This method retrieves information associated with a person identified via
their specified LIMS-ID. It sends a GET request to fetch the relevant data,
and returns a structured person data model upon a successful response.
:param person_lims_id: The unique LIMS-ID identifying the person for whom
data is to be fetched.
:return: The detailed person data represented as a PersonData object.
"""
self.logger.info('Getting person data for LIMS-ID "%s"', person_lims_id)
resp = self._send_request(
"GET",
f"person/{_url_param(person_lims_id)}",
handle_http_errors={404: "Person with given LIMS-ID was not found."},
)
return _parse_response_for_model(PersonData, resp, data_from_key="response")
[docs]
def create_or_update_person(self, person_data: PersonUpdateData | dict[str, Any]) -> int:
"""
Allows to create a new person entry, or updates an existing one. Only the id field is required.
Fields that are null will not override existing values on update.
Updates or creates a person record in Varvis. If a dictionary is provided
for `person_data`, it will be validated against the `PersonUpdateData` schema. If validation
fails, an error will be raised to ensure proper input format.
:param person_data: The data required to create or update a person record. Can be
either an instance of `PersonUpdateData` or a dictionary that adheres to the
`PersonUpdateData` schema.
:raises ValueError: If the provided `person_data` is not valid based on the
`PersonUpdateData` schema.
:return: Internal person ID of the created or updated person.
"""
if isinstance(person_data, dict):
try:
person_data = PersonUpdateData.model_validate(person_data)
except ValidationError as e:
raise ValueError(
"Parameter `person_data` must be a PersonUpdateData object or a dictionary that can be "
"validated against the PersonUpdateData schema"
) from e
self.logger.info('Creating or updating person record with LIMS-ID "%s"', person_data.id)
resp = self._send_modeldata(
"person",
person_data,
method="PUT",
handle_http_errors={400: "Could not create or update person entry."},
)
try:
return int(resp.content)
except ValueError:
raise VarvisError("Response could not be parsed. Expected internal person ID as integer.")
[docs]
def get_case_report(self, person_lims_id: str, draft: bool = False, inactive: bool = False) -> CaseReport:
"""
Retrieve a case report for a given person LIMS-ID.
The method communicates with the server using a GET request to fetch the case
report of a person identified by the provided LIMS-ID. It allows for additional
filters like draft and inactive status.
:param person_lims_id: The unique identifier for a person in the LIMS system.
:param draft: Set to true if a draft report with pending changes is explicitly requested instead of
the final report (submitted report).
:param inactive: Flag indicating whether inactive report items should be also returned (by default
inactive report items are not returned).
:return: A case report object fetched from the server.
"""
enabled_flags = []
if draft or inactive:
if draft:
enabled_flags.append("draft")
if inactive:
enabled_flags.append("inactive")
query_params = "?" + "&".join(map(lambda x: f"{x}=true", enabled_flags))
else:
query_params = ""
self.logger.info(
'Getting case report for person LIMS-ID "%s"%s',
person_lims_id,
f" ({', '.join(enabled_flags)})" if enabled_flags else "",
)
resp = self._send_request(
"GET",
f"cases/{_url_param(person_lims_id)}/report{query_params}",
handle_http_errors={
404: "Person with given LIMS-ID was not found or no report exists for the given criteria."
},
)
return _parse_response_for_model(CaseReport, resp, data_from_key="response")
[docs]
def get_report_info_for_persons(self) -> list[PersonReportItem]:
"""
Fetches the report information for all persons.
This method sends a "GET" request to the "cases/reports/info" endpoint to
retrieve detailed report information for persons. The response is parsed
into a list of `PersonReportItem` objects for further use within the
application.
:return: A list of `PersonReportItem` objects containing the report
information for all persons.
"""
self.logger.info("Getting report information for persons")
resp = self._send_request("GET", "cases/reports/info")
return _parse_response_for_model_list(PersonReportItem, resp, "response")
[docs]
def get_person_analyses(self, person_lims_id: str) -> list[AnalysisItem]:
"""
Gets the list of analyses associated with a person based on the provided LIMS ID.
The analyses are returned as a list of `AnalysisItem` objects.
This method is very similar to ``get_analyses()``, but it retrieves analyses for a
specific person. In contrast to ``get_analyses()``, this API endpoint doesn't allow
for filtering by analysis ID.
:param person_lims_id: The unique LIMS ID representing the person whose analyses
are being retrieved.
:return: A list of AnalysisItem objects containing information about the analyses
associated with the provided person LIMS ID.
"""
self.logger.info('Getting analyses for person LIMS ID "%s"', person_lims_id)
resp = self._send_request(
"GET",
f"person/{_url_param(person_lims_id)}/analyses",
handle_http_errors={400: "Person with given LIMS-ID was not found."},
)
return _parse_response_for_model_list(AnalysisItem, resp)
[docs]
def find_analyses_by_filename(self, filename: str | list[str]) -> list[FindByInputFileNameAnalysisItem]:
"""
Find analyses by searching for the given filename components within customer-provided input file names.
:param filename: A single filename or a list of filenames as substrings to search for in the analyses.
If a list of strings is provided, *all* of the provided strings must be found in a customer input file
name for the corresponding analysis to be included in the result (AND operator). Provided strings must be
non-empty.
:return: A list of `FindByInputFileNameAnalysisItem` objects resulting from the search.
"""
if isinstance(filename, str):
filename = [filename]
filename = [f for f in (f.strip() for f in filename) if f]
if len(filename) == 0:
raise ValueError("Parameter `filename` must be a non-empty string or a non-empty list of strings")
self.logger.info(
"Getting analyses by searching for filename component(s) %s",
" AND ".join(f'"{f}"' for f in filename),
)
query_param = "&".join(f"customerProvidedInputFileName={_url_param(f)}" for f in filename)
resp = self._send_request(
"GET",
"analysis-list/find-by-customer-provided-input-file-name?" + query_param,
)
return _parse_response_for_model_list(FindByInputFileNameAnalysisItem, resp, data_from_key="response")
[docs]
def get_virtual_panel_summaries(self) -> list[VirtualPanelSummary]:
"""
Retrieves the summaries for all virtual panels, except for the virtual panel containing all genes.
This method sends a GET request to the "virtual-panels" endpoint to fetch a
list of virtual panel summaries. The response is parsed to return
a structured list of `VirtualPanelSummary` objects.
:return: A list of `VirtualPanelSummary` instances representing the
parsed virtual panel summaries.
"""
self.logger.info("Getting virtual panel summaries")
resp = self._send_request("GET", "virtual-panels")
return _parse_response_for_model_list(VirtualPanelSummary, resp, data_from_key="response")
[docs]
def get_virtual_panel(self, virtual_panel_id: int) -> VirtualPanelData:
"""
Retrieves the information of a virtual panel based on the given ID. This method
sends a GET request to fetch the virtual panel data and parses the response
into a VirtualPanelData model. If the virtual panel ID is not found, a
VarvisError is raised.
:param virtual_panel_id: The ID of the virtual panel to retrieve
:return: An instance of VirtualPanelData containing the information of the virtual panel
:raises VarvisError: If the virtual panel with the specified ID is not found
"""
self.logger.info(f"Getting virtual panel {virtual_panel_id}")
resp = self._send_request("GET", f"virtual-panel/{virtual_panel_id}")
try:
return _parse_response_for_model(VirtualPanelData, resp, data_from_key="response")
except VarvisError as exc:
raise VarvisError(f"Virtual panel with ID {virtual_panel_id} not found") from exc
[docs]
def get_all_genes(self) -> list[VarvisGene]:
"""
Retrieves a list of all genes and their details. The details are reduced to information
important for virtual panels.
This method sends a GET request to the "virtual-panel-genes" endpoint
to fetch a list of available genes. The response is parsed into a list
of VarvisGene models and returned.
:return: A list containing instances of VarvisGene parsed from the response.
"""
self.logger.info("Getting all genes")
resp = self._send_request("GET", "virtual-panel-genes")
return _parse_response_for_model_list(VarvisGene, resp, data_from_key="response")
[docs]
def get_file_download_links(self, analysis_id: int) -> AnalysisFileDownloadLinks:
"""
Retrieves the file download links associated with the provided analysis ID.
:param analysis_id: The unique identifier of the analysis for which
the file download links are to be retrieved.
:return: An instance of AnalysisFileDownloadLinks containing the
download links for the specified analysis.
"""
self.logger.info("Getting file download links for analysis ID %d", analysis_id)
resp = self._send_request(
"GET",
f"analysis/{analysis_id}/get-file-download-links",
handle_http_errors={400: "Analysis with given ID was not found."},
)
return _parse_response_for_model(AnalysisFileDownloadLinks, resp, data_from_key="response")
[docs]
def create_or_update_virtual_panel(self, virtual_panel_data: VirtualPanelUpdateData | dict[str, Any]) -> int:
"""
Creates or updates a virtual panel entry based on the provided data. If the
provided data doesn't include an ID, a new virtual panel will be created. If
an ID is specified in the data, an existing virtual panel with that ID will
be updated accordingly. Validates the input data against the
VirtualPanelUpdateData schema before processing.
:param virtual_panel_data: Data for the virtual panel. Must be either an
instance of VirtualPanelUpdateData or a dictionary conforming to the
VirtualPanelUpdateData schema.
:return: ID of the created or updated virtual panel.
"""
if isinstance(virtual_panel_data, dict):
try:
virtual_panel_data = VirtualPanelUpdateData.model_validate(virtual_panel_data)
except ValidationError as e:
raise ValueError(
"Parameter `virtual_panel_data` must be a VirtualPanelUpdateData instance or a dictionary that "
"can be validated against the VirtualPanelUpdateData schema."
) from e
if virtual_panel_data.id is None:
self.logger.info("Creating a new virtual panel")
else:
self.logger.info("Updating virtual panel with ID %d", virtual_panel_data.id)
resp = self._send_modeldata("virtual-panel", virtual_panel_data)
vp_id = _parse_response_for_primitive(resp, "response")
if not isinstance(vp_id, int):
raise VarvisError("Parsed response item should be an integer")
return vp_id
[docs]
def download_files(
self,
analysis_id: int,
output_path: str | os.PathLike,
file_patterns: str | list[str] | None = None,
allow_overwrite: bool = False,
show_progress_bar: bool = False,
max_parallel_downloads: int = 1,
only_collect_urls: bool = False,
) -> dict[str, Path]:
"""
Downloads files for a given analysis ID while supporting filtering by file patterns, parallel downloads,
progress display, and overwrite control.
The function fetches download links and attempts to download the files concurrently
up to the maximum number of parallel downloads. Files can be filtered using
specific patterns and duplicate downloads or invalid files are skipped.
Each file is downloaded to a temporary file in the output directory and atomically
moved to its destination after a successful transfer. Failed or interrupted downloads
preserve existing destination files and remove partial temporary files. Logs provide
information about failed downloads, skipped files, and completion status.
:param analysis_id: Integer representing the unique ID for the analysis data to be downloaded.
:param output_path: Path where the downloaded files will be stored. Accepts path-like objects.
:param file_patterns: File name pattern(s) used to include files for download. Can be a single string
or a list of strings. If None, all files are included.
:param allow_overwrite: Boolean flag determining whether to overwrite existing files in the output path.
:param show_progress_bar: Boolean flag indicating whether to display progress bars for each download.
:param max_parallel_downloads: Maximum number of files to download concurrently. Must be at least 1.
:param only_collect_urls: Boolean flag indicating whether to only collect download URLs without actually
downloading files. Useful when collecting file download URLs for multiple analyses
and then submitting them to
:meth:`_varvis_client.VarvisClient.download_files_from_urls_parallel`.
:return: If ``only_collect_urls`` is False: A dictionary mapping the downloaded file names to their download
location as Path objects. Otherwise: A dictionary that maps download URLs to the corresponding target file
Path objects.
"""
output_path = Path(output_path)
if not output_path.exists():
raise ValueError(f"Output path does not exist: {output_path}")
if max_parallel_downloads < 1:
raise ValueError("Parameter `max_parallel_downloads` must be at least 1")
download_links = self.get_file_download_links(analysis_id)
if not only_collect_urls:
self.logger.info("Downloading files for analysis ID %d", analysis_id)
if isinstance(file_patterns, str):
file_patterns = [file_patterns]
if file_patterns and not only_collect_urls:
self.logger.info(
"> Using file pattern(s) %s",
", ".join(f'"{pat}"' for pat in file_patterns),
)
# first only collect the valid download links
valid_download_links: dict[str, Path] = {} # maps link to output file path string
for link_object in download_links.apiFileLinks:
if link_object.fileName is None or link_object.downloadLink is None:
continue
output_file_name = link_object.fileName.strip()
try:
_validate_download_filename(output_file_name)
output_file_path = output_path / output_file_name
output_file_path.resolve().relative_to(output_path.resolve()) # will raise ValueError for on walk up
except (ValueError, OSError):
self.logger.error(
'> Skipping file "%s" because it has an invalid name',
output_file_name,
)
continue
if file_patterns:
output_file_name_lwr = output_file_name.lower()
if not any(fnmatchcase(output_file_name_lwr, pat.lower()) for pat in file_patterns):
self.logger.info(
'> Skipping file "%s" because it does not match any of the provided patterns',
output_file_name,
)
continue
if link_object.currentlyArchived:
self.logger.error(
'> Skipping file "%s" because it is marked as currently archived with estimated restore time %s',
output_file_name,
link_object.estimatedRestoreTime,
)
continue
if output_file_path.exists():
if allow_overwrite:
self.logger.warning('> Will overwrite existing file "%s"', output_file_path)
else:
self.logger.error(
'> Skipping file "%s" because it already exists at "%s"',
output_file_name,
output_file_path,
)
continue
if link_object.downloadLink in valid_download_links:
self.logger.error(
'> Skipping file "%s" because its download link was already collected',
output_file_name,
)
continue
valid_download_links[link_object.downloadLink] = output_file_path
if only_collect_urls:
return valid_download_links
res: dict[str, Path] = self.download_files_from_urls_parallel(
valid_download_links,
max_parallel_downloads,
show_progress_bar,
return_messages=False,
)
return res
[docs]
@staticmethod
def check_download_files_for_path_collisions(urls_and_targets: dict[str, Path]) -> list[str]:
"""
Return a list of duplicate target paths in ``urls_and_targets``, if any. If there are no collisions, an
empty list will be returned.
:param urls_and_targets: dictionary mapping download URLs to target paths
:return: A list of duplicate target paths, if any.
"""
n_items_per_target_path = Counter(map(lambda f: str(f.resolve()), urls_and_targets.values()))
return [target_path for target_path, n in n_items_per_target_path.items() if n > 1]
@overload
def download_files_from_urls_parallel(
self,
urls_and_targets: dict[str, Path],
max_parallel_downloads: int,
show_progress_bar: bool,
return_messages: Literal[False],
) -> dict[str, Path]: ...
@overload
def download_files_from_urls_parallel(
self,
urls_and_targets: dict[str, Path],
max_parallel_downloads: int,
show_progress_bar: bool,
return_messages: Literal[True],
) -> tuple[dict[str, Path], list[tuple[int, str]]]: ...
@overload
def download_files_from_urls_parallel(
self,
urls_and_targets: dict[str, Path],
max_parallel_downloads: int,
show_progress_bar: bool,
return_messages: bool,
) -> dict[str, Path] | tuple[dict[str, Path], list[tuple[int, str]]]: ...
[docs]
def download_files_from_urls_parallel(
self,
urls_and_targets: dict[str, Path],
max_parallel_downloads: int,
show_progress_bar: bool,
return_messages: bool,
) -> dict[str, Path] | tuple[dict[str, Path], list[tuple[int, str]]]:
"""
Downloads multiple files from HTTPS URLs to specified target locations concurrently.
This function utilizes a thread pool executor to download multiple files in parallel. Each file
is downloaded to a specified target location provided in the dictionary. The function keeps
track of progress and logs success or failure for each file download. Optionally, a progress
bar can be shown for each download to visualize the current progress. URLs without an HTTPS
scheme and network location are rejected. Downloads are written to temporary files beside their
destinations and atomically moved into place only after successful completion. Partial files are
removed after failures or interruptions. When an uncompressed response declares a valid
``Content-Length``, its downloaded byte count is verified before the file is committed. Batches
containing multiple URLs that resolve to the same target path are rejected before workers start.
Requests share a pooled session and retry transport failures up to ``download_max_retries``
times, with exponential backoff and full jitter based on ``download_backoff_factor_seconds``.
Interrupted response bodies are not retried.
Pool sizes default to the worker count and control retained connections rather than
worker concurrency. The pooled session closes after all workers finish.
:param urls_and_targets: A dictionary where keys are URLs pointing to files to download and
values are their respective output file paths where the downloaded files should be stored.
:param max_parallel_downloads: The maximum number of downloads that can run concurrently.
:param show_progress_bar: A boolean flag to indicate whether the progress bar should be
displayed for file downloads.
:param return_messages: A boolean flag to indicate whether to additionally return a list of
log messages.
:return: A dictionary mapping filenames to their respective output paths for successfully
downloaded files. If ``return_messages`` is True, additionally returns a list of log
messages.
:raises ValueError: If multiple download URLs resolve to the same target path.
"""
def download_single_file(
sess: Session,
url: str,
output_file_path: Path,
download_num: int,
logger: logging.Logger,
ssl_verify: bool,
request_timeout: float | tuple[float, float],
show_progress_bar: bool,
download_chunk_size: int,
interrpt_event: threading.Event,
) -> bool:
def human_readable_size(size_bytes: int) -> str:
units = ["B", "KB", "MB", "GB", "TB", "PB"]
size = float(size_bytes)
for unit in units:
if size < 1024.0:
return f"{size:.2f} {unit}"
size /= 1024.0
return f"{size:.2f} PB"
dl_num1 = download_num + 1
parsed_url = urlsplit(url)
if parsed_url.scheme.lower() != "https" or not parsed_url.netloc:
logger.error(
"Download #%d: Refusing to download a URL that is not HTTPS: %s",
dl_num1,
url,
)
return False
if interrpt_event.is_set():
logger.warning(f"Download #{dl_num1}: Download interrupted")
return False
try:
resp = sess.get(url, stream=True, verify=ssl_verify, timeout=request_timeout)
except requests.RequestException:
logger.exception(f"Download #{dl_num1}: Error on download request")
return False
if not resp.ok:
logger.error(
f"Download #{dl_num1}: HTTP error {resp.status_code} while downloading the file: {resp.reason}"
)
resp.close()
return False
n_downloaded_bytes = 0
content_length = None
if resp.headers.get("Content-Encoding", "identity") == "identity":
try:
content_length = int(resp.headers["Content-Length"])
except (KeyError, ValueError):
logger.info(
f"Download #{dl_num1}: No valid Content-Length in HTTP response header -- will omit "
"final content length check of the downloaded file"
)
else:
logger.info(
f"Download #{dl_num1}: Content-Encoding signals compressed HTTP content -- will omit "
"final content length check of the downloaded file"
)
progress_bar_output_stream = None
if show_progress_bar:
for hndlr in logger.handlers:
if isinstance(hndlr, logging.StreamHandler):
progress_bar_output_stream = hndlr.stream
break
else:
if content_length is not None:
size_log_str = f"of size {human_readable_size(content_length)}"
else:
size_log_str = "of unknown size"
logger.info(
f'Download #{dl_num1}: Downloading file {size_log_str} to output file "{output_file_path}"...'
)
logger.debug(f"Download #{dl_num1}: URL: {url}")
with tqdm(
desc=f"Download #{dl_num1}",
file=progress_bar_output_stream,
total=content_length,
unit="B",
unit_scale=True,
miniters=1,
disable=not show_progress_bar,
) as progress_bar:
tempfile_name = None
try:
with tempfile.NamedTemporaryFile(
mode="wb", delete=False, dir=output_file_path.parent, suffix=".part"
) as f:
# store the temporary file name for later deletion if necessary
tempfile_name = f.name
# retain the file permissions
file_mode = output_file_path.stat().st_mode if output_file_path.exists() else 0o664
os.fchmod(f.fileno(), file_mode)
# receive and write the file chunks
for chunk in resp.iter_content(chunk_size=download_chunk_size):
if interrpt_event.is_set():
raise KeyboardInterrupt()
if show_progress_bar:
progress_bar.update(len(chunk))
f.write(chunk)
n_downloaded_bytes += len(chunk)
if interrpt_event.is_set():
raise KeyboardInterrupt()
if content_length is not None and n_downloaded_bytes != content_length:
logger.error(
f"Download #{dl_num1}: Number of downloaded bytes ({n_downloaded_bytes}) does not "
f"match number of expected bytes ({content_length})"
)
resp.close()
return False
os.replace(tempfile_name, output_file_path)
except (
requests.RequestException,
OSError,
KeyboardInterrupt,
) as exc:
if isinstance(exc, KeyboardInterrupt):
logger.warning(f"Download #{dl_num1}: Download interrupted")
else:
logger.exception(f"Download #{dl_num1}: Error while downloading the file")
if tempfile_name:
try:
os.unlink(tempfile_name)
except OSError:
pass
resp.close()
return False
finally:
# clean up temporary file
if tempfile_name:
try:
os.unlink(tempfile_name)
except OSError:
pass
resp.close()
return True
# check for duplicate target paths to prevent collisions
target_path_collisions = VarvisClient.check_download_files_for_path_collisions(urls_and_targets)
if target_path_collisions:
collisions_str = "\n".join(target_path_collisions)
raise ValueError(
"Provided `urls_and_targets` contain duplicate target paths for different download URLs:\n"
f"{collisions_str}"
)
# set up a connection pool
with Session() as sess:
sess_adapter = HTTPAdapter(
pool_connections=self.download_pool_size or max_parallel_downloads,
pool_maxsize=self.download_pool_max_size or max_parallel_downloads,
max_retries=_DownloadRetry(
total=self.download_max_retries,
backoff_factor=self.download_backoff_factor_seconds,
),
)
sess.mount("http://", sess_adapter)
sess.mount("https://", sess_adapter)
# download in parallel
interruption_event = threading.Event()
with ThreadPoolExecutor(max_workers=max_parallel_downloads) as executor:
# enter try-block that catches program cancallation (ctrl-c) as KeyboardInterrupt exception
try:
# submit tasks ("futures") to threads
futures = {} # maps Future objects to URLs
for i, (url, target_path) in enumerate(urls_and_targets.items()):
fut = executor.submit(
download_single_file,
sess,
url,
target_path,
download_num=i,
logger=self.logger,
ssl_verify=self.ssl_verify,
request_timeout=self.request_timeout,
show_progress_bar=show_progress_bar,
download_chunk_size=self.download_chunk_size,
interrpt_event=interruption_event,
)
futures[fut] = (url, i)
# iterate through completed tasks (order is random)
collected_messages = [] # for deferring messages when using progress bars
downloaded_files = {}
for fut in as_completed(futures):
url, download_num = futures[fut]
downloaded_target_path = Path(urls_and_targets[url])
file_name = downloaded_target_path.name
try:
success = fut.result()
except Exception as exc:
msg = f'Download #{download_num + 1}: Error while downloading file "{file_name}": {exc}'
if show_progress_bar:
collected_messages.append((logging.ERROR, msg))
else:
self.logger.error(msg)
else:
if success:
downloaded_files[file_name] = downloaded_target_path
msg = f'Download #{download_num + 1}: Successfully downloaded file "{file_name}"'
if show_progress_bar:
collected_messages.append((logging.INFO, msg))
else:
self.logger.info(msg)
else:
msg = f'Download #{download_num + 1}: Could not download file "{file_name}"'
if show_progress_bar:
collected_messages.append((logging.ERROR, msg))
else:
self.logger.error(msg)
except KeyboardInterrupt:
# download interrupted
interruption_event.set()
raise
if show_progress_bar and not return_messages:
# log the collected messages after all downloads completed
for lvl, msg in collected_messages:
self.logger.log(lvl, msg)
# return results
if return_messages:
return downloaded_files, collected_messages
else:
return downloaded_files
[docs]
def request(
self,
endpoint: str,
method: str = "GET",
stream: bool = True,
raise_for_status: bool = True,
handle_http_errors: dict[int, str] | bool = True,
allow_retries: bool = True,
**kwargs: Any,
) -> requests.Response:
"""
Sends an authenticated HTTP request to the specified Varvis endpoint.
:param endpoint: The endpoint to which the request will be sent.
:param method: The HTTP method to use for the request. Defaults to "GET".
:param stream: Whether to stream the response. Defaults to True.
:param raise_for_status: Whether to raise an exception for HTTP error responses.
Defaults to True.
:param handle_http_errors: Specifies custom error handling behavior. If True,
default error messages are used. If False, no custom handling is applied.
A dictionary mapping status codes to custom messages can also be provided.
Defaults to True.
:param allow_retries: Whether to allow retries for failed requests. Defaults to True.
:param kwargs: Additional arguments to be passed to the request.
:return: The HTTP response object returned by the request.
"""
if handle_http_errors is True:
handle_http_errors_arg = DEFAULT_HTTP_ERROR_MESSAGES
elif handle_http_errors is False:
handle_http_errors_arg = None
else:
handle_http_errors_arg = handle_http_errors
return self._send_request(
method=method,
endpoint=endpoint,
auto_fix_endpoint=False,
stream=stream,
raise_for_status=raise_for_status,
handle_http_errors=handle_http_errors_arg,
allow_retries=allow_retries,
**kwargs,
)
def _send_modeldata(
self,
endpoint: str,
model: BaseModel,
method: str = "POST",
handle_http_errors: dict[int, str] | None = None,
) -> requests.Response:
"""
Sends serialized model data to the specified endpoint using the given HTTP method.
This function serializes the provided model instance to a dictionary and sends
it as a JSON payload to a specified URL endpoint. The request method used for
this operation can be customized, defaulting to "POST". This function assumes
proper formatting and validation of the `model` before its invocation.
:param endpoint: The specific part of the URL to append to the base endpoint.
:param model: An instance of a class inheriting from BaseModel, representing
the model data to send.
:param method: The HTTP method to use for the request, defaulting to "POST".
:param handle_http_errors: Optional dict of HTTP error codes to handle explicitly. The dict maps HTTP
error codes to custom error messages. If None, the default error messages are used.
If one of the supplied HTTP errors occurs, a ``VarvisError`` will be raised. Defaults to None.
:return: The response from the HTTP request.
"""
data = model.model_dump()
return self._send_request(method, endpoint, handle_http_errors=handle_http_errors, json=data)
def _recover_session(self, state: _SessionState) -> _SessionState:
"""
Share one bounded re-login sequence among requests using the same session.
One worker performs up to ``backoff_max_tries`` login attempts. Other workers
wait for that session's recovery future and receive its result or failure,
including workers whose forced-logout responses arrive after recovery finishes.
Explicit logout invalidates the session's generation and wakes waiting workers
with a shared failure instead of allowing further login attempts.
Login retries use deterministic exponential backoff and a budget separate from HTTP
request attempts. Waiters hold neither client lock while waiting for the outcome.
:param state: The invalidated session record whose recovery outcome is shared.
:return: The replacement authenticated session record.
:raises VarvisError: If the shared login attempt budget is exhausted.
"""
# elect one recovery owner while keeping the outcome attached to the old session
with self._recovery_lock:
if state.logout_generation != self._logout_generation:
raise VarvisError("Session recovery cancelled by logout")
owner = not state.recovery.running() and not state.recovery.done()
if owner:
state.recovery.set_running_or_notify_cancel()
self._active_recoveries.append(state)
if not owner:
# the future waits without holding the session lock needed by the login owner
return state.recovery.result()
# keep the entire login retry budget independent of the number of request workers
try:
last_error: Exception | None = None
for attempt in range(1, self.backoff_max_tries + 1):
with self._session_lock:
# reject recovery invalidated during backoff or before lock acquisition
if state.logout_generation != self._logout_generation:
raise VarvisError("Session recovery cancelled by logout")
try:
if self.login(raise_for_status=False):
recovered = self._session_state
if recovered is None:
raise RuntimeError("Login succeeded without initializing session state")
state.recovery.set_result(recovered)
return recovered
except (requests.RequestException, VarvisError) as exc:
last_error = exc
# release the session lock between attempts so other requests can join recovery
if attempt < self.backoff_max_tries:
wait_sec = 2 ** (attempt - 1) * self.backoff_factor_seconds
self.logger.debug("Retrying shared login in %.1f seconds", wait_sec)
time.sleep(wait_sec)
raise VarvisError(
f"Failed to reauthenticate with Varvis API after {self.backoff_max_tries} attempts"
) from last_error
except BaseException as exc:
# always wake waiting workers, including when the recovery owner is interrupted
with self._recovery_lock:
if not state.recovery.done():
state.recovery.set_exception(exc)
raise
return state.recovery.result()
finally:
# retain sleeping owners across replacement sessions until they have stopped
with self._recovery_lock:
self._active_recoveries.remove(state)
def _send_request(
self,
method: str,
endpoint: str,
auto_fix_endpoint: bool = True,
stream: bool = True,
raise_for_status: bool = True,
handle_http_errors: dict[int, str] | None = None,
allow_retries: bool = True,
**kwargs: Any,
) -> requests.Response:
"""
Sends an HTTP request to the specified endpoint and handles retries with optional
re-login in cases of failed requests or invalid session scenarios.
This method constructs the full API URL dynamically depending on the endpoint and
manages session-based retries when needed. Certain API quirks, such as session
invalidations, are taken into account and handled appropriately. Session acquisition
waits for login to finish, while requests run outside the session lock. Forced logout
responses invalidate only the session used for that request. Concurrent requests
share one bounded re-login sequence and its success or failure. A later request
may start fresh recovery after a failed sequence. The first request requires an
explicit login, and disabled retries do not initiate recovery.
Request retry delays use full jitter, chosen uniformly between zero and
``backoff_factor_seconds * 2 ** (retry_number - 1)``.
Before sending, delayed workers follow successive recovery outcomes until reaching
a session without known recovery, without consuming HTTP attempts.
Explicit logout invalidates existing workers so they cannot initiate recovery
or retry requests after a later explicit login. In-flight I/O is not interrupted.
:param method: HTTP method to use for the request (e.g., "GET", "POST"). Type should match a valid HTTP method string.
:param endpoint: Partial URL to append to the API base URL. Determines the specific endpoint being addressed.
:param auto_fix_endpoint: If True, automatically prepend "api/" to the endpoint in the cases where it is necessary.
:param stream: Whether to stream the HTTP response content. Defaults to True.
:param raise_for_status: If True, raises an HTTP error for non-successful status codes. Defaults to True.
:param handle_http_errors: Optional dict of HTTP error codes to handle explicitly. The dict maps HTTP
error codes to custom error messages. If None, the default error messages are used.
If one of the supplied HTTP errors occurs, a ``VarvisError`` will be raised. Defaults to None.
:param allow_retries: Whether to allow retrying the request in case of failure. Defaults to True.
:param kwargs: Additional request options passed to `requests.Session.request()`.
:return: HTTP response object from the `requests` library containing the server's response.
"""
# join the current recovery group before waiting for any in-progress login I/O
is_authentication = endpoint.removeprefix("/") in {"login", "logout"}
with self._recovery_lock:
session_state = self._session_state
if not is_authentication and session_state is not None and session_state.recovery.done():
# a new request gets a fresh budget while older requests retain the failed outcome
session_state = _SessionState(session_state.session, session_state.logout_generation)
self._session_state = session_state
if not is_authentication and session_state is not None and session_state.recovery.running():
# wait outside both locks so workers arriving during the last login share its outcome
session_state = self._recover_session(session_state)
# capture an authenticated session, including when the initial explicit login just finished
with self._session_lock:
if session_state is None:
with self._recovery_lock:
session_state = self._session_state
request_session = self._session if is_authentication or session_state is None else session_state.session
if request_session is None and session_state is None:
raise VarvisError("Session not initialized -- you need to login first")
needs_recovery = not is_authentication and not self.logged_in
if needs_recovery and allow_retries and session_state is not None:
session_state = self._recover_session(session_state)
request_session = session_state.session
elif needs_recovery:
raise VarvisError("Session not authenticated -- you need to login first")
endpoint = endpoint.removeprefix("/")
if auto_fix_endpoint:
urlpart_components = endpoint.split("?")[0].split("/")
# strangely, some endpoints are prefixed with "api/" and others don't, so we need to handle these cases:
# - endpoint starting with "login", "logout", "pending-cnv", "analysis-list", etc.: NO prefix
# - endpoint with the pattern "person/.../analyses": NO prefix (e.g. "/person/{personLimsId}/analyses", etc.)
# - endpoint starting with "virtual-panel" and method is not GET: prefix
# - all other endpoints will also get an "/api" prefix (e.g. "/api/analyses/{analysisId}")
if not (
urlpart_components[0]
in {
"login",
"logout",
"pending-cnv",
"analysis-list",
"virtual-panels",
"virtual-panel",
"virtual-panel-genes",
}
or (urlpart_components[0] == "person" and urlpart_components[-1] == "analyses")
) or (urlpart_components[0] == "virtual-panel" and method != "GET"):
endpoint = "api/" + endpoint
# construct final URL and send request
request_url = self.api_url + endpoint
resp = None
last_transport_error: requests.RequestException | None = None
transport_failure_kind: str | None = None
n_tries = 1
max_retries = self.backoff_max_tries if allow_retries else 1
while True:
self.logger.debug(
"Try #%d/%d. Sending %s request to %s",
n_tries,
max_retries,
method,
request_url,
)
retry = False # reset on each try
# follow successive recoveries before sending through a replacement session
while (
not is_authentication
and session_state is not None
and (session_state.recovery.running() or session_state.recovery.done())
):
session_state = self._recover_session(session_state)
request_session = session_state.session
# stop old workers before another send even after a later explicit login
with self._session_lock:
if (
not is_authentication
and session_state is not None
and session_state.logout_generation != self._logout_generation
):
raise VarvisError("Session recovery cancelled by logout")
if request_session is not None:
# send the request using the current request session
try:
resp = request_session.request(
method,
request_url,
stream=stream,
allow_redirects=False,
timeout=self.request_timeout,
**kwargs,
)
self.logger.debug("Response received with status code %d", resp.status_code)
if resp.status_code == 302 and resp.headers.get("Location", "").removesuffix(
"/"
) == self.api_url.removesuffix("/"):
# Varvis API quirk: in case there are too many concurrent requests for the same session, the
# session seems to be invalidated and the response is a redirect to the base URL; we need to
# login again after waiting a bit
self.logger.debug("Possible forced logout from API detected -- will try to login again")
retry = True
# ignore stale redirects from a session that has already been replaced
with self._session_lock:
if self._session is request_session:
self._loggedin_csrf = None
resp = None
if allow_retries and n_tries < max_retries and session_state is not None:
session_state = self._recover_session(session_state)
request_session = session_state.session
except requests.ConnectTimeout as exc:
self.logger.debug("Connection timeout")
last_transport_error = exc
transport_failure_kind = "connect_timeout"
retry = True
except requests.ReadTimeout as exc:
self.logger.debug("Read timeout")
last_transport_error = exc
transport_failure_kind = "read_timeout"
retry = True
except requests.Timeout as exc:
self.logger.debug("Timeout error")
last_transport_error = exc
transport_failure_kind = "timeout"
retry = True
except requests.ConnectionError as exc:
self.logger.debug("Connection error")
last_transport_error = exc
transport_failure_kind = "connection"
retry = True
if retry and n_tries < max_retries:
# spread request retries uniformly across the exponential backoff window
max_wait = 2 ** (n_tries - 1) * self.backoff_factor_seconds
wait_sec = random.uniform(0, max_wait) # nosec B311
n_tries += 1
self.logger.debug("Retrying in %.1f seconds", wait_sec)
time.sleep(wait_sec)
else:
break
if resp is None:
if last_transport_error is not None and transport_failure_kind is not None:
# retain transport failure identity for callers
error_class = (
VarvisTimeoutError if isinstance(last_transport_error, requests.Timeout) else VarvisTransportError
)
raise error_class(transport_failure_kind, n_tries) from last_transport_error
raise VarvisError(f"Failed to send request to Varvis API after {n_tries} attempts")
if handle_http_errors and resp.status_code in handle_http_errors:
custom_error_msg = handle_http_errors[resp.status_code]
if custom_error_msg:
custom_error_msg = " " + custom_error_msg
try:
error_data = resp.json()
except JSONDecodeError:
error_data = {}
_raise_varvis_error(
error_data,
f"Varvis API returned HTTP status code {resp.status_code}.{custom_error_msg}",
)
elif raise_for_status:
resp.raise_for_status()
return resp
def _reset_state_on_logout(self, clear_recovery: bool = True) -> None:
"""
Clear authentication while holding the session lock.
:param clear_recovery: Whether to discard the shared recovery record. Failed
login attempts retain it so participating requests share the final outcome.
"""
# serialize authentication state changes
with self._session_lock:
self._session = None
self._loggedin_csrf = None
if clear_recovery:
with self._recovery_lock:
self._session_state = None