👔 Update records in REDCap if already exist.

Author: shnizzedy

python_jobs/src/hbnmigration/_config_variables/redcap_variables/__init__.py

@@ -7,10 +7,18 @@
 )
 """Ripple endpoints."""
 
+headers = ImportWithFallback.literal(".redcap_variables", "headers", {})
+"""REDCap headers."""
+
 Tokens = ImportWithFallback.literal(".redcap_variables", "Tokens", NotImplemented)
 """REDCap API tokens."""
 
 redcap_import_file = ImportWithFallback.literal(
     ".redcap_variables", "redcap_import_file", NotImplemented
 )
 """Path to REDCap import data."""
+
+redcap_update_file = ImportWithFallback.literal(
+    ".redcap_variables", "redcap_update_file", NotImplemented
+)
+"""Path to REDCap update data."""

python_jobs/src/hbnmigration/_config_variables/redcap_variables/__init__.pyi

@@ -1,5 +1,11 @@
 """Type stubs for REDCap variables."""
 
-from .redcap_variables import Endpoints, redcap_import_file, Tokens
+from .redcap_variables import (
+    Endpoints,
+    headers,
+    redcap_import_file,
+    redcap_update_file,
+    Tokens,
+)
 
-__all__ = ["Endpoints", "Tokens", "redcap_import_file"]
+__all__ = ["Endpoints", "Tokens", "headers", "redcap_import_file", "redcap_update_file"]

python_jobs/src/hbnmigration/_config_variables/redcap_variables/redcap_variables.pyi

@@ -10,9 +10,12 @@ class Endpoints(EndpointsABC):
     @property
     def base_url(self) -> str: ...
 
+headers: dict[str, str]
+
 @dataclass
 class Tokens:
     pid247: str
     pid757: str
 
 redcap_import_file: Path
+redcap_update_file: Path

python_jobs/src/hbnmigration/_config_variables/ripple_variables/__init__.py

@@ -15,3 +15,7 @@
 
 study_ids = ImportWithFallback.literal(".ripple_variables", "study_ids", {})
 """Ripple study IDs."""
+
+ripple_import_file = ImportWithFallback.literal(
+    ".ripple_variables", "ripple_import_file", NotImplemented
+)

python_jobs/src/hbnmigration/from_ripple/to_redcap.py

@@ -3,15 +3,15 @@
 from dataclasses import dataclass
 import logging
 from pathlib import Path
-from typing import Literal
+from typing import Literal, Optional
 
 import numpy as np
 import pandas as pd
 import requests
 
 from .._config_variables import redcap_variables, ripple_variables
 from ..exceptions import NoData
-from ..utility_functions import initialize_logging, yesterday
+from ..utility_functions import fetch_api_data, initialize_logging, yesterday
 
 initialize_logging()
 logger = logging.getLogger(__name__)
@@ -125,8 +125,58 @@ def prepare_redcap_data(df: pd.DataFrame) -> None:
     """Prepare Ripple API returned data to be imported into REDCap."""
     copy_selected_redcap_df = set_redcap_columns(df)
 
