Coverage for cc_modules/cc_taskschedule.py: 77%

1""" 

2camcops_server/cc_modules/cc_taskschedule.py 

3 

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

5 

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

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

8 

9 This file is part of CamCOPS. 

10 

11 CamCOPS 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 CamCOPS 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 CamCOPS. If not, see <https://www.gnu.org/licenses/>. 

23 

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

25 

26""" 

27 

28import logging 

29from typing import Any, List, Optional, Tuple, TYPE_CHECKING 

30from urllib.parse import urlencode, urlunsplit 

31 

32from cardinal_pythonlib.uriconst import UriSchemes 

33from pendulum import DateTime as Pendulum, Duration 

34 

35from sqlalchemy import cast, Numeric 

36from sqlalchemy.orm import Mapped, mapped_column, relationship 

37from sqlalchemy.sql.functions import func 

38from sqlalchemy.sql.schema import ForeignKey 

39from sqlalchemy.sql.sqltypes import BigInteger, UnicodeText 

40 

41from camcops_server.cc_modules.cc_email import Email 

42from camcops_server.cc_modules.cc_formatter import SafeFormatter 

43from camcops_server.cc_modules.cc_group import Group 

44from camcops_server.cc_modules.cc_pyramid import Routes 

45from camcops_server.cc_modules.cc_simpleobjects import IdNumReference 

46from camcops_server.cc_modules.cc_sqlalchemy import Base 

47from camcops_server.cc_modules.cc_sqla_coltypes import ( 

48 EmailAddressColType, 

49 JsonColType, 

50 PendulumDateTimeAsIsoTextColType, 

51 PendulumDurationAsIsoTextColType, 

52 TableNameColType, 

53) 

54from camcops_server.cc_modules.cc_task import ( 

55 Task, 

56 tablename_to_task_class_dict, 

57) 

58from camcops_server.cc_modules.cc_taskcollection import ( 

59 TaskFilter, 

60 TaskCollection, 

61 TaskSortMethod, 

62) 

63 

64 

65if TYPE_CHECKING: 

66 from sqlalchemy.sql.elements import Cast 

67 from camcops_server.cc_modules.cc_request import CamcopsRequest 

68 

69log = logging.getLogger(__name__) 

70 

71 

72# ============================================================================= 

73# ScheduledTaskInfo 

74# ============================================================================= 

75 

76 

77class ScheduledTaskInfo(object): 

78 """ 

79 Simple representation of a scheduled task (which may also contain the 

80 actual completed task, in its ``task`` member, if there is one). 

81 """ 

82 

83 def __init__( 

84 self, 

85 shortname: str, 

86 tablename: str, 

87 is_anonymous: bool, 

88 task: Optional[Task] = None, 

89 start_datetime: Optional[Pendulum] = None, 

90 end_datetime: Optional[Pendulum] = None, 

91 ) -> None: 

92 self.shortname = shortname 

93 self.tablename = tablename 

94 self.is_anonymous = is_anonymous 

95 self.task = task 

96 self.start_datetime = start_datetime 

97 self.end_datetime = end_datetime 

98 

99 @property 

100 def due_now(self) -> bool: 

101 """ 

102 Are we in the range [start_datetime, end_datetime)? 

103 """ 

104 if not self.start_datetime or not self.end_datetime: 

105 return False 

106 return self.start_datetime <= Pendulum.now() < self.end_datetime 

107 

108 @property 

109 def is_complete(self) -> bool: 

110 """ 

111 Returns whether its associated task is complete.. 

112 """ 

113 if not self.task: 

114 return False 

115 return self.task.is_complete() 

116 

117 @property 

118 def is_identifiable_and_incomplete(self) -> bool: 

119 """ 

120 If this is an anonymous task, returns ``False``. 

121 If this is an identifiable task, returns ``not is_complete``. 

122 """ 

123 if self.is_anonymous: 

124 return False 

125 return not self.is_complete 

126 

127 @property 

128 def due_now_identifiable_and_incomplete(self) -> bool: 

129 """ 

130 Is this task currently due, identifiable, and incomplete? 

131 """ 

132 return self.due_now and self.is_identifiable_and_incomplete 

133 

134 

135# ============================================================================= 

136# PatientTaskSchedule 

137# ============================================================================= 

138 

139 

140class PatientTaskSchedule(Base): 

141 """ 

142 Joining table that associates a patient with a task schedule 

143 """ 

144 

145 __tablename__ = "_patient_task_schedule" 

146 

147 id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) 

148 patient_pk: Mapped[int] = mapped_column(ForeignKey("patient._pk")) 

