refactor(score-sets): extract CSV export logic into score_set_csv module

src/mavedb/lib/clinvar/constants.py

@@ -1,8 +1,14 @@
 import os
+import re
 from pathlib import Path
 
 from urllib3.util.retry import Retry
 
+CLINVAR_NS_PATTERN = re.compile(r"^clinvar\.(\d+)_(0[1-9]|1[0-2])$")
+"""Pattern for ClinVar-versioned namespaces of the form "clinvar.YEAR_MONTH",
+e.g. "clinvar.2024_01" for January 2024.
+"""
+
 TSV_VARIANT_ARCHIVE_BASE_URL = "https://ftp.ncbi.nlm.nih.gov/pub/clinvar/tab_delimited/archive"
 
 NCBI_REQUEST_HEADERS = {

src/mavedb/lib/clinvar/utils.py

@@ -8,7 +8,7 @@
 import sys
 from datetime import datetime
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Optional
 
 import requests
 from filelock import FileLock
@@ -17,6 +17,7 @@
 from mavedb.lib.clinvar.constants import (
     CLINVAR_CACHE_DIR,
     CLINVAR_FIELDS_TO_KEEP,
+    CLINVAR_NS_PATTERN,
     NCBI_REQUEST_HEADERS,
     NCBI_RETRY_STRATEGY,
     TSV_VARIANT_ARCHIVE_BASE_URL,
@@ -28,6 +29,22 @@
 logger = logging.getLogger(__name__)
 
 
+def parse_clinvar_namespace(ns: str) -> Optional[str]:
+    """Parse a ClinVar-versioned namespace into its db_version string.
+
+    Namespaces are of the form ``"clinvar.YEAR_MONTH"`` (e.g. ``"clinvar.2024_01"``
+    for January 2024). The corresponding ``db_version`` stored in
+    ``clinical_controls`` is ``"MONTH_YEAR"`` (e.g. ``"01_2024"``).
+
+    Returns ``None`` if *ns* does not match the expected pattern.
+    """
+    m = CLINVAR_NS_PATTERN.match(ns)
+    if not m:
+        return None
+    year, month = m.group(1), m.group(2)
+    return f"{month}_{year}"
+
+
 def _ncbi_session() -> requests.Session:
     session = requests.Session()
     session.headers.update(NCBI_REQUEST_HEADERS)

src/mavedb/lib/mave/utils.py

@@ -31,3 +31,12 @@ def is_csv_null(value):
     if value == 0:
         return value
     return not value or NULL_VALUES_RE.fullmatch(str(value).strip().lower())
+
+
+_CSV_OUTPUT_NULL_RE = re.compile(r"\s+|none|nan|na|undefined|n/a|null|nil", flags=re.IGNORECASE)
+
+
+def is_csv_output_null(value):
+    """Return True if a value should be replaced with the NA sentinel in CSV output."""
+    value = str(value).strip().lower()
+    return _CSV_OUTPUT_NULL_RE.fullmatch(value) or not value