refactor: ♻️ download field metadata with pytask

pyproject.toml

@@ -25,6 +25,7 @@ license-files = ["LICENSE-MIT.md"]
 requires-python = ">=3.12"
 dependencies = [
   "polars>=1.42.1",
+  "pytask>=0.6.0",
   "python-dotenv>=1.2.2",
   "requests>=2.32.5",
   "seedcase-flower>=0.30.1",
@@ -44,8 +45,16 @@ extend = ".config/ruff.toml"
 python_version = 3.12
 strict = true
 
+[tool.pytask.ini_options]
+paths = ["src/feasibility_data"]
+task_files = ["task_*.py", "build.py"]
+
 [dependency-groups]
 dev = [
   "mypy>=2.1.0",
   "types-requests>=2.32.4.20260107",
 ]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

src/feasibility_data/build.py

@@ -0,0 +1,21 @@
+from importlib.resources import files
+from pathlib import Path
+from typing import Annotated
+
+from pytask import Product
+
+from feasibility_data.metadata.redcap.core import (
+    download_redcap_metadata,
+)
+
+SRC = Path(str(files("feasibility_data"))).joinpath("..").resolve()
+BLD = SRC.joinpath("..", "bld").resolve()
+
+BLD_REDCAP = BLD / "redcap"
+
+
+def task_download_field_metadata(
+    field_metadata_path: Annotated[Path, Product] = BLD_REDCAP / "field_metadata.json",
+) -> None:
+    """Download field metadata to `BLD_REDCAP`."""
+    download_redcap_metadata(field_metadata_path, "metadata")

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,10 @@
+import json
+from pathlib import Path
+from typing import Any
+
+
+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)