149 schedule_id: Mapped[int] = mapped_column(ForeignKey("_task_schedule.id")) 

150 start_datetime: Mapped[Optional[Pendulum]] = mapped_column( 

151 PendulumDateTimeAsIsoTextColType, 

152 comment=( 

153 "Schedule start date for the patient. Due from/within " 

154 "durations for a task schedule item are relative to this." 

155 ), 

156 ) 

157 settings: Mapped[Optional[Any]] = mapped_column( 

158 JsonColType, 

159 comment="Task-specific settings for this patient", 

160 ) 

161 

162 patient = relationship("Patient", back_populates="task_schedules") 

163 task_schedule: Mapped["TaskSchedule"] = relationship( 

164 back_populates="patient_task_schedules", 

165 cascade_backrefs=False, 

166 ) 

167 

168 emails = relationship( 

169 "PatientTaskScheduleEmail", 

170 back_populates="patient_task_schedule", 

171 cascade="all, delete", 

172 cascade_backrefs=False, 

173 ) 

174 

175 def get_list_of_scheduled_tasks( 

176 self, req: "CamcopsRequest" 

177 ) -> List[ScheduledTaskInfo]: 

178 """ 

179 Tasks scheduled for this patient. 

180 """ 

181 

182 task_list = [] 

183 

184 task_class_lookup = tablename_to_task_class_dict() 

185 

186 for tsi in self.task_schedule.items: 

187 start_datetime = None 

188 end_datetime = None 

189 task = None 

190 

191 if self.start_datetime is not None: 

192 start_datetime = self.start_datetime.add( 

193 days=tsi.due_from.days 

194 ) 

195 end_datetime = self.start_datetime.add(days=tsi.due_by.days) 

196 

197 task = self.find_scheduled_task( 

198 req, tsi, start_datetime, end_datetime 

199 ) 

200 

201 task_class = task_class_lookup[tsi.task_table_name] 

202 

203 task_list.append( 

204 ScheduledTaskInfo( 

205 task_class.shortname, 

206 tsi.task_table_name, 

207 is_anonymous=task_class.is_anonymous, 

208 task=task, 

209 start_datetime=start_datetime, 

210 end_datetime=end_datetime, 

211 ) 

212 ) 

213 

214 return task_list 

215 

216 def find_scheduled_task( 

217 self, 

218 req: "CamcopsRequest", 

219 tsi: "TaskScheduleItem", 

220 start_datetime: Pendulum, 

221 end_datetime: Pendulum, 

222 ) -> Optional[Task]: 

223 """ 

224 Returns the most recently uploaded task that matches the patient (by 

225 any ID number, i.e. via OR), task type and timeframe 

226 """ 

227 taskfilter = TaskFilter() 

228 for idnum in self.patient.idnums: 

229 idnum_ref = IdNumReference( 

230 which_idnum=idnum.which_idnum, idnum_value=idnum.idnum_value 

231 ) 

232 taskfilter.idnum_criteria.append(idnum_ref) 

233 

234 taskfilter.task_types = [tsi.task_table_name] 

235 

236 taskfilter.start_datetime = start_datetime 

237 taskfilter.end_datetime = end_datetime 

238 

239 # TODO: Improve error reporting 

240 # Shouldn't happen in normal operation as the task schedule item form 

241 # validation will ensure the dates are correct. 

242 # However, it's quite easy to write tests with unintentionally 

243 # inconsistent dates. 

244 # If we don't assert this here, we get a more cryptic assertion 

245 # failure later: 

246 # 

247 # cc_taskcollection.py _fetch_tasks_from_indexes() 

248 # assert self._all_indexes is not None 

249 assert not taskfilter.dates_inconsistent() 

250 

251 collection = TaskCollection( 

252 req=req, 

253 taskfilter=taskfilter, 

254 sort_method_global=TaskSortMethod.CREATION_DATE_DESC, 

255 ) 

256 

257 if len(collection.all_tasks) > 0: 

258 return collection.all_tasks[0] 

259 

260 return None 

261 

262 def email_body(self, req: "CamcopsRequest") -> str: 

263 """ 

264 Body content (HTML) for an e-mail to the patient -- the schedule's 

265 template, populated with patient-specific information. 

266 """ 

267 template_dict = dict( 

268 access_key=self.patient.uuid_as_proquint, 

269 android_launch_url=self.launch_url(req, UriSchemes.HTTP), 

270 ios_launch_url=self.launch_url(req, "camcops"), 

271 forename=self.patient.forename, 

272 server_url=req.route_url(Routes.CLIENT_API), 

273 surname=self.patient.surname, 

274 ) 

