Set up environment variables for api clients

Author: “Carmel

.gitignore

@@ -1,6 +1,10 @@
 venv/
+.venv/
+
 new_ldot_workflows/ldot_config.json
 new_ldot_workflows/qualtrics_config.json
+new_ldot_workflows/__pycache__/
+.env
 
 scratch.py
 logs/api_calls.log

app.py

@@ -2,12 +2,14 @@
 from dataclasses import dataclass
 from typing import Optional
 
-from flask import Flask, json, render_template, request, jsonify
+from dotenv import load_dotenv
+from flask import Flask, render_template, request, jsonify
 import requests
 import yaml
-import json
+import os
 
 from new_ldot_workflows.ldot_client import LdotClient
+from new_ldot_workflows.qualtrics_client import QualtricsClient
 from new_ldot_workflows.b_1_get_new_subjects import get_new_subjects
 from new_ldot_workflows.b_2_get_qualtrics_links import add_individuals_to_survey
 from new_ldot_workflows.b_2_send_links_to_ldot import send_links_to_ldot
@@ -22,6 +24,28 @@
 CONFIG_PATH = Path(__file__).resolve().with_name("study_configs.yaml")
 STUDIES = yaml.safe_load(CONFIG_PATH.read_text(encoding="utf-8"))
 
+load_dotenv()
+
+LDOT_TOKEN_URL = os.environ["LDOT_TOKEN_URL"]
+LDOT_API_URL = os.environ["LDOT_API_URL"]
+LDOT_client_id = os.environ["LDOT_client_id"]
+LDOT_client_secret = os.environ["LDOT_client_secret"]
+
+QUALTRICS_BASE_URL = os.environ["QUALTRICS_BASE_URL"]
+QUALTRICS_API_TOKEN = os.environ["QUALTRICS_API_TOKEN"]
+
+
+ldot_client = LdotClient(
+    token_url=LDOT_TOKEN_URL,
+    api_url=LDOT_API_URL,
+    client_id=LDOT_client_id,
+    client_secret=LDOT_client_secret
+)
+
+qualtrics_client = QualtricsClient(
+    api_url=QUALTRICS_BASE_URL,
+    token=QUALTRICS_API_TOKEN
+)
 
 @dataclass
 class StudySettings:
@@ -67,20 +91,6 @@ def get_study_settings(study_key: str):
     )
 
 
-with open("new_ldot_workflows/ldot_config.json") as f:
-    config = json.load(f)
-LDOT_TOKEN_URL = config["LDOT_TOKEN_URL"]
-LDOT_API_URL = config["LDOT_API_URL"]
-CLIENT_ID = config["client_id"]
-CLIENT_SECRET = config["client_secret"]
-
-ldot_client = LdotClient(
-    token_url=LDOT_TOKEN_URL,
-    api_url=LDOT_API_URL,
-    client_id=CLIENT_ID,
-    client_secret=CLIENT_SECRET
-)
-
 @app.route("/")
 def index():
     return render_template("index.html", studies=STUDIES)
