Skip to content

Commit 9d0c0ff

Browse files
author
Suchard
committed
Add ready functions that people to survey and get progress. Add requirements.
0 parents  commit 9d0c0ff

5 files changed

Lines changed: 249 additions & 0 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
get_individual_progress.py

‎add_individual_to_survey.py‎

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
import requests
2+
from qualtrics_settings import BASE_URL, HEADERS, SURVEYIDS, QualtricsAPIError
3+
4+
def get_mailing_list_id_of_distribution(survey_id: str, distribution_id: str) -> str:
5+
"""Get the mailing list ID connected to a distribution
6+
7+
Args:
8+
survey_id (str): The ID of the survey to add someone to.
9+
distribution_id (str): The ID of the distribution to get the mailing list ID for.
10+
11+
Raises:
12+
QualtricsAPIError: If there is an error while fetching the mailing list ID or format is unexpected.
13+
14+
Returns:
15+
str: The mailing list ID connected to the distribution.
16+
"""
17+
print("Finding mailing list ID of distribution... ")
18+
endpoint = f"{BASE_URL}/distributions/{distribution_id}"
19+
parameters = {
20+
"surveyId": survey_id
21+
}
22+
try:
23+
response = requests.get(endpoint, headers=HEADERS, params=parameters)
24+
response.raise_for_status()
25+
mailing_list_id = response.json()["result"]["recipients"]["mailingListId"]
26+
27+
return mailing_list_id
28+
except requests.exceptions.RequestException as exc:
29+
raise QualtricsAPIError(
30+
f"Error while fetching mailing list ID for distribution {distribution_id}."
31+
) from exc
32+
except (KeyError, TypeError, ValueError) as exc:
33+
raise QualtricsAPIError(
34+
f"Unexpected response format when fetching mailing list ID for distribution {distribution_id}."
35+
) from exc
36+
37+
def get_directory_id() -> str:
38+
"""Get the directory ID for a Qualtrics, assuming there is just one.
39+
40+
Raises:
41+
QualtricsAPIError: If there is an error while fetching the directory ID or format is unexpected.
42+
43+
Returns:
44+
str: The directory ID for the Qualtrics account.
45+
"""
46+
print("Getting directory ID... ")
47+
endpoint = f"{BASE_URL}/directories"
48+
parameters = {
49+
"includeCount": False
50+
}
51+
try:
52+
response = requests.get(endpoint, headers=HEADERS, params=parameters)
53+
response.raise_for_status()
54+
data = response.json()
55+
directory_id = data["result"]["elements"][0]["directoryId"]
56+
57+
return directory_id
58+
except requests.exceptions.RequestException as exc:
59+
raise QualtricsAPIError(
60+
f"Error while getting the directory ID."
61+
) from exc
62+
except (KeyError, TypeError, ValueError) as exc:
63+
raise QualtricsAPIError(
64+
f"Unexpected response format when fetching directory ID."
65+
) from exc
66+
67+
def check_contact_in_mailing_list(participant_study_id: str, mailing_list_id: str, directory_id: str) -> bool:
68+
"""Find who is already in that mailing list, check if the participant's ID is already there.
69+
70+
Args:
71+
participant_study_id (str): ID of the participant to check if they are in the mailing list.
72+
mailing_list_id (str): The ID of the mailing list to check.
73+
directory_id (str): The ID of the directory containing the mailing list.
74+
75+
Returns:
76+
bool: True if the participant's ID is already in the mailing list, False otherwise.
77+
"""
78+
print("Checking for contact in mailing list... ")
79+
endpoint = f"{BASE_URL}/directories/{directory_id}/mailinglists/{mailing_list_id}/contacts"
80+
try:
81+
response = requests.get(endpoint, headers=HEADERS)
82+
response.raise_for_status()
83+
contacts = response.json()["result"]["elements"]
84+
except requests.exceptions.RequestException as exc:
85+
raise QualtricsAPIError(
86+
f"Error while checking for contact in mailing list."
87+
) from exc
88+
except (KeyError, TypeError, ValueError) as exc:
89+
raise QualtricsAPIError(
90+
f"Unexpected response format when checking for contact in mailing list."
91+
) from exc
92+
93+
contact_external_data_references = [contact["extRef"] for contact in contacts if "extRef" in contact]
94+
95+
96+
if participant_study_id not in contact_external_data_references:
97+
return False
98+
return True
99+
100+
def add_contact_to_mailing_list(participant_study_id: str, mailing_list_id: str, directory_id: str, embedded_data_field: str) -> None:
101+
"""Add contact to a mailing list
102+
103+
Args:
104+
participant_study_id (str): ID of the participant whose progress you want to add.
105+
mailing_list_id (str): Mailing list to add to.
106+
directory_id (str): Directory containing the mailing list.
107+
embedded_data_field (str): The field name in Qualtrics where the participant_study_id is stored.
108+
109+
Raises:
110+
QualtricsAPIError: If there is an error while adding the contact or format is unexpected.
111+
"""
112+
print("Adding contact to mailing list... ")
113+
endpoint = f"{BASE_URL}/directories/{directory_id}/mailinglists/{mailing_list_id}/contacts"
114+
contact_information_payload ={
115+
"extRef": participant_study_id,
116+
"embeddedData": {embedded_data_field: participant_study_id}
117+
}
118+
try:
119+
response = requests.post(endpoint, headers=HEADERS, json=contact_information_payload)
120+
response.raise_for_status()
121+
122+
except requests.exceptions.RequestException as exc:
123+
raise QualtricsAPIError(
124+
f"Error while adding contact to mailing list."
125+
) from exc
126+
except (KeyError, TypeError, ValueError) as exc:
127+
raise QualtricsAPIError(
128+
f"Unexpected response format when adding contact to mailing list."
129+
) from exc
130+
131+
def get_personal_link(survey_id: str, distribution_id: str, participant_study_id: str) -> str:
132+
"""Get the person link of an individual for a specified survey.
133+
134+
Args:
135+
survey_id (str): The ID of the survey to get the personal link for.
136+
distribution_id (str): The ID of the distribution of personal links.
137+
participant_study_id (str): The ID of the participant whose.
138+
139+
Returns:
140+
str: Personal link for the participant, looks like
141+
https://survey.uu.nl/jfe/form/SV_efCMOg6wHU0T8ii?Q_CHLqe2Pdrma&_g_=g
142+
"""
143+
print("Fetching personal link... ")
144+
endpoint = f"{BASE_URL}/distributions/{distribution_id}/links"
145+
parameters = {
146+
"surveyId": str(survey_id)
147+
}
148+
try:
149+
response = requests.get(endpoint, headers=HEADERS, params=parameters)
150+
response.raise_for_status()
151+
data = response.json()
152+
personal_link = [item["link"] for item in data["result"]["elements"] if ("externalDataReference" in item) and (item["externalDataReference"] == participant_study_id)][0]
153+
154+
return personal_link
155+
except requests.exceptions.RequestException as exc:
156+
raise QualtricsAPIError(
157+
f"Error while fetching personal link of individual {participant_study_id}."
158+
) from exc
159+
except (KeyError, TypeError, ValueError) as exc:
160+
raise QualtricsAPIError(
161+
f"Unexpected response format when fetching personal link of individual {participant_study_id}."
162+
) from exc
163+
164+
def check_inputs_validity(participant_study_id: str, embedded_data_field: str, survey_id: str, distribution_id: str) -> bool:
165+
print("Checking input validity... ")
166+
if not isinstance(participant_study_id, str) or participant_study_id.strip() == "":
167+
raise ValueError("Invalid participant_study_id. It should be a non-empty string.")
168+
169+
if not isinstance(embedded_data_field, str) or embedded_data_field.strip() == "":
170+
raise ValueError("Invalid embedded_data_field. It should be a non-empty string.")
171+
172+
if not isinstance(survey_id, str) or not survey_id.startswith("SV_"):
173+
raise ValueError("Invalid survey_id. It should be a string starting with 'SV_'.")
174+
175+
if not isinstance(distribution_id, str) or distribution_id.strip() == "":
176+
raise ValueError("Invalid distribution_id. It should be a non-empty string.")
177+
178+
return True
179+
180+
181+
def add_individual_to_survey(participant_study_id, embedded_data_field, survey_id, distribution_id):
182+
check_inputs_validity(participant_study_id, embedded_data_field, survey_id, distribution_id)
183+
184+
mailing_list_id = get_mailing_list_id_of_distribution(survey_id, distribution_id)
185+
directory_id = get_directory_id()
186+
contact_exists = check_contact_in_mailing_list(participant_study_id, mailing_list_id, directory_id)
187+
188+
if not contact_exists:
189+
add_contact_to_mailing_list(participant_study_id, mailing_list_id, directory_id)
190+
else:
191+
print(f"Participant study ID {participant_study_id} already exists in this mailing list")
192+
193+
personal_link = get_personal_link(survey_id, distribution_id, participant_study_id)
194+
return personal_link
195+
196+
if __name__ == "__main__":
197+
participant_study_id = "11111TEST11111"
198+
survey_id=SURVEYIDS.my_test_survey2_id
199+
distribution_id="EMD_SZFeoK7LAJBHU4d"
200+
embedded_data_field = "study_id_child"
201+
202+
link = add_individual_to_survey(participant_study_id, embedded_data_field, survey_id, distribution_id)
203+
print(link)

