Coverage for crateweb/consent/tasks.py: 31%

1""" 

2crate_anon/crateweb/consent/tasks.py 

3 

4=============================================================================== 

5 

6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry. 

7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk). 

8 

9 This file is part of CRATE. 

10 

11 CRATE is free software: you can redistribute it and/or modify 

12 it under the terms of the GNU General Public License as published by 

13 the Free Software Foundation, either version 3 of the License, or 

14 (at your option) any later version. 

15 

16 CRATE is distributed in the hope that it will be useful, 

17 but WITHOUT ANY WARRANTY; without even the implied warranty of 

18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

19 GNU General Public License for more details. 

20 

21 You should have received a copy of the GNU General Public License 

22 along with CRATE. If not, see <https://www.gnu.org/licenses/>. 

23 

24=============================================================================== 

25 

26**Celery tasks for the CRATE consent-to-contact system.** 

27 

28See also :mod:`crate_anon.crateweb.consent.celery`, which defines the ``app``. 

29 

30**If you get a "received unregistered task" error:** 

31 

321. Restart the Celery worker. That may fix it. 

33 

342. If that fails, consider: SPECIFY ABSOLUTE NAMES FOR TASKS. 

35 e.g. with ``@shared_task(name="myfuncname")``. 

36 Possible otherwise that if this module is imported in different ways 

37 (e.g. absolute, relative), you'll get a "Received unregistered task" 

38 error. 

39 https://docs.celeryq.org/en/latest/userguide/tasks.html#task-names 

40 

41 

42**Acknowledgement/not doing things more than once:** 

43 

44- https://docs.celeryproject.org/en/latest/userguide/tasks.html 

45 

46- default is to acknowledge on receipt of request, not after task completion; 

47 that prevents things from happening more than once. 

48 If you can guarantee your function is idempotent, you can acknowledge after 

49 completion. 

50 

51- https://docs.celeryproject.org/en/latest/faq.html#faq-acks-late-vs-retry 

52 

53- We'll stick with the default (slightly less reliable but won't be run more 

54 than once). 

55 

56 

57**Circular imports:** 

58 

59- https://stackoverflow.com/questions/17313532/django-import-loop-between-celery-tasks-and-my-models 

60 

61- The potential circularity is: 

62 

63 - At launch, Celery must import tasks, which could want to import models. 

64 

65 - At launch, Django loads models, which may use tasks. 

66 

67- Simplest solution is to keep tasks very simple (as below) and use delayed 

68 imports here. 

69 

70 

71**Race condition:** 

72 

73- Django: 

74 

75 (1) existing object 

76 (2) amend with form 

77 (3) save() 

78 (4) call function.delay(obj.id) 

79 

80 Object is received by Celery in the state before save() at step 3. 

81 

82- https://celery.readthedocs.org/en/latest/userguide/tasks.html#database-transactions 

83 

84- https://stackoverflow.com/questions/26862942/django-related-objects-are-missing-from-celery-task-race-condition 

85 

86- https://code.djangoproject.com/ticket/14051 

87 

88- https://github.com/aaugustin/django-transaction-signals 

89 

90- SOLUTION: 

91 https://docs.djangoproject.com/en/dev/topics/db/transactions/#django.db.transaction.on_commit 

92 

93 .. code-block:: python 

94 

95 from django.db import transaction 

96 transaction.on_commit(lambda: blah.delay(blah)) 

97 

98 Requires Django 1.9. As of 2015-11-21, that means 1.9rc1 

99 

100""" # noqa: E501 

101 

102import logging 

103from celery import shared_task 

104from django.conf import settings 

105from django.contrib.auth import get_user_model 

106from django.urls import set_script_prefix 

107 

108log = logging.getLogger(__name__) 

109 

110 

111# noinspection PyCallingNonCallable 

112@shared_task(ignore_result=True) 

113def add(x: float, y: float) -> float: 

114 """ 

115 Task to add two numbers. For testing! 

116 

117 Args: 

118 x: a float 

119 y: another float 

120 

121 Returns: 

122 x + y 

123 

124 """ 

125 log.debug("add") 

126 return x + y 

127 

128 

129# noinspection PyCallingNonCallable,PyPep8Naming 

130@shared_task(ignore_result=True) 

131def resend_email(email_id: int, user_id: int) -> None: 