-    # Save the new dataframe to a CSV file
-    copy_selected_redcap_df.to_csv(redcap_variables.redcap_import_file, index=False)
+    # Split into update and new
+    to_update, new_subjects = get_redcap_subjects_to_update(copy_selected_redcap_df)
+
+    # Save the new dataframes to CSV files
+    if not to_update.empty:
+        to_update.to_csv(redcap_variables.redcap_update_file, index=False)
+    if not new_subjects.empty:
+        new_subjects.to_csv(redcap_variables.redcap_import_file, index=False)
+
+
+def get_redcap_subjects_to_update(
+    df: pd.DataFrame,
+) -> tuple[pd.DataFrame, pd.DataFrame]:
+    """
+    Get subjects from REDCap that need to be updated vs. new subjects.
+
+    Returns
+    -------
+    pd.DataFrame
+        subjects to update
+    pd.DataFrame
+        new subjects
+
+    """
+    redcap_participant_consent_data = {
+        "token": redcap_variables.Tokens.pid247,
+        "content": "record",
+        "action": "export",
+        "format": "csv",
+        "type": "flat",
+        "csvDelimiter": "",
+        "fields": "mrn,record_id",
+        "rawOrLabel": "raw",
+        "rawOrLabelHeaders": "raw",
+        "exportCheckboxLabel": "false",
+        "exportSurveyFields": "false",
+        "exportDataAccessGroups": "false",
+        "returnFormat": "csv",
+    }
+    df_redcap_consent_instruments = fetch_api_data(
+        Endpoints.REDCap.base_url,
+        redcap_variables.headers,
+        redcap_participant_consent_data,
+    )
+    to_update = df[df["mrn"].isin(df_redcap_consent_instruments["mrn"])].copy()
+    to_update = to_update.drop(columns=["record_id"]).merge(
+        df_redcap_consent_instruments[["mrn", "record_id"]], on="mrn", how="left"
+    )
+    to_update = to_update[["record_id", "mrn", "email_consent"]]
+    return to_update[to_update["mrn"].isin(df_redcap_consent_instruments["mrn"])], df[
+        ~df["mrn"].isin(df_redcap_consent_instruments["mrn"])
+    ]
 
 
 def prepare_ripple_to_ripple(df: pd.DataFrame) -> dict[str, str]:
@@ -160,17 +210,26 @@ def prepare_ripple_to_ripple(df: pd.DataFrame) -> dict[str, str]:
     return ripple_import_files
 
 
-def push_to_redcap(project_token: str) -> None:
+def push_to_redcap(project_token: str, update: Optional[bool] = None) -> None:
     """Push the HBN Potential Participants MRN and Contact email to RedCap."""
-    # Read the content of the CSV file
+    if update is None:
+        # Try both
+        for _update in [True, False]:
+            push_to_redcap(project_token, _update)
+        return
+    filepath = (
+        redcap_variables.redcap_update_file
+        if update
+        else redcap_variables.redcap_import_file
+    )
+    if not filepath.exists():
+        return
     try:
-        with open(redcap_variables.redcap_import_file, "r") as file:
+        # Read the content of the CSV file
+        with open(filepath, "r") as file:
             csv_content = file.read()
     except FileNotFoundError:
-        logger.exception(
-            "Error: The file '%s' was not found.", redcap_variables.redcap_import_file
-        )
-        # You might want to create a placeholder or exit
+        logger.exception("Error: The file '%s' was not found.", filepath)
         csv_content = ""  # Default to empty string if file not found
     if csv_content:
         data = {
@@ -180,7 +239,7 @@ def push_to_redcap(project_token: str) -> None:
             "format": "csv",
             "type": "flat",
             "overwriteBehavior": "normal",
-            "forceAutoNumber": "true",
+            "forceAutoNumber": str(not update).lower,
             "data": csv_content,
             "returnContent": "count",
             "returnFormat": "csv",
@@ -240,11 +299,12 @@ def set_status_in_ripple(ripple_study: str, ripple_import_file: str) -> None:
         raise
 
 
-def cleanup() -> None:
+def cleanup(temp_files: list[str | Path]) -> None:
     """Delete temporary files."""
     for filepath in [
         redcap_variables.redcap_import_file,
-        ripple_variables.ripple_import_file,
+        redcap_variables.redcap_update_file,
+        *[Path(filepath) for filepath in temp_files],
     ]:
         try:
             filepath.unlink(missing_ok=True)
@@ -258,6 +318,7 @@ def main(project_status: Literal["dev", "prod"] = "prod") -> None:
         "dev": {"token": redcap_variables.Tokens.pid757},
         "prod": {"token": redcap_variables.Tokens.pid247},
     }
+    ripple_import_files: dict[str, str] = {}
     try:
         filtered_ripple_df = request_potential_participants()
         prepare_redcap_data(filtered_ripple_df)
@@ -268,7 +329,7 @@ def main(project_status: Literal["dev", "prod"] = "prod") -> None:
     except NoData:
         pass
     finally:
-        cleanup()
+        cleanup(list(ripple_import_files.values()))
 
 
 if __name__ == "__main__":