‎get_distributions.py‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
import requests
2+
from qualtrics_settings import BASE_URL, HEADERS, SURVEYIDS
3+
4+
def get_distributions(survey_id):
5+
endpoint = f"{BASE_URL}/distributions"
6+
parameters = {
7+
"surveyId": survey_id,
8+
"distributionRequestType": "GeneratedInvite"
9+
}
10+
11+
response = requests.get(endpoint, headers=HEADERS, params=parameters)
12+
13+
if response.status_code == 200:
14+
result = response.json()["result"]["elements"]
15+
distribution_ids = [f'{dist["id"]} created: {dist["createdDate"]}' for dist in result]
16+
return distribution_ids
17+
18+
else:
19+
print(f"Error in get_distributions_of_survey: {response.text}")
20+
return None
21+
22+
if __name__ == "__main__":
23+
result = get_distributions(SURVEYIDS.my_test_survey2_id)
24+
for res in result:
25+
print(res)
26+

‎qualtrics_settings.py‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
"""
2+
Qualtrics API settings
3+
4+
Change X-API-TOKEN. You can request it on your account under Account Settings.
5+
Add your survey IDs to the SURVEYIDS class. You can find it in the URL when you open your survey. It always starts with "SV_".
6+
"""
7+
8+
BASE_URL = "https://fra1.qualtrics.com/API/v3"
9+
HEADERS = {
10+
"Content-Type": "application/json",
11+
"X-API-TOKEN": "u2ufrDKnw5plXTrzkTnoV0rGtxC0hZJ2rjVAguDD"
12+
}
13+
14+
class QualtricsAPIError(Exception):
15+
pass
16+
17+
class SURVEYIDS():
18+
my_test_survey_id = "SV_0llWVSZNOQOorSC"
19+
my_test_survey2_id = "SV_efCMOg6wHU0T8ii"

‎requirements.txt‎

392 Bytes
Binary file not shown.

0 commit comments

Comments
 (0)