"""
ClinVar SQLite Schema Generator.
This module provides schema generation for ClinVar databases:
- RCV (Reference ClinVar Assertion) - condition-centric format
- VCV (Variant Call Variation) - variant-centric format
The SQL schemas are based on the following XSD:
- RCV: ClinVar_RCV_weekly.xsd v2.2 (August 6, 2025)
- VCV: ClinVar_VCV.xsd v2.5 (August 6, 2025)
Notes
-----
This module creates only the database schema structure (empty tables).
To populate databases with actual ClinVar data, use clinvar_parser.py.
"""
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
# imports
import os
import json
import sys
import logging
import argparse
from pathlib import Path
from clinvar_build.errors import (
is_type,
)
from clinvar_build.utils.general import (
_check_directory,
_check_directory_readable,
)
from clinvar_build.utils.parser_tools import (
configure_logging,
SQLiteParser,
)
from clinvar_build.utils.config_tools import check_environ
from clinvar_build.constants import (
DBNames,
ParserNames,
)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
# initiating a logger
logger = logging.getLogger(__name__)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class JSONSchemaLoader:
"""
Loader that derives SQLite Data Definition Language (DDL) from a parsed
JSON config file.
Reads a parse JSON config file and converts entity definitions, constraint
annotations, and index specifications into
``CREATE TABLE`` and ``CREATE INDEX`` SQL strings suitable for
``ClinVarSchemaGenerator.create_schema()``.
Attributes
----------
tables : `list` [`str`] or `None`
Complete ``CREATE TABLE`` SQL statements. Populated after
``load()`` is called.
indexes : `list` [`str`] or `None`
Complete ``CREATE INDEX`` SQL statements. Populated after
``load()`` is called.
Class Attributes
----------------
CAST_TYPE_MAP : `dict` [`str`, `str`]
Mapping from parse JSON ``cast`` values to SQLite column types.
Methods
-------
load(json_path)
Load a parse JSON config and build DDL dicts.
Examples
--------
>>> from pathlib import Path
>>> loader = JSONSchemaLoader()
>>> tables, indexes = loader.load(Path("resources/.../rcv_parse.json"))
>>> tables # list of CREATE TABLE statements
>>> indexes # list of CREATE INDEX statements
"""
# NOTE class attributes
# cast value to SQLite column type
CAST_TYPE_MAP: dict[str, str] = {
"int": "INTEGER",
"float": "REAL",
"bool": "BOOLEAN",
}
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __init__(self) -> None:
"""
Initialise the schema loader.
"""
self._config: dict | None = None
self.tables: list[str] | None = None
self.indexes: list[str] | None = None
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __repr__(self) -> str:
"""
Return unambiguous string representation.
"""
if self._config is None:
return f"{type(self).__name__}(not loaded)"
n_tables = len(self.tables) if self.tables else 0
n_indexes = len(self.indexes) if self.indexes else 0
return f"{type(self).__name__}(tables={n_tables}, indexes={n_indexes})"
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __str__(self) -> str:
"""
Return human-readable string representation.
"""
if self._config is None:
return (
f"{type(self).__name__}\n"
"Available methods:\n"
" - load(json_path): Load parse JSON and build DDL"
)
n_tables = len(self.tables) if self.tables else 0
n_indexes = len(self.indexes) if self.indexes else 0
return (f"{type(self).__name__}\nLoaded: {n_tables} tables, {n_indexes} "
"indexes"
)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def load(
self,
path: Path,
) -> tuple[list[str], list[str]]:
"""
Load schema Data Definition Language (DDL) from a parse JSON config
file.
Derives table and index DDL from the entity definitions and
constraint annotations in the JSON, returning lists in the same
format expected by ``ClinVarSchemaGenerator.create_schema()``.
Parameters
----------
path : `Path`
Path to a parse JSON config file (``*_parse.json``).
Returns
-------
tables : `list` [`str`]
Complete ``CREATE TABLE`` SQL statements.
indexes : `list` [`str`]
Complete ``CREATE INDEX`` SQL statements.
Examples
--------
>>> from pathlib import Path
>>> loader = JSONSchemaLoader()
>>> tables, indexes = loader.load(Path("rcv_parse.json"))
"""
is_type(path, (Path, str))
# load the json data
with open(path) as fh:
self._config = json.load(fh)
# initiate tables
self.tables = []
# Progress table from _meta
_m = ParserNames.meta
if _m in self._config and (ParserNames.meta_column in self._config[_m]):
self.tables.append(self._build_progress_table(self._config[_m]))
# NOTE makes the actual tables
# Entity tables (skip underscore-prefixed metadata keys)
# These will contain the actual data
for key, entity in self._config.items():
if key.startswith("_"):
continue
self.tables.append(self._build_entity_table_sql(entity))
# NOTE sets the indexes
_i = ParserNames.idxs
self.indexes = []
for index in self._config.get(_i, []):
cols = ", ".join(index[ParserNames.cols])
_idx = (
f"CREATE INDEX IF NOT EXISTS {index[ParserNames.name]} ON "
f"{index[ParserNames.table]}({cols})")
self.indexes.append(_idx)
# Return
return self.tables, self.indexes
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def _find_parent_table(self, parent: str) -> str:
"""
Return the table name of the entity whose ``returns_id`` matches.
Parameters
----------
parent : `str`
The ``parent_id`` column name to look up.
Returns
-------
table_name : `str`
The ``sql.table_name`` of the matching parent entity.
Raises
------
ValueError
If no entity with the matching ``returns_id`` is found.
Examples
--------
>>> loader = JSONSchemaLoader()
>>> loader._config = {
... "_meta": {"progress_table": "ParseProgress"},
... "VariationArchive": {
... "returns_id": "variation_archive_id",
... "parent_id": None,
... "sql": {"table_name": "VariationArchive"},
... },
... "SimpleAllele": {
... "returns_id": "simple_allele_id",
... "parent_id": "variation_archive_id",
... "sql": {"table_name": "SimpleAllele"},
... },
... }
>>> loader._find_parent_table("variation_archive_id")
'VariationArchive'
"""
is_type(parent, str)
for key, entity in self._config.items():
# ignore these tables
if key.startswith("_"):
continue
# extract these
if entity.get(ParserNames.rtrn_id) == parent:
return entity[ParserNames.sql][ParserNames.tab_name]
raise ValueError(f"No entity found with returns_id='{parent}'")
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@staticmethod
def _build_progress_table(meta: dict) -> str:
"""
Build CREATE TABLE SQL for the progress table from ``_meta``.
Parameters
----------
meta : `dict`
The ``_meta`` section from a parse JSON config, which must
contain ``progress_table`` and ``progress_columns`` keys.
The former including a string value for the table name, the latter
should contain a dictionary with the column names, and the column
`type`, the `default` value (optional), and `non_null` boolean
(optional).
Returns
-------
sql : `str`
``CREATE TABLE IF NOT EXISTS ...`` statement.
Examples
--------
>>> meta = {
... "progress_table": "ParseProgress",
... "progress_columns": {
... "xml_file": {"type": "TEXT", "not_null": True},
... "records_processed": {"type": "INTEGER", "default": 0},
... },
... }
>>> JSONSchemaLoader._build_progress_table(meta)
"CREATE TABLE IF NOT EXISTS ParseProgress (
id INTEGER PRIMARY KEY AUTOINCREMENT, xml_file TEXT NOT NULL,
status TEXT DEFAULT 'pending'
)"
"""
# get the meta data
table_name = meta[ParserNames.meta_table]
progress_cols = meta.get(ParserNames.meta_column, {})
# initiate the columns
parts = ["id INTEGER PRIMARY KEY AUTOINCREMENT"]
for col_name, col_def in progress_cols.items():
col_type = col_def.get(ParserNames.meta_type, "TEXT")
nn = " NOT NULL" if col_def.get(ParserNames.meta_nn) else ""
default = col_def.get(ParserNames.meta_default)
if default is None:
default_str = ""
elif isinstance(default, str):
default_str = f" DEFAULT '{default}'"
else:
default_str = f" DEFAULT {default}"
parts.append(f"{col_name} {col_type}{nn}{default_str}")
# return string
return (f"CREATE TABLE IF NOT EXISTS {table_name} "
f"(" + ", ".join(parts) + ")"
)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# NOTE consider making each ### step into a private method
def _build_entity_table_sql(
self,
entity: dict,
) -> str:
"""
Build CREATE TABLE SQL for one parse JSON entity.
Parameters
----------
entity : `dict`
Entity definition dict from the parse JSON.
Returns
-------
sql : `str`
Complete ``CREATE TABLE IF NOT EXISTS ...`` statement.
Examples
--------
>>> loader = JSONSchemaLoader()
>>> loader._config = {
... "Parent": {
... "returns_id": "parent_id",
... "parent_id": None,
... "attributes": {"name": {"xml_attr": "N"}},
... "sql": {"table_name": "Parent"},
... },
... }
>>> entity = loader._config["Parent"]
>>> sql = loader._build_entity_table_sql(entity)
>>> "CREATE TABLE IF NOT EXISTS Parent" in sql
True
"""
# get the table name and meta-data/constraints
table_name = entity[ParserNames.sql][ParserNames.tab_name]
parent_id_col: str | None = entity.get(ParserNames.prnt_id)
not_null_cols: set[str] = set(entity.get(ParserNames.meta_nn, []))
unique_constraints: list[list[str]] = entity.get("unique", [])
fk_constraint: str = entity.get("foreign_key_constraint",
ParserNames.meta_default,)
defaults: dict = entity.get("defaults", {})
# ### initiate table by adding the ID column
parts = ["id INTEGER PRIMARY KEY AUTOINCREMENT"]
# ### the id column
# NOTE adding entity_type and entity_id if needed for
# parent / polymorphic columns
if parent_id_col ==ParserNames.enty_id:
parts.append("entity_type TEXT NOT NULL")
parts.append("entity_id INTEGER NOT NULL")
elif parent_id_col:
# if a simple string (evaluated as truthy)
# check if it is allowed to be null or not
nn = " NOT NULL" if parent_id_col in not_null_cols else ""
parts.append(f"{parent_id_col} INTEGER{nn}")
# ### the rest of the columns
# Data columns from attributes
for col_name, col_def in entity.get(ParserNames.attr, {}).items():
# getting column type
_cast = col_def.get(ParserNames.cast)
col_type = self.CAST_TYPE_MAP.get(_cast or "", "TEXT")
# not null constraint
nn = " NOT NULL" if col_name in not_null_cols else ""
# default values
default_val = defaults.get(col_name)
if default_val is None:
default_str = ""
elif isinstance(default_val, str):
default_str = f" DEFAULT '{default_val}'"
else:
default_str = f" DEFAULT {default_val}"
parts.append(f"{col_name} {col_type}{nn}{default_str}")
# ### FOREIGN KEY constraint
# ensure these are deleted from other tables as well - CASCADE
if (
parent_id_col
and (parent_id_col != ParserNames.enty_id)
and (fk_constraint != "none")
):
parent_table = self._find_parent_table(parent_id_col)
cascade_str = (
" ON DELETE CASCADE"
if fk_constraint == "cascade"
else ""
)
parts.append(
f"FOREIGN KEY ({parent_id_col}) "
f"REFERENCES {parent_table}(id)"
f"{cascade_str}"
)
# ### UNIQUE constraints
for unique_cols in unique_constraints:
parts.append(f"UNIQUE({', '.join(unique_cols)})")
return (
f"CREATE TABLE IF NOT EXISTS {table_name} "
f"(" + ", ".join(parts) + ")"
)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
class ClinVarSchemaGenerator(SQLiteParser):
"""
Builder for ClinVar SQLite database schemas.
This class provides methods to create database schemas for
ClinVar RCV (condition-centric), and VCV (variant-centric),
It handles connecting to the SQLite database, executing table and index
creation, and safely closing the connection.
Attributes
----------
conn : `sqlite3.Connection` or `None`
Active SQLite connection, or None if not connected.
cursor : `sqlite3.Cursor` or `None`
Cursor for executing SQL statements, or None if not connected.
Methods
-------
create_schema(db_path, db_config, db_indices=None, name=None)
Create a SQLite schema for a given ClinVar database.
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __init__(self):
"""Initialise the schema builder."""
super().__init__()
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __str__(self) -> str:
"""
Return human-readable string representation.
Returns
-------
str
Human-readable description of the builder
"""
if self.conn:
return f"{type(self).__name__} (active connection)"
return (
f"{type(self).__name__}\n"
"Available methods:\n"
" - create_schema(db_path, db_config, db_indices, name): "
"Create database schema"
)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def create_schema(
self,
db_path: str | Path,
db_config: list[str],
db_indices: list[str] | None,
name: str | None = None,
) -> None:
"""
Create a SQLite schema for ClinVar (RCV, VCV).
Parameters
----------
db_path : `str` or `Path`
Path to SQLite database file.
db_config : `list` [`str`]
List of SQL CREATE TABLE statements.
db_indices : `list` [`str`] or `None`, default `None`
List of SQL CREATE INDEX statements.
name : `str` or `None`, default `None`
Optional name for logging. Defaults to the filename stem of db_path.
Returns
-------
None
"""
is_type(db_config, list)
is_type(db_indices, (type(None), list))
is_type(name, (type(None), str))
is_type(db_path, (Path, str))
# extract the filename from the db_path.
if name is None:
name = Path(db_path).stem
# Initiate logger and connect database
logger.info(f"Creating {name} schema: {db_path}")
with self._connection(db_path):
for sql in db_config:
logger.debug(f"Adding tables: {sql}")
self.cursor.execute(sql)
if db_indices:
for sql in db_indices:
logger.debug(f"Adding indices: {sql}")
self.cursor.execute(sql)
logger.info(f"{name} schema created successfully: {db_path}")
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
def _parse_arguments() -> argparse.ArgumentParser:
"""
Parse command-line arguments for schema generation.
Returns
-------
argparse.ArgumentParser
Configured argument parser.
"""
parser = argparse.ArgumentParser(
description="Generate ClinVar SQLite database schemas"
)
parser.add_argument(
"directory",
type=str,
help="Directory path where database files will be created",
)
parser.add_argument("--rcv", action="store_true", help="Create RCV schema")
parser.add_argument("--vcv", action="store_true", help="Create VCV schema")
parser.add_argument(
"-v",
"--verbose",
action="count",
default=0,
help="Increase verbosity (-v for INFO, -vv for DEBUG)",
)
return parser
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
[docs]
def main():
"""
Command-line interface for schema generation.
Examples
--------
# Create RCV schema
python clinvar_schema.py /data --rcv
# Create create RCV and VCV schemas
python clinvar_schema.py /data --rcv --vcv
"""
parser = _parse_arguments()
args = parser.parse_args()
# Configure logging based on verbosity
configure_logging(args.verbose)
# checking path
_check_directory(args.directory)
_check_directory_readable(args.directory)
# add file names
file_rcv = os.path.join(args.directory, DBNames.rcv)
file_vcv = os.path.join(args.directory, DBNames.vcv)
# resolve config directory (sql/ subdir)
sql_path = os.path.join(check_environ(), "sql")
# initialise loader and builder
loader = JSONSchemaLoader()
builder = ClinVarSchemaGenerator()
if args.rcv:
logger.info("Parsing RCV configuration")
rcv_tables, rcv_indices = loader.load(
Path(os.path.join(sql_path, ParserNames.rcv))
)
builder.create_schema(
db_path=file_rcv,
db_config=rcv_tables,
db_indices=rcv_indices,
)
if args.vcv:
logger.info("Parsing VCV configuration")
vcv_tables, vcv_indices = loader.load(
Path(os.path.join(sql_path, ParserNames.vcv))
)
builder.create_schema(
db_path=file_vcv,
db_config=vcv_tables,
db_indices=vcv_indices,
)
# Finished
logger.info("Schema generation completed successfully")
# Print summary for user
created_schemas = [
(name, path)
for flag, name, path in [
(args.rcv, "RCV", file_rcv),
(args.vcv, "VCV", file_vcv),
]
if flag
]
print("\n" + "=" * 70)
print("SCHEMA GENERATION SUMMARY")
print("=" * 70)
print(f"Output Directory: {args.directory}")
print(f"\nCreated Schemas ({len(created_schemas)}):")
for schema_name, schema_path in created_schemas:
print(f" ✓ {schema_name:12s} → {schema_path}")
print("=" * 70)
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
if __name__ == "__main__":
try:
main()
sys.exit(0)
except KeyboardInterrupt:
print("\nInterrupted by user")
sys.exit(1)
except Exception as e:
print(f"Error: {e}")
sys.exit(1)