275 

276 formatter = TaskScheduleEmailTemplateFormatter() 

277 return formatter.format( 

278 self.task_schedule.email_template, **template_dict 

279 ) 

280 

281 def launch_url(self, req: "CamcopsRequest", scheme: str) -> str: 

282 # Matches intent-filter in AndroidManifest.xml 

283 # And CFBundleURLSchemes in Info.plist 

284 

285 # iOS doesn't care about these: 

286 netloc = "camcops.org" 

287 path = "/register/" 

288 fragment = "" 

289 

290 query_dict = { 

291 "default_single_user_mode": "true", 

292 "default_server_location": req.route_url(Routes.CLIENT_API), 

293 "default_access_key": self.patient.uuid_as_proquint, 

294 } 

295 query = urlencode(query_dict) 

296 

297 return urlunsplit((scheme, netloc, path, query, fragment)) 

298 

299 @property 

300 def email_sent(self) -> bool: 

301 """ 

302 Has an e-mail been sent to the patient for this schedule? 

303 """ 

304 return any(e.email.sent for e in self.emails) 

305 

306 

307def task_schedule_item_sort_order() -> Tuple["Cast", "Cast"]: 

308 """ 

309 Returns a tuple of sorting functions for use with SQLAlchemy ORM queries, 

310 to sort task schedule items. 

311 

312 The durations are currently stored as seconds e.g. P0Y0MT2594592000.0S 

313 and the seconds aren't zero padded, so we need to do some processing 

314 to get them in the order we want. 

315 

316 This will fail if durations ever get stored any other way. 

317 

318 Note that MySQL does not permit "CAST(... AS DOUBLE)" or "CAST(... AS 

319 FLOAT)"; you need to use NUMERIC or DECIMAL. However, this raises a warning 

320 when running self-tests under SQLite: "SAWarning: Dialect sqlite+pysqlite 

321 does *not* support Decimal objects natively, and SQLAlchemy must convert 

322 from floating point - rounding errors and other issues may occur. Please 

323 consider storing Decimal numbers as strings or integers on this platform 

324 for lossless storage." 

325 """ 

326 due_from_order = cast(func.substr(TaskScheduleItem.due_from, 7), Numeric()) 

327 due_by_order = cast(func.substr(TaskScheduleItem.due_by, 7), Numeric()) 

328 

329 return due_from_order, due_by_order 

330 

331 

332# ============================================================================= 

333# Emails sent to patient 

334# ============================================================================= 

335 

336 

337class PatientTaskScheduleEmail(Base): 

338 """ 

339 Represents an email send to a patient for a particular task schedule. 

340 """ 

341 

342 __tablename__ = "_patient_task_schedule_email" 

343 

344 id: Mapped[int] = mapped_column( 

345 primary_key=True, 

346 autoincrement=True, 

347 comment="Arbitrary primary key", 

348 ) 

349 patient_task_schedule_id: Mapped[int] = mapped_column( 

350 ForeignKey(PatientTaskSchedule.id), 

351 comment=( 

352 f"FK to {PatientTaskSchedule.__tablename__}." 

353 f"{PatientTaskSchedule.id.name}" 

354 ), 

355 ) 

356 email_id: Mapped[int] = mapped_column( 

357 BigInteger, 

358 ForeignKey(Email.id), 

359 comment=f"FK to {Email.__tablename__}.{Email.id.name}", 

360 ) 

361 

362 patient_task_schedule = relationship( 

363 PatientTaskSchedule, 

364 back_populates="emails", 

365 cascade_backrefs=False, 

366 ) 

367 email = relationship(Email, cascade="all, delete") 

368 

369 

370# ============================================================================= 

371# Task schedule 

372# ============================================================================= 

373 

374 

375class TaskSchedule(Base): 

376 """ 

377 A named collection of task schedule items 

378 """ 

379 

380 __tablename__ = "_task_schedule" 

381 

382 id: Mapped[int] = mapped_column( 

383 primary_key=True, 

384 autoincrement=True, 

385 comment="Arbitrary primary key", 

386 ) 

387 

388 group_id: Mapped[int] = mapped_column( 

389 ForeignKey(Group.id), 

390 comment="FK to {}.{}".format(Group.__tablename__, Group.id.name), 

391 ) 

392 

393 name: Mapped[Optional[str]] = mapped_column(UnicodeText, comment="name") 

394 

395 email_subject: Mapped[str] = mapped_column( 

396 UnicodeText, 

397 comment="email subject", 

398 default="", 

399 ) 

400 email_template: Mapped[str] = mapped_column( 

401 UnicodeText, 

402 comment="email template", 

403 default="", 

404 ) 

405 email_from: Mapped[Optional[str]] = mapped_column( 

406 EmailAddressColType, comment="Sender's e-mail address" 

407 ) 

408 email_cc: Mapped[Optional[str]] = mapped_column( 

409 UnicodeText, 

410 comment="Send a carbon copy of the email to these addresses", 

411 ) 

412 email_bcc: Mapped[Optional[str]] = mapped_column( 

413 UnicodeText, 

414 comment="Send a blind carbon copy of the email to these addresses", 

415 ) 

416 

417 items: Mapped[list["TaskScheduleItem"]] = relationship( 

418 back_populates="task_schedule", 

419 order_by=task_schedule_item_sort_order, 

420 cascade="all, delete", 

421 cascade_backrefs=False, 

422 ) 

423 

424 group = relationship(Group) 

425 

426 patient_task_schedules = relationship( 

427 "PatientTaskSchedule", 

428 back_populates="task_schedule", 

429 cascade="all, delete", 

430 cascade_backrefs=False, 

431 ) 

432 

433 def user_may_edit(self, req: "CamcopsRequest") -> bool: 

434 """ 

435 May the current user edit this schedule? 

436 """ 

437 return req.user.may_administer_group(self.group_id) 

438 

439 

440class TaskScheduleItem(Base): 

441 """ 

442 An individual item in a task schedule 

443 """ 

444 

445 __tablename__ = "_task_schedule_item" 

446 

447 id: Mapped[int] = mapped_column( 

448 primary_key=True, 

449 autoincrement=True, 

450 comment="Arbitrary primary key", 

451 ) 

452 

453 schedule_id: Mapped[int] = mapped_column( 

454 ForeignKey(TaskSchedule.id), 

455 comment="FK to {}.{}".format( 

456 TaskSchedule.__tablename__, TaskSchedule.id.name 

457 ), 

458 ) 

459 

460 task_table_name: Mapped[Optional[str]] = mapped_column( 

461 TableNameColType, 

462 index=True, 

463 comment="Table name of the task's base table", 

464 ) 

465 

466 due_from: Mapped[Optional[Duration]] = mapped_column( 

467 PendulumDurationAsIsoTextColType, 

468 comment=( 

469 "Relative time from the start date by which the task may be " 

470 "started" 

471 ), 

472 ) 

473 

474 due_by: Mapped[Optional[Duration]] = mapped_column( 

475 PendulumDurationAsIsoTextColType, 

476 comment=( 

477 "Relative time from the start date by which the task must be " 

478 "completed" 

479 ), 

480 ) 

481 

482 task_schedule = relationship( 

483 "TaskSchedule", back_populates="items", cascade_backrefs=False 

484 ) 

485 

486 @property 

487 def task_shortname(self) -> str: 

488 """ 

489 Short name of the task being scheduled. 

490 """ 

491 task_class_lookup = tablename_to_task_class_dict() 

492 

493 return task_class_lookup[self.task_table_name].shortname 

494 

495 @property 

496 def due_within(self) -> Optional[Duration]: 

497 """ 

498 Returns the "due within" property, e.g. "due within 7 days (of being 

499 scheduled to start)". This is calculated from due_from and due_by. 

500 """ 

501 if self.due_by is None: 

502 # Should not be possible if created through the form 

503 return None 

504 

505 if self.due_from is None: 

506 return self.due_by 

507 

508 return self.due_by - self.due_from 

509 

510 def description(self, req: "CamcopsRequest") -> str: 

511 """ 

512 Description of this schedule item -- which task, due when. 

513 """ 

514 _ = req.gettext 

515 

516 if self.due_from is None: 

517 # Should not be possible if created through the form 

518 due_days = "?" 

519 else: 

520 due_days = str(self.due_from.in_days()) 

521 

522 return _("{task_name} @ {due_days} days").format( 

523 task_name=self.task_shortname, due_days=due_days 

524 ) 

525 

526 

527class TaskScheduleEmailTemplateFormatter(SafeFormatter): 

528 """ 

529 Safe template formatter for task schedule e-mails. 

530 """ 

531 

532 def __init__(self) -> None: 

533 super().__init__( 

534 [ 

535 "access_key", 

536 "android_launch_url", 

537 "forename", 

538 "ios_launch_url", 

539 "server_url", 

540 "surname", 

541 ] 

542 )