-
Notifications
You must be signed in to change notification settings - Fork 19
/
_source.py
249 lines (192 loc) · 9.01 KB
/
_source.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
import uuid
from datetime import date
from flask import jsonify, make_response
from microsetta_private_api.api._account import _validate_account_access
from microsetta_private_api.api.literals import SRC_NOT_FOUND_MSG,\
SRC_NO_DELETE_MSG
from microsetta_private_api.exceptions import RepoException
from microsetta_private_api.model.source import Source, HumanInfo, NonHumanInfo
from microsetta_private_api.repo.source_repo import SourceRepo
from microsetta_private_api.repo.consent_repo import ConsentRepo
from microsetta_private_api.repo.sample_repo import SampleRepo
from microsetta_private_api.repo.survey_answers_repo import SurveyAnswersRepo
from microsetta_private_api.repo.survey_template_repo import SurveyTemplateRepo
from microsetta_private_api.repo.transaction import Transaction
def read_sources(account_id, token_info, source_type=None):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
sources = source_repo.get_sources_in_account(account_id, source_type)
api_sources = [x.to_api() for x in sources]
# TODO: Also support 404? Or is that not necessary?
return jsonify(api_sources), 200
def create_source(account_id, body, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
source_id = str(uuid.uuid4())
name = body["source_name"]
source_type = body['source_type']
if source_type == Source.SOURCE_TYPE_HUMAN:
# TODO: Unfortunately, humans require a lot of special handling,
# and we started mixing Source calls used for transforming to/
# from the database with source calls to/from the api.
# Would be nice to split this out better.
source_info = HumanInfo.from_dict(body,
consent_date=date.today(),
date_revoked=None)
# the "legacy" value of the age_range enum is not valid to use when
# creating a new source, so do not allow that.
# NB: Not necessary to do this check when updating a source as
# only source name and description (not age_range) may be updated.
if source_info.age_range == "legacy":
raise RepoException("Age range may not be set to legacy.")
else:
source_info = NonHumanInfo.from_dict(body)
new_source = Source(source_id,
account_id,
source_type,
name,
source_info)
source_repo.create_source(new_source)
# Must pull from db to get creation_time, update_time
s = source_repo.get_source(account_id, new_source.id)
t.commit()
response = jsonify(s.to_api())
response.status_code = 201
response.headers['Location'] = '/api/accounts/%s/sources/%s' % \
(account_id, source_id)
return response
def read_source(account_id, source_id, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
source = source_repo.get_source(account_id, source_id)
if source is None:
return jsonify(code=404, message=SRC_NOT_FOUND_MSG), 404
return jsonify(source.to_api()), 200
def check_source_ffq_prereqs(account_id, source_id, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
st_repo = SurveyTemplateRepo(t)
birth_year, gender, height, weight =\
st_repo.fetch_user_basic_physiology(account_id, source_id)
if birth_year is None or gender is None or height is None or\
weight is None:
return jsonify({"basic_info": False}), 200
return jsonify({"basic_info": True}), 200
def update_source(account_id, source_id, body, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
source = source_repo.get_source(account_id, source_id)
if source is None:
return jsonify(code=404, message=SRC_NOT_FOUND_MSG), 404
source.name = body["source_name"]
# every type of source has a name but not every type has a description
if getattr(source.source_data, "description", False):
source.source_data.description = body.get(
"source_description", None)
source_repo.update_source_data_api_fields(source)
# I wonder if there's some way to get the creation_time/update_time
# during the insert/update...
source = source_repo.get_source(account_id, source_id)
t.commit()
# TODO: 422? Not sure this can actually happen anymore ...
return jsonify(source.to_api()), 200
def scrub_source(account_id, source_id, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
consent_repo = ConsentRepo(t)
sample_repo = SampleRepo(t)
sur_repo = SurveyAnswersRepo(t)
# The interface has historically enforced this constraint, but it
# wasn't codified into the API
samples = sample_repo.get_samples_by_source(account_id, source_id)
if len(samples) > 0:
return jsonify(code=422, message=SRC_NO_DELETE_MSG), 422
# fetch and scrub all surveys
surveys = sur_repo.list_answered_surveys(account_id, source_id)
for survey_id in surveys:
sur_repo.scrub(account_id, source_id, survey_id)
# scrub all consents associated with source
consent_repo.scrub(account_id, source_id)
# scrub the source
source_repo.scrub(account_id, source_id)
t.commit()
return '', 204
def create_human_source_from_consent(account_id, body, token_info):
_validate_account_access(token_info, account_id)
# Must convert consent form body into object processable by create_source.
# Not adding any error handling here because if 'participant_name' isn't
# here, we SHOULD be getting an error.
source = {
'source_type': Source.SOURCE_TYPE_HUMAN,
'source_name': body['participant_name'],
'consent': {
'age_range': body['age_range']
}
}
child_keys = {'parent_1_name', 'assent_obtainer'}
intersection = child_keys.intersection(body)
if intersection:
source['consent']['child_info'] = {}
for key in intersection:
source['consent']['child_info'][key] = body[key]
# NB: Don't expect to handle errors 404, 422 in this function; expect to
# farm out to `create_source`
return create_source(account_id, source, token_info)
def check_duplicate_source_name(account_id, body, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
source_name = body['participant_name']
source = source_repo.get_duplicate_source_name(
account_id, source_name)
return jsonify(source), 200
def check_prompt_survey_update(account_id, source_id, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
s_t_r = SurveyTemplateRepo(t)
prompt_update = s_t_r.check_prompt_survey_update(source_id)
return jsonify({"prompt": prompt_update}), 200
def get_external_reports(account_id, source_id, token_info):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
reports = source_repo.get_external_reports(source_id)
for r in reports:
r.file_contents = ""
return jsonify(reports), 200
def get_external_report(
account_id, source_id, external_report_id, token_info
):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
reports = source_repo.get_external_reports(
source_id, external_report_id
)
if len(reports) != 1:
return jsonify(code=404, message="Report not found"), 404
report = reports[0]
# Trying to jsonify the actual contents gets ugly, so we return
# everything else here, and the contents in get_external_report_bytes
report.file_contents = ""
return jsonify(report.to_api()), 200
def get_external_report_bytes(
account_id, source_id, external_report_id, token_info
):
_validate_account_access(token_info, account_id)
with Transaction() as t:
source_repo = SourceRepo(t)
reports = source_repo.get_external_reports(
source_id, external_report_id
)
if len(reports) != 1:
return jsonify(code=404, message="Report not found"), 404
report = reports[0]
response = make_response(bytes(report.file_contents))
response.headers.set("Content-Type", report.file_type)
return response