Client

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>

class varvis_connector._varvis_client.VarvisClient(api_url, username, password, https_proxy=None, ssl_verify=True, connection_timeout=10.0, backoff_factor_seconds=0.5, backoff_max_tries=5, download_chunk_size=5242880, logger=<Logger not_initialized (Level -1)>, _session=None, _loggedin_csrf=None)

Bases: object

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.

Variables:

• api_url – The base URL for the Varvis API.

• username – The username used for Varvis authentication.

• password – The password used for Varvis authentication.

• https_proxy – Optional HTTPS proxy URL to use for API requests.

• ssl_verify – Whether to verify SSL certificates for API requests.

• connection_timeout – Timeout in seconds for API requests.

• backoff_factor_seconds – Base factor for exponential backoff between retries.

• backoff_max_tries – Maximum number of retry attempts for failed requests.

• logger – Logger instance for recording operations and errors.

• _session – The HTTP session used for Varvis API communication, initialized during login and cleared during logout.

• _loggedin_csrf – The CSRF token used for API requests after login.

Parameters:

• api_url (str)

• username (str)

• password (str)

• https_proxy (str | None)

• ssl_verify (bool)

• connection_timeout (float)

• backoff_factor_seconds (float)

• backoff_max_tries (int)

• download_chunk_size (int)

• logger (Logger)

• _session (Session | None)

• _loggedin_csrf (str | None)

property logged_in: bool

Check if the user is logged in.

login(raise_for_status=True)

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.

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.

Parameters:

raise_for_status (bool) – If true, raise an error if the login fails due to HTTP errors.

Raises:

• VarvisError – If the login failed.

• HTTPError – Indicates a failure during the HTTP communication with the varvis API.

Return type:

bool

Returns:

True if the login was successful, False otherwise.

logout()

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 upon successful logout. If the user is not logged in, the method will not perform any action and logs a corresponding message.

Raises:

HTTPError – Indicates a failure during the logout HTTP request.

Return type:

None

get_snv_annotations(analysis_id)

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.

Parameters:

analysis_id (int) – The unique identifier of the analysis for which SNV annotations are being retrieved.

Return type:

SnvAnnotationData

Returns:

An object of SnvAnnotationData containing the SNV annotations for the specified analysis.

get_cnv_target_results(person_lims_id, analysis_ids, virtual_panel_id=1)

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.

Parameters:

• person_lims_id (str) – The LIMS ID of the person for whom the CNV target results are being queried.

• analysis_ids (int | list[int]) – A single analysis ID or a list of analysis IDs to query CNV results for.

