Make Authenticated Requests if Envvars are Set

Author: bencap

settings/.env.dev

@@ -19,6 +19,13 @@ POSTGRES_DB=gene_normalizer
 
 UTA_DB_URL=postgresql://anonymous:[email protected]:5432/uta/uta_20180821
 
+####################################################################################################
+# Environment variables for MaveDB connection
+####################################################################################################
+
+MAVEDB_BASE_URL=http://localhost:8000
+MAVEDB_API_KEY=
+
 ####################################################################################################
 # Environment variables for seqrepo
 ####################################################################################################

src/dcd_mapping/mavedb_data.py

@@ -15,7 +15,9 @@
 
 from dcd_mapping.resource_utils import (
     LOCAL_STORE_PATH,
+    MAVEDB_BASE_URL,
     ResourceAcquisitionError,
+    authentication_header,
     http_download,
 )
 from dcd_mapping.schemas import ScoreRow, ScoresetMetadata, UniProtRef
@@ -42,7 +44,11 @@ def get_scoreset_urns() -> set[str]:
 
     :return: set of URN strings
     """
-    r = requests.get("https://api.mavedb.org/api/v1/experiments/", timeout=30)
+    r = requests.get(
+        f"{MAVEDB_BASE_URL}/api/v1/experiments/",
+        timeout=30,
+        headers=authentication_header(),
+    )
     r.raise_for_status()
     scoreset_urn_lists = [
         experiment["scoreSetUrns"]
@@ -78,7 +84,11 @@ def get_human_urns() -> list[str]:
     scoreset_urns = get_scoreset_urns()
     human_scoresets: list[str] = []
     for urn in scoreset_urns:
-        r = requests.get(f"https://api.mavedb.org/api/v1/score-sets/{urn}", timeout=30)
+        r = requests.get(
+            f"{MAVEDB_BASE_URL}/api/v1/score-sets/{urn}",
+            timeout=30,
+            headers=authentication_header(),
+        )
         try:
             r.raise_for_status()
         except requests.exceptions.HTTPError:
@@ -127,8 +137,8 @@ def get_raw_scoreset_metadata(
         dcd_mapping_dir = LOCAL_STORE_PATH
     metadata_file = dcd_mapping_dir / f"{scoreset_urn}_metadata.json"
     if not metadata_file.exists():
-        url = f"https://api.mavedb.org/api/v1/score-sets/{scoreset_urn}"
-        r = requests.get(url, timeout=30)
+        url = f"{MAVEDB_BASE_URL}/api/v1/score-sets/{scoreset_urn}"
+        r = requests.get(url, timeout=30, headers=authentication_header())
         try:
             r.raise_for_status()
         except requests.HTTPError as e:
@@ -246,7 +256,7 @@ def get_scoreset_records(
         if urn == "urn:mavedb:00000053-a-1":
             _get_experiment_53_scores(scores_csv, silent)
         else:
-            url = f"https://api.mavedb.org/api/v1/score-sets/{urn}/scores"
+            url = f"{MAVEDB_BASE_URL}/api/v1/score-sets/{urn}/scores"
             try:
                 http_download(url, scores_csv, silent)
             except requests.HTTPError as e:

src/dcd_mapping/resource_utils.py

@@ -6,6 +6,9 @@
 import requests
 from tqdm import tqdm
 
+MAVEDB_API_KEY = os.environ.get("MAVEDB_API_KEY")
+MAVEDB_BASE_URL = os.environ.get("MAVEDB_BASE_URL")
+
 LOCAL_STORE_PATH = Path(
     os.environ.get(
         "DCD_MAPPING_RESOURCES_DIR", Path.home() / ".local" / "share" / "dcd_mapping"
@@ -19,6 +22,11 @@ class ResourceAcquisitionError(Exception):
     """Raise when resource acquisition fails."""
 
 
+def authentication_header() -> dict | None:
+    """Fetch with api key envvar, if available."""
+    return {"X-API-key": MAVEDB_API_KEY} if MAVEDB_API_KEY is not None else None
+
+
 def http_download(url: str, out_path: Path, silent: bool = True) -> Path:
     """Download a file via HTTP.
 
@@ -30,7 +38,9 @@ def http_download(url: str, out_path: Path, silent: bool = True) -> Path:
     """
     if not silent:
         click.echo(f"Downloading {out_path.name} to {out_path.parents[0].absolute()}")
-    with requests.get(url, stream=True, timeout=30) as r:
+    with requests.get(
+        url, stream=True, timeout=30, headers=authentication_header()
+    ) as r:
         r.raise_for_status()
         total_size = int(r.headers.get("content-length", 0))
         with out_path.open("wb") as h: