Clinvar Database

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A Python client for querying the NCBI ClinVar database via E-utilities."""

# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "scienceskillscommon",
#   "python-dotenv",
# ]
# [tool.uv.sources]
# scienceskillscommon = { path = "../../scienceskillscommon" }
# ///

from __future__ import annotations

import argparse
import json
import os
import sys
from typing import Any
import urllib.parse
import xml.etree.ElementTree as ET

import dotenv
from science_skills.scienceskillscommon import http_client


class _Response:

  def __init__(self, status_code: int, content: bytes):
    self.status_code = status_code
    self.content = content


class RateLimitError(Exception):
  """Raised when the NCBI API rate limit is exceeded."""


class ClinVarClient:
  """A robust Python client for querying the NCBI ClinVar database.

  This client uses NCBI E-utilities and handles rate limiting, API key
  authentication, and complex XML/JSON response parsing.
  """

  BASE_URL = 'https://eutils.ncbi.nlm.nih.gov/entrez/eutils/'

  def __init__(self):
    # Look for the NCBI API Key in the environment
    self.api_key = os.environ.get('NCBI_API_KEY')

    # NCBI limits: 10 req/sec with key, 3 req/sec without key
    self.rate_limit = 10 if self.api_key else 3
    self.client = http_client.HttpClient(self.BASE_URL, qps=self.rate_limit)

  def _request(self, endpoint: str, params: dict[str, Any]) -> '_Response':
    """Makes an HTTP request to the given E-utilities endpoint.

    Args:
      endpoint: The API endpoint to call (e.g. 'esearch.fcgi').
      params: Query parameters for the request.

    Returns:
      A `_Response` with status code and content bytes.

    Raises:
      RateLimitError: If a 429 status is received.
      RuntimeError: On any other HTTP or network error.
    """
    if self.api_key:
      params['api_key'] = self.api_key

    url = urllib.parse.urljoin(self.BASE_URL, endpoint)
    query_string = urllib.parse.urlencode(params, doseq=True)
    full_url = f'{url}?{query_string}'

    try:
      resp = self.client.fetch(full_url)
      return _Response(resp.status_code, resp.data)
    except http_client.HttpError as exc:
      if exc.status_code == 429:
        raise RateLimitError(
            'HTTP 429 Too Many Requests hit on NCBI E-utilities.\nAGENT'
            ' INSTRUCTION: Pause execution and inform the user that an NCBI API'
            ' Key is required to increase the rate limit, providing the URL'
            ' https://www.ncbi.nlm.nih.gov/clinvar/docs/api_http/ for'
            ' instructions on how to obtain one. The user will need to set the'
            ' NCBI_API_KEY environment variable and advise the agent to retry.'
        ) from exc
      raise RuntimeError(
          f'HTTP {exc.status_code} error from NCBI E-utilities: {exc}'
      ) from exc
    except Exception as exc:
      raise RuntimeError(
          f'Failed to fetch data from NCBI E-utilities: {exc}'
      ) from exc

  def count_variants(self, query: str) -> int:
    """Returns the total number of variants matching a query.

    This is a lightweight call that does not fetch any variant IDs. Use it
    to check result set size before committing to a full search.

    Args:
        query: NCBI Entrez search query string.

    Returns:
        Total number of matching variant IDs.
    """
    params = {
        'db': 'clinvar',
        'term': query,
        'rettype'