• virtual_panel_id (int | None) – (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 type:

CnvTargetResults

Returns:

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.

• requests.HTTPError – For errors occurring during the API request if they are unrelated to 422 status codes.

• VarvisError – If the API response indicates an invalid analysis ID.

get_internal_person_id(person_lims_id)

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.

Parameters:

person_lims_id (str) – The LIMS-ID of the person whose internal ID is to be retrieved.

Return type:

int

Returns:

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.

• requests.HTTPError – For HTTP errors other than 404.

get_pending_cnv_segments(*, person_id=None, person_lims_id=None, analysis_ids, virtual_panel_id=1)

Retrieves pending CNV segments based for a given person (identified either by internal ID or LIMS ID) and associated analysis IDs.

Parameters:

• person_id (int | None) – Internal ID of the person. Must be an integer or None and must be given if person_lims_id is None.

• person_lims_id (str | None) – LIMS ID of the person. Must be a string or None and must be given if person_id is None.

• analysis_ids (int | list[int] | None) – Analysis IDs associated with the person. Can be an integer or a list of integers.

• virtual_panel_id (int | None) – 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 type:

PendingCnvData

Returns:

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.

• VarvisError – Raised if the response from the Varvis API does not contain the expected data.

get_qc_case_metrics(person_lims_id)

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.

Parameters:

person_lims_id (str) – The LIMS ID of the person whose QC case metrics are to be retrieved.

Return type:

QCCaseMetricsData

Returns:

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.

get_coverage_data(person_lims_id, virtual_panel_id=1)

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.

Parameters:

• person_lims_id (str) – A string representing the person LIMS ID for whom the coverage data is to be retrieved.

• virtual_panel_id (int | None) – 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 type:

list[CoverageData]

Returns:

A list of CoverageData objects containing validated coverage information.

Raises:

VarvisError – If the validation of any coverage data item fails.

get_analyses(analysis_ids=None)

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.

Parameters:

analysis_ids (int | list[int] | None) – Optional list of analysis IDs to filter the results by. If provided, only analyses with matching IDs will be returned.

Return type:

list[AnalysisItem]

Returns:

A list of AnalysisItem objects retrieved from the varvis API.

get_person(person_lims_id)

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.

Parameters:

person_lims_id (str) – The unique LIMS-ID identifying the person for whom data is to be fetched.

Return type:

PersonData

Returns:

The detailed person data represented as a PersonData object.

create_or_update_person(person_data)

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.

Parameters:

person_data (PersonUpdateData | dict[str, Any]) – 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 type:

int

Returns:

Internal person ID of the created or updated person.

get_case_report(person_lims_id, draft=False, inactive=False)

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.

Parameters:

• person_lims_id (str) – The unique identifier for a person in the LIMS system.

• draft (bool) – Set to true if a draft report with pending changes is explicitly requested instead of the final report (submitted report).

• inactive (bool) – Flag indicating whether inactive report items should be also returned (by default inactive report items are not returned).

Return type:

CaseReport

Returns:

A case report object fetched from the server.

get_report_info_for_persons()

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 type:

list[PersonReportItem]

Returns:

A list of PersonReportItem objects containing the report information for all persons.

get_person_analyses(person_lims_id)

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.

Parameters:

person_lims_id (str) – The unique LIMS ID representing the person whose analyses are being retrieved.

Return type:

list[AnalysisItem]

Returns:

A list of AnalysisItem objects containing information about the analyses associated with the provided person LIMS ID.

find_analyses_by_filename(filename)

Find analyses by searching for the given filename components within customer-provided input file names.

Parameters:

filename (str | list[str]) – 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 type:

list[FindByInputFileNameAnalysisItem]

Returns:

A list of FindByInputFileNameAnalysisItem objects resulting from the search.

get_virtual_panel_summaries()

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 type:

list[VirtualPanelSummary]

Returns:

A list of VirtualPanelSummary instances representing the parsed virtual panel summaries.

get_virtual_panel(virtual_panel_id)

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.

Parameters:

virtual_panel_id (int) – The ID of the virtual panel to retrieve

Return type:

VirtualPanelData

Returns:

An instance of VirtualPanelData containing the information of the virtual panel

Raises:

VarvisError – If the virtual panel with the specified ID is not found

get_all_genes()

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 type:

list[VarvisGene]

Returns:

A list containing instances of VarvisGene parsed from the response.

get_file_download_links(analysis_id)

Retrieves the file download links associated with the provided analysis ID.

Parameters:

analysis_id (int) – The unique identifier of the analysis for which the file download links are to be retrieved.

Return type:

AnalysisFileDownloadLinks

Returns:

An instance of AnalysisFileDownloadLinks containing the download links for the specified analysis.

create_or_update_virtual_panel(virtual_panel_data)

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.

Parameters:

virtual_panel_data (VirtualPanelUpdateData | dict[str, Any]) – Data for the virtual panel. Must be either an instance of VirtualPanelUpdateData or a dictionary conforming to the VirtualPanelUpdateData schema.

Return type:

int

Returns:

ID of the created or updated virtual panel.

download_files(analysis_id, output_path, file_patterns=None, allow_overwrite=False, show_progress_bar=False, max_parallel_downloads=1, only_collect_urls=False)

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. Logs provide information about failed downloads, skipped files, and completion status.

Parameters:

• analysis_id (int) – Integer representing the unique ID for the analysis data to be downloaded.

• output_path (str | PathLike) – Path where the downloaded files will be stored. Accepts path-like objects.

• file_patterns (str | list[str] | None) – 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.

• allow_overwrite (bool) – Boolean flag determining whether to overwrite existing files in the output path.

• show_progress_bar (bool) – Boolean flag indicating whether to display progress bars for each download.

• max_parallel_downloads (int) – Maximum number of files to download concurrently. Must be at least 1.

• only_collect_urls (bool) – 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 _varvis_client.VarvisClient.download_files_from_urls_parallel().

Return type:

dict[str, Path]

Returns:

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.

download_files_from_urls_parallel(urls_and_targets, max_parallel_downloads, show_progress_bar, return_messages)

Overloads:

• self, urls_and_targets (dict[str, Path]), max_parallel_downloads (int), show_progress_bar (bool), return_messages (Literal[False]) → dict[str, Path]

• 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]]]

• 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]]]

Parameters:

• urls_and_targets (dict[str, Path])

• max_parallel_downloads (int)

• show_progress_bar (bool)

• return_messages (bool)

Return type:

dict[str, Path] | tuple[dict[str, Path], list[tuple[int, str]]]

Downloads multiple files from given 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.

Parameters:

• urls_and_targets (dict[str, Path]) – 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.

• max_parallel_downloads (int) – The maximum number of downloads that can run concurrently.

• show_progress_bar (bool) – A boolean flag to indicate whether the progress bar should be displayed for file downloads.

• return_messages (bool) – A boolean flag to indicate whether to additionally return a list of log messages.

Returns:

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.

Return type:

dict[str, Path] | tuple[dict[str, Path], list[tuple[int, str]]]

request(endpoint, method='GET', stream=True, raise_for_status=True, handle_http_errors=True, allow_retries=True, **kwargs)

Sends an authenticated HTTP request to the specified Varvis endpoint.

Parameters:

• endpoint (str) – The endpoint to which the request will be sent.

• method (str) – The HTTP method to use for the request. Defaults to “GET”.

• stream (bool) – Whether to stream the response. Defaults to True.

• raise_for_status (bool) – Whether to raise an exception for HTTP error responses. Defaults to True.

• handle_http_errors (dict[int, str] | bool) – 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.

• allow_retries (bool) – Whether to allow retries for failed requests. Defaults to True.

• kwargs (Any) – Additional arguments to be passed to the request.

Return type:

Response

Returns:

The HTTP response object returned by the request.