refactor: ♻️ commonise REDCap API and JSON logic

src/feasibility_data/common/redcap/__init__.py

@@ -0,0 +1 @@
+"""REDCap common functions."""

src/feasibility_data/common/redcap/api.py

@@ -0,0 +1,60 @@
+import os
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+@dataclass
+class APIConfig:
+    """Configuration for the REDCap API."""
+
+    env_key: str
+    url: str
+
+
+class Center(Enum):
+    """The centers in the study."""
+
+    # TODO: Update these once we can connect them.
+    Aarhus = APIConfig(env_key="REDC_AAR_API_KEY", url="https://redcap.auh.dk/api/")
+    Copenhagen = APIConfig(
+        env_key="REDC_CPH_API_KEY", url="https://redcap.regionh.dk/api/"
+    )
+    Odense = APIConfig(env_key="REDC_ODN_API_KEY", url="https://redcap.sdu.dk/api/")
+    Test = APIConfig(env_key="TEST_API_KEY", url="https://redcap.au.dk/api/")
+
+
+def get_from_redcap(
+    data: dict[str, str],
+    center: Center = Center.Copenhagen,
+) -> requests.Response:
+    """Send a request to the REDCap API."""
+    token = os.environ.get(center.value.env_key)
+    if not token:
+        raise RuntimeError(f"{center.value.env_key} environment variable is not set.")
+
+    data["token"] = token
+
+    response = requests.post(center.value.url, data=data, timeout=60)
+    response.raise_for_status()
+
+    return response
+
+
+def get_json_from_redcap(
+    content: str,
+    center: Center = Center.Copenhagen,
+) -> Any:
+    """Send a request to the REDCap API and return the JSON response."""
+    data = {
+        "content": content,
+        "format": "json",
+        "returnFormat": "json",
+    }
+    response = get_from_redcap(data, center)
+    return response.json()

src/feasibility_data/common/redcap/json.py

@@ -0,0 +1,16 @@
+import json
+from pathlib import Path
+from typing import Any
+
+
+def read_json(path: Path) -> Any:
+    """Read a JSON file and return the data."""
+    with open(path) as f:
+        return json.load(f)
+
+
+def write_json(path: Path, data: Any) -> None:
+    """Write data to a JSON file."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)

src/feasibility_data/data/redcap/__init__.py

@@ -0,0 +1 @@
+"""REDCap data functions."""

src/feasibility_data/data/redcap/raw.py

@@ -1,48 +1,37 @@
-import os
 from datetime import datetime
 from io import StringIO
 from pathlib import Path
 
 import polars as pl
-import requests
-import seedcase_soil as so
-from dotenv import load_dotenv
 
-load_dotenv()
+from feasibility_data.common.redcap.api import Center, get_from_redcap
 
 
-def write_raw() -> Path:
-    """Writes the raw data to `raw/redcap/<timestamp>.csv.gz`."""
-    data = _request_raw()
+def download_redcap_data(
+    raw_data_dir: Path,
+    center: Center,
+) -> None:
+    """Download the data."""
+    response = get_from_redcap(
+        data={
+            "content": "record",
+            "action": "export",
+            "format": "csv",
+            "type": "flat",
+            "csvDelimiter": ";",
+            "rawOrLabel": "raw",
+            "rawOrLabelHeaders": "raw",
+            "exportCheckboxLabel": "false",
+            "exportSurveyFields": "false",
+            "exportDataAccessGroups": "false",
+            "returnFormat": "json",
+        },
+        center=center,
+    )
+    data = response.text
+
     df = pl.read_csv(StringIO(data), separator=";", infer_schema=False)
     timestamp = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
-    file_path = Path("raw") / "redcap" / f"{timestamp}.csv.gz"
-    file_path.parent.mkdir(parents=True, exist_ok=True)
-    df.write_csv(file_path, compression="gzip")
-    so.pretty_print(f"Saved data to '{file_path}'.")
-    return file_path
-
-
-def _request_raw() -> str:
-    """Gets the data from REDCap as CSV."""
-    token = os.environ.get("REDC_CPH_API_KEY")
-    if not token:
-        raise RuntimeError("REDC_CPH_API_KEY environment variable is not set.")
-
-    data = {
-        "token": token,
-        "content": "record",
-        "action": "export",
-        "format": "csv",
-        "type": "flat",
-        "csvDelimiter": ";",
-        "rawOrLabel": "raw",
-        "rawOrLabelHeaders": "raw",
-        "exportCheckboxLabel": "false",
-        "exportSurveyFields": "false",
-        "exportDataAccessGroups": "false",
-        "returnFormat": "json",
-    }
-    response = requests.post("https://redcap.regionh.dk/api/", data=data, timeout=60)
-    response.raise_for_status()
-    return response.text
+    data_path = raw_data_dir / f"{timestamp}.csv.gz"
+    raw_data_dir.parent.mkdir(parents=True, exist_ok=True)
+    df.write_csv(data_path, compression="gzip")

src/feasibility_data/metadata/redcap/__init__.py

@@ -0,0 +1 @@
+"""REDCap metadata functions."""

src/feasibility_data/metadata/redcap/core.py

@@ -1,67 +1,10 @@
-import json
-import os
-from dataclasses import dataclass
-from enum import Enum
 from pathlib import Path
 
-import requests
-from dotenv import load_dotenv
+from feasibility_data.common.redcap.api import get_json_from_redcap
+from feasibility_data.common.redcap.json import write_json
 
-load_dotenv()
 
-
-@dataclass
-class APIConfig:
-    """Configuration for the REDCap API."""
-
-    env_key: str
-    url: str
-
-
-class Center(Enum):
-    """The centers in the study."""
-
-    # TODO: Update these once we can connect them.
-    Aarhus = APIConfig(env_key="REDC_AAR_API_KEY", url="https://redcap.auh.dk/api/")
-    Copenhagen = APIConfig(
-        env_key="REDC_CPH_API_KEY", url="https://redcap.regionh.dk/api/"
-    )
-    Odense = APIConfig(env_key="REDC_ODN_API_KEY", url="https://redcap.sdu.dk/api/")
-
-
-# TODO: Update path to point to pytask build folder once we have one.
-def write_dictionary(
-    raw_dictionary: list[dict[str, str]],
-    path: Path = Path("src/feasibility_data/metadata/redcap/dictionary.json"),
-) -> Path:
-    """Write dictionary from REDCap.
-
-    Args:
-        raw_dictionary: The raw dictionary data from REDCap. Gotten from `request_dictionary()`.
-        path: The path to write the dictionary to.
-
-    Returns:
-        The path to the new dictionary file.
-    """
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w") as f:
-        json.dump(raw_dictionary, f, indent=2, ensure_ascii=False)
-    return path
-
-
-def request_raw_dictionary(center: Center = Center.Copenhagen) -> list[dict[str, str]]:
-    """Gets the data dictionary from REDCap."""
-    token = os.environ.get(center.value.env_key)
-    if not token:
-        raise RuntimeError(f"{center.value.env_key} environment variable is not set.")
-
-    data = {
-        "token": token,
-        "content": "metadata",
-        "format": "json",
-        "returnFormat": "json",
-    }
-    response = requests.post(center.value.url, data=data, timeout=30)
-    response.raise_for_status()
-    dictionary: list[dict[str, str]] = response.json()
-    return dictionary
+def download_redcap_metadata(path: Path, content: str) -> None:
+    """Download data from REDCap and save it to a JSON file."""
+    data = get_json_from_redcap(content)
+    write_json(path, data)