@@ -156,6 +166,7 @@ def button2():
     try:
         subject_id_to_link_dict = add_individuals_to_survey(
             ldot_client,
+            qualtrics_client,
             new_subject_ids,
             study_variables.ldot_study_id,
             study_variables.id_deelnemer_entity,
@@ -230,6 +241,7 @@ def button4():
     try:
         participant_to_progress_dict = get_individual_progress(
             ldot_client,
+            qualtrics_client,
             study_variables.ldot_study_id,
             study_variables.id_deelnemer_entity,
             study_variables.id_location,

new_ldot_workflows/__pycache__/b_2_get_qualtrics_links.cpython-311.pyc

@@ -1,24 +1,17 @@
 import requests
-import json
-
 from new_ldot_workflows.logging_utils import logged_request, QualtricsAPIError
 
-with open("new_ldot_workflows/qualtrics_config.json") as f:
-    config = json.load(f)
-QUALTRICS_BASE_URL = config["QUALTRICS_BASE_URL"]
-HEADERS = config["HEADERS"]
-
-def check_contact_in_mailing_list(study_identifier: str, mailing_list_id: str, directory_id: str) -> bool:
+def check_contact_in_mailing_list(qualtrics_client, study_identifier: str, mailing_list_id: str, directory_id: str) -> bool:
     """Find who is already in that mailing list, check if the participant's ID is already there."""
 
     print("Checking for contact in mailing list... ")
     try:
         response = logged_request(
             "GET",
-            f"{QUALTRICS_BASE_URL}/directories/{directory_id}/mailinglists/{mailing_list_id}/contacts",
+            f"{qualtrics_client.api_url}/directories/{directory_id}/mailinglists/{mailing_list_id}/contacts",
             function_name="check_contact_in_mailing_list",
             service="Qualtrics",
-            headers=HEADERS,
+            headers=qualtrics_client.headers,
             raise_for_status=True,
         )
         contacts =  response.json()["result"]["elements"]
@@ -37,7 +30,7 @@ def check_contact_in_mailing_list(study_identifier: str, mailing_list_id: str, d
         return False
     return True
 
-def add_contact_to_mailing_list(study_identifier: str, embedded_data_field: str, mailing_list_id: str, directory_id: str) -> None:
+def add_contact_to_mailing_list(qualtrics_client, study_identifier: str, embedded_data_field: str, mailing_list_id: str, directory_id: str) -> None:
     """Add contact to a mailing list"""
 
     contact_information_payload ={
@@ -49,10 +42,10 @@ def add_contact_to_mailing_list(study_identifier: str, embedded_data_field: str,
     try:
         response = logged_request(
             "POST",
-            f"{QUALTRICS_BASE_URL}/directories/{directory_id}/mailinglists/{mailing_list_id}/contacts",
+            f"{qualtrics_client.api_url}/directories/{directory_id}/mailinglists/{mailing_list_id}/contacts",
             function_name="add_contact_to_mailing_list",
             service="Qualtrics",
-            headers=HEADERS,
+            headers=qualtrics_client.get_headers(),
             json=contact_information_payload,
             raise_for_status=True,
         )
@@ -66,7 +59,7 @@ def add_contact_to_mailing_list(study_identifier: str, embedded_data_field: str,
             f"Unexpected response format when adding contact to mailing list."
         ) from exc
 
-def get_personal_link(qualtrics_survey_id: str, distribution_id: str, study_identifier: str) -> str:
+def get_personal_link(qualtrics_client, qualtrics_survey_id: str, distribution_id: str, study_identifier: str) -> str:
     """Get the person link of an individual for a specified survey. Returns personal link for the participant, looks like
         https://survey.uu.nl/jfe/form/SV_efCMOg6wHU0T8ii?Q_CHLqe2Pdrma&_g_=g
     """
@@ -77,10 +70,10 @@ def get_personal_link(qualtrics_survey_id: str, distribution_id: str, study_iden
     try:
         response = logged_request(
             "GET",
-            f"{QUALTRICS_BASE_URL}/distributions/{distribution_id}/links",
+            f"{qualtrics_client.api_url}/distributions/{distribution_id}/links",
             function_name="get_personal_link",
             service="Qualtrics",
-            headers=HEADERS,
+            headers=qualtrics_client.headers,
             params=parameters,
             raise_for_status=True,
         )
@@ -97,7 +90,7 @@ def get_personal_link(qualtrics_survey_id: str, distribution_id: str, study_iden
             f"Unexpected response format when fetching personal link of individual {study_identifier}."
         ) from exc    
 
-def add_individuals_to_survey(ldot_client, new_subject_ids: list, ldot_study_id: str, id_deelnemer_entity: str, embedded_data_field: str, distribution_id: str, qualtrics_survey_id: str, mailing_list_id: str, directory_id: str):
+def add_individuals_to_survey(ldot_client, qualtrics_client, new_subject_ids: list, ldot_study_id: str, id_deelnemer_entity: str, embedded_data_field: str, distribution_id: str, qualtrics_survey_id: str, mailing_list_id: str, directory_id: str):
     def convert_api_subject_id_to_study_identifier(api_subject_id):
         """Convert subject ID used by the Ldot API to study identifier that can be used in Qualtrics surveys"""
 
@@ -122,15 +115,15 @@ def convert_api_subject_id_to_study_identifier(api_subject_id):
         study_identifier = convert_api_subject_id_to_study_identifier(subject_id)
         print(f"Processing subject ID: {subject_id} (Study Identifier: {study_identifier})")
 
-        contact_exists = check_contact_in_mailing_list(study_identifier, mailing_list_id, directory_id)
+        contact_exists = check_contact_in_mailing_list(qualtrics_client, study_identifier, mailing_list_id, directory_id)
 
         if not contact_exists:
             print(f"Adding participant study ID {study_identifier} to mailing list...")
-            add_contact_to_mailing_list(study_identifier, embedded_data_field, mailing_list_id, directory_id)
+            add_contact_to_mailing_list(qualtrics_client, study_identifier, embedded_data_field, mailing_list_id, directory_id)
         else:
             print(f"Participant study ID {study_identifier} already exists in this mailing list")
 
-        personal_link = get_personal_link(qualtrics_survey_id, distribution_id, study_identifier)
+        personal_link = get_personal_link(qualtrics_client, qualtrics_survey_id, distribution_id, study_identifier)
 
         participant_to_link_dict[subject_id] = personal_link
 
@@ -147,9 +140,9 @@ def convert_api_subject_id_to_study_identifier(api_subject_id):
     distribution_id="EMD_7AEa416lRwFhrkF"
     directory_id="POOL_10pyxk9leSUisrT"
 
-    link = add_individuals_to_survey(participant_study_id, ldot_study_id, id_deelnemer_entity, embedded_data_field, distribution_id, qualtrics_survey_id, mailing_list_id, directory_id)
+    # link = add_individuals_to_survey(ldot_client, qualtrics_client, participant_study_id, ldot_study_id, id_deelnemer_entity, embedded_data_field, distribution_id, qualtrics_survey_id, mailing_list_id, directory_id)
 
-    print(link)
+    # print(link)
 
 
     # print(get_mailing_list_id_of_distribution("SV_efCMOg6wHU0T8ii",  "EMD_SZFeoK7LAJBHU4d"))

new_ldot_workflows/__pycache__/b_4_get_individual_progress.cpython-311.pyc

@@ -6,14 +6,8 @@
 import json
 
 from new_ldot_workflows.logging_utils import QualtricsAPIError, logged_request
-# from ldot_client import LdotClient
 
-with open("new_ldot_workflows/qualtrics_config.json") as f:
-    config = json.load(f)
-QUALTRICS_BASE_URL = config["QUALTRICS_BASE_URL"]
-HEADERS = config["HEADERS"]
-
-def create_data_export(survey_id: str, incomplete_bool: bool = True) -> str:
+def create_data_export(qualtrics_client, survey_id: str, incomplete_bool: bool = True) -> str:
     """Create a data export request for a given survey.
 
     Args:
@@ -32,10 +26,10 @@ def create_data_export(survey_id: str, incomplete_bool: bool = True) -> str:
     try:
         response = logged_request(
             "POST",
-            f"{QUALTRICS_BASE_URL}/surveys/{survey_id}/export-responses",
+            f"{qualtrics_client.api_url}/surveys/{survey_id}/export-responses",
             function_name="create_data_export",
             service="Qualtrics",
-            headers=HEADERS,
+            headers=qualtrics_client.headers,
             json=payload,
             raise_for_status=True,
         )
@@ -51,10 +45,11 @@ def create_data_export(survey_id: str, incomplete_bool: bool = True) -> str:
         ) from exc
 
 
-def check_export_progress(request_id: str, survey_id: str, returns: str) -> str|float:
+def check_export_progress(qualtrics_client, request_id: str, survey_id: str, returns: str) -> str|float:
     """Check if a an export request is complete
 
     Args:
+        qualtrics_client: The Qualtrics API client.
         request_id (str): The ID of the export request.
         survey_id (str): The ID of the survey that was exported.
         returns (str): The type of information to return ("percent_complete" or "fileID").
@@ -68,10 +63,10 @@ def check_export_progress(request_id: str, survey_id: str, returns: str) -> str|
     try:
         result = logged_request(
             "GET",
-            f"{QUALTRICS_BASE_URL}/surveys/{survey_id}/export-responses/{request_id}",
+            f"{qualtrics_client.api_url}/surveys/{survey_id}/export-responses/{request_id}",
             function_name="check_export_progress",
             service="Qualtrics",
-            headers=HEADERS,
+            headers=qualtrics_client.headers,
             raise_for_status=True,
         )
 
@@ -91,10 +86,11 @@ def check_export_progress(request_id: str, survey_id: str, returns: str) -> str|
         ) from exc
 
 
-def download_export(file_id: str, survey_id: str) -> pd.DataFrame:
+def download_export(qualtrics_client, file_id: str, survey_id: str) -> pd.DataFrame:
     """Download the exported survey data and convert to DataFrame.
 
     Args:
+        qualtrics_client: The Qualtrics API client.
         file_id (str): The ID of the file to download.
         survey_id (str): The ID of the survey associated with the file.
     Raises:
@@ -106,10 +102,10 @@ def download_export(file_id: str, survey_id: str) -> pd.DataFrame:
     try:
         download = logged_request(
             "GET",
-            f"{QUALTRICS_BASE_URL}/surveys/{survey_id}/export-responses/{file_id}/file",
+            f"{qualtrics_client.api_url}/surveys/{survey_id}/export-responses/{file_id}/file",
             function_name="download_export",
             service="Qualtrics",
-            headers=HEADERS,
+            headers=qualtrics_client.headers,
             stream=True,
             raise_for_status=True,
         )
@@ -155,28 +151,29 @@ def check_inputs_validity(participant_study_id: str, embedded_data_field: str, s
     return True
 
 
-def get_responses_as_df(survey_id: str, incomplete_bool: bool) -> pd.DataFrame:
+def get_responses_as_df(qualtrics_client, survey_id: str, incomplete_bool: bool) -> pd.DataFrame:
     """Wrapper function to request export, wait for it, download it, and convert to DataFrame.
 
     Args:
+        qualtrics_client: The Qualtrics API client.
         survey_id (str): The ID of the survey to get responses from.
         incomplete_bool (bool): Whether to get responses that are in progress or completed.
 
     Returns:
         pd.DataFrame: The survey response data as a DataFrame.
     """
     # Request a data export
-    request_id = create_data_export(survey_id, incomplete_bool=incomplete_bool)
+    request_id = create_data_export(qualtrics_client, survey_id, incomplete_bool=incomplete_bool)
 
     # Ping the export report repeatedly until it's ready
     export_progress = 0.0
     while export_progress < 100:
-        export_progress = check_export_progress(request_id, survey_id, returns="percent_complete")
+        export_progress = check_export_progress(qualtrics_client, request_id, survey_id, returns="percent_complete")
         time.sleep(2)
 
     # When export is complete, get the file ID
-    file_id = check_export_progress(request_id, survey_id, returns="fileID")
-    df = download_export(file_id, survey_id)
+    file_id = check_export_progress(qualtrics_client, request_id, survey_id, returns="fileID")
+    df = download_export(qualtrics_client, file_id, survey_id)
     return df
 
 
@@ -206,16 +203,16 @@ def subject_id_to_study_identifier(ldot_client, ldot_study_id: str, id_deelnemer
     return subject_id_to_study_identifier_dict
 
 
-def get_individual_progress(ldot_client, ldot_study_id: str, id_deelnemer_entity: str, id_location: str, subject_ids: list, embedded_data_field: str, survey_id: str) -> float:
+def get_individual_progress(ldot_client, qualtrics_client,ldot_study_id: str, id_deelnemer_entity: str, id_location: str, subject_ids: list, embedded_data_field: str, survey_id: str) -> float:
     """Wrapper function that gets the individual progress of a participant in a survey."""
     # Check for invalid inputs
     check_inputs_validity(subject_ids[0], embedded_data_field, survey_id) # TODO: Check this
 
     participant_to_progress_dict = {}
     subject_id_to_study_identifier_dict = subject_id_to_study_identifier(ldot_client, ldot_study_id, id_deelnemer_entity, id_location, subject_ids)
 
-    completed_responses_df = get_responses_as_df(survey_id, incomplete_bool=False)
-    incompleted_responses_df = get_responses_as_df(survey_id, incomplete_bool=True)
+    completed_responses_df = get_responses_as_df(qualtrics_client, survey_id, incomplete_bool=False)
+    incompleted_responses_df = get_responses_as_df(qualtrics_client, survey_id, incomplete_bool=True)
     df = pd.concat([completed_responses_df, incompleted_responses_df], ignore_index=True)
 
     if not embedded_data_field in df.columns:
@@ -233,20 +230,21 @@ def get_individual_progress(ldot_client, ldot_study_id: str, id_deelnemer_entity
     return participant_to_progress_dict
 
 if __name__ == "__main__":
-    with open("ldot_config.json") as f:
-        config = json.load(f)
-    LDOT_TOKEN_URL = config["LDOT_TOKEN_URL"]
-    LDOT_API_URL = config["LDOT_API_URL"]
-    CLIENT_ID = config["client_id"]
-    CLIENT_SECRET = config["client_secret"]
-
-    ldot_client = LdotClient(
-        token_url=LDOT_TOKEN_URL,
-        api_url=LDOT_API_URL,
-        client_id=CLIENT_ID,
-        client_secret=CLIENT_SECRET
-    )
-    print(ldot_client.headers)
+    pass
+    # with open("ldot_config.json") as f:
+    #     config = json.load(f)
+    # LDOT_TOKEN_URL = config["LDOT_TOKEN_URL"]
+    # LDOT_API_URL = config["LDOT_API_URL"]
+    # CLIENT_ID = config["client_id"]
+    # CLIENT_SECRET = config["client_secret"]
+
+    # ldot_client = LdotClient(
+    #     token_url=LDOT_TOKEN_URL,
+    #     api_url=LDOT_API_URL,
+    #     client_id=CLIENT_ID,
+    #     client_secret=CLIENT_SECRET
+    # )
+    # print(ldot_client.headers)
 
     # # # Example usage
     # survey_id="SV_efCMOg6wHU0T8ii"

new_ldot_workflows/b_2_get_qualtrics_links.py

@@ -0,0 +1,17 @@
+import requests
+from datetime import datetime, timedelta
+
+class QualtricsClient:
+    def __init__(self, api_url, token):
+        self.api_url = api_url
+        self.api_token = token
+
+        # Initialize headers with the current token (if any)
+        self.headers = self.get_headers()
+
+    def get_headers(self):
+        return {
+            "Content-Type": "application/json",
+            "X-API-TOKEN": self.api_token
+        }
+