132 """ 

133 Celery task to resend a pre-existing e-mail. 

134 

135 Args: 

136 email_id: ID of the e-mail 

137 user_id: ID of the sending user 

138 

139 Callers include 

140 - :meth:`crate_anon.crateweb.core.admin.EmailDevAdmin.resend` 

141 

142 Creates/saves an 

143 :class:`crate_anon.crateweb.consent.models.EmailTransmission`. 

144 """ 

145 from crate_anon.crateweb.consent.models import Email # delayed import 

146 

147 log.debug("resend_email") 

148 User = get_user_model() 

149 email = Email.objects.get(pk=email_id) 

150 user = User.objects.get(pk=user_id) 

151 email.resend(user) 

152 

153 

154# noinspection PyCallingNonCallable 

155@shared_task(ignore_result=True) 

156def process_contact_request(contact_request_id: int) -> None: 

157 """ 

158 Celery task to act on a contact request. For example, might send an e-mail 

159 to a clinician, or generate a letter to the researcher. 

160 

161 Callers include 

162 - :meth:`crate_anon.crateweb.consent.models.ContactRequest.create` 

163 

164 Sets ``processed = True`` and ``processed_at`` for the 

165 :class:`crate_anon.crateweb.consent.models.ContactRequest`. 

166 

167 Args: 

168 contact_request_id: PK of the contact request 

169 """ 

170 from crate_anon.crateweb.consent.models import ( 

171 ContactRequest, 

172 ) # delayed import 

173 

174 log.debug("process_contact_request") 

175 set_script_prefix(settings.FORCE_SCRIPT_NAME) # see site_absolute_url 

176 contact_request = ContactRequest.objects.get( 

177 pk=contact_request_id 

178 ) # type: ContactRequest 

179 contact_request.process_request() 

180 

181 

182# noinspection PyCallingNonCallable 

183@shared_task(ignore_result=True) 

184def finalize_clinician_response(clinician_response_id: int) -> None: 

185 """ 

186 Celery task to do the thinking associated with a clinician's response to 

187 a contact request. For example, might generate letters to patients and 

188 notify the Research Database Manager of work to be done. 

189 

190 Callers include 

191 - :meth:`crate_anon.crateweb.consent.views.finalize_clinician_response_in_background` 

192 

193 Sets ``processed = True`` and ``processed_at`` for the 

194 :class:`crate_anon.crateweb.consent.models.ClinicianResponse`. 

195 

196 Args: 

197 clinician_response_id: PK of the clinician response 

198 """ # noqa: E501 

199 from crate_anon.crateweb.consent.models import ( 

200 ClinicianResponse, 

201 ) # delayed import 

202 

203 log.debug("finalize_clinician_response") 

204 clinician_response = ClinicianResponse.objects.get( 

205 pk=clinician_response_id 

206 ) 

207 clinician_response.finalize_b() # second part of processing 

208 

209 

210@shared_task(ignore_result=True) 

211def refresh_all_consent_modes() -> None: 

212 """ 

213 Celery task to refresh all consent modes. Uses 

214 :meth:`crate_anon.crateweb.consent.models.ConsentMode.refresh_from_primary_clinical_record` 

215 so it uses the correct consent mode, i.e. external primary clinical record 

216 takes priority. 

217 """ 

218 from crate_anon.crateweb.consent.models import ( 

219 ConsentMode, 

220 ) # delayed import 

221 from django.contrib.auth.models import User # delayed import 

222 

223 log.debug("refresh_all_consent_modes") 

224 

225 # Get a superuser to be the 'created_by' user for the consent modes. 

226 # Maybe we should: (1) add an 'is_rdbm' field to auth user or 

227 # (2) have a superuser especially for automatic tasks? 

228 auto_creator = User.objects.first() 

229 all_consent_modes = ConsentMode.objects.all() 

230 for cm in all_consent_modes: 

231 ConsentMode.refresh_from_primary_clinical_record( 

232 nhs_number=cm.nhs_number, created_by=auto_creator 

233 ) 

234 

235 

236# noinspection PyCallingNonCallable 

237@shared_task(ignore_result=True) 

238def process_consent_change(consent_mode_id: int) -> None: 

239 """ 

240 Celery task to do the thinking associated with a change of consent mode 

241 (e.g. might send a withdrawal letter to a researcher). 

242 

243 Callers include: 

244 - :meth:`crate_anon.crateweb.core.admin.ConsentModeMgrAdmin.save_model` 

245 

246 Sets ``processed = True`` for the 

247 :class:`crate_anon.crateweb.consent.models.ConsentMode`, 

248 if ``current == True`` and ``needs_processing == True``. 

249 

250 .. todo:: process_consent_change: don't process twice 

251 

252 Args: 

253 consent_mode_id: PK of the consent mode 

254 """ 

255 from crate_anon.crateweb.consent.models import ( 

256 ConsentMode, 

257 ) # delayed import 

258 

259 log.debug("process_consent_change") 

260 consent_mode = ConsentMode.objects.get(pk=consent_mode_id) 

261 consent_mode.process_change() 

262 

263 

264# noinspection PyCallingNonCallable 

265@shared_task(ignore_result=True) 

266def process_patient_response(patient_response_id: int) -> None: 

267 """ 

268 Celery task to do the thinking associated with a patient's decision. 

269 For example, might send a letter to a researcher. 

270 

271 Sets ``processed = True`` and ``processed_at`` for the 

272 :class:`crate_anon.crateweb.consent.models.PatientResponse`. 

273 

274 Args: 

275 patient_response_id: PK of the patient response 

276 """ 

277 from crate_anon.crateweb.consent.models import ( 

278 PatientResponse, 

279 ) # delayed import 

280 

281 log.debug("process_patient_response") 

282 patient_response = PatientResponse.objects.get(pk=patient_response_id) 

283 patient_response.process_response() 

284 

285 

286# noinspection PyCallingNonCallable 

287@shared_task(ignore_result=True) 

288def test_email_rdbm_task() -> None: 

289 """ 

290 Celery task to test the e-mail system by e-mailing the Research Database 

291 Manager. 

292 """ 

293 log.debug("test_email_rdbm_task") 

294 subject = "TEST MESSAGE FROM RESEARCH DATABASE COMPUTER" 

295 text = ( 

296 "Success! The CRATE framework can communicate via Celery with its " 

297 "message broker, so it can talk to an 'offline' copy of itself " 

298 "for background processing. And it can e-mail you." 

299 ) 

300 email_rdbm_task(subject, text) # Will this work as a function? Yes. 

301 

302 

303# noinspection PyCallingNonCallable 

304@shared_task(ignore_result=True) 

305def email_rdbm_task(subject: str, text: str) -> None: 

306 """ 

307 Celery task to e-mail the Research Database Manager. 

308 

309 Creates/saves an 

310 :class:`crate_anon.crateweb.consent.models.Email` and an 

311 :class:`crate_anon.crateweb.consent.models.EmailTransmission`. 

312 

313 Args: 

314 subject: e-mail subject 

315 text: e-mail body text 

316 """ 

317 from crate_anon.crateweb.consent.models import Email # delayed import 

318 

319 log.debug("email_rdbm_task") 

320 email = Email.create_rdbm_text_email(subject, text) 

321 et = email.send() 

322 if et is None: 

323 log.error("Failed to send e-mail") 

324 return 

325 if et.sent: 

326 log.info(str(et)) 

327 else: 

328 log.error(str(et)) 

329 

330 

331# noinspection PyCallingNonCallable 

332@shared_task(ignore_result=True) 

333def resubmit_unprocessed_tasks_task() -> None: 

334 """ 

335 Celery task to finish up any outstanding work. 

336 Use this with caution. 

337 

338 The idea is that if a previous Celery task crashed, it will have been 

339 removed from the Celery queue, but not completed. 

340 As of 2018-06-29, we make sure that we have completion flags. This task 

341 then works through anything unprocessed, and tries to process it. 

342 

343 All work gets added to the Celery queue. 

344 """ 

345 from crate_anon.crateweb.consent.models import ( 

346 ClinicianResponse, 

347 ) # delayed import 

348 from crate_anon.crateweb.consent.models import ( 

349 ConsentMode, 

350 ) # delayed import 

351 from crate_anon.crateweb.consent.models import ( 

352 ContactRequest, 

353 ) # delayed import 

354 from crate_anon.crateweb.consent.models import ( 

355 PatientResponse, 

356 ) # delayed import 

357 

358 log.debug("resubmit_unprocessed_tasks_task") 

359 

360 for patient_response in PatientResponse.get_unprocessed(): 

361 process_patient_response.delay(patient_response.id) 

362 

363 for clinician_response in ClinicianResponse.get_unprocessed(): 

364 finalize_clinician_response.delay(clinician_response.id) 

365 

366 for consent_mode in ConsentMode.get_unprocessed(): 

367 process_consent_change.delay(consent_mode.id) 

368 

369 for contact_request in ContactRequest.get_unprocessed(): 

370 process_contact_request.delay(contact_request.id)