Coverage for cc_modules/cc_redcap.py: 27%

264 statements  

« prev     ^ index     » next       coverage.py v7.9.2, created at 2025-07-15 14:23 +0100

1"""camcops_server/cc_modules/cc_redcap.py 

2 

3=============================================================================== 

4 

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

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

7 

8 This file is part of CamCOPS. 

9 

10 CamCOPS is free software: you can redistribute it and/or modify 

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

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

13 (at your option) any later version. 

14 

15 CamCOPS is distributed in the hope that it will be useful, 

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

17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

18 GNU General Public License for more details. 

19 

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

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

22 

23=============================================================================== 

24 

25**Implements communication with REDCap.** 

26 

27- For general information about REDCap, see https://www.project-redcap.org/. 

28 

29- The API documentation is not provided there, but is available from 

30 your local REDCap server. Pick a project. Choose "API" from the left-hand 

31 menu. Follow the "REDCap API documentation" link. 

32 

33- We use PyCap (https://pycap.readthedocs.io/ or 

34 https://github.com/redcap-tools/PyCap). See also 

35 https://redcap-tools.github.io/projects/. PyCap is no longer being actively 

36 developed though the author is still responding to issues and pull requests. 

37 

38We use an XML fieldmap to describe how the rows in CamCOPS task tables are 

39translated into REDCap records. See :ref:`REDCap export <redcap>`. 

40 

41REDCap does not assign instance IDs for repeating instruments so we need to 

42query the database in order to determine the next instance ID. It is possible 

43to create a race condition if more than one client is trying to update the same 

44record at the same time. 

45 

46""" 

47 

48from enum import Enum 

49import io 

50import logging 

51from typing import Any, Dict, Iterable, List, Optional, TYPE_CHECKING, Union 

52import xml.etree.cElementTree as ElementTree 

53 

54from asteval import Interpreter, make_symbol_table 

55from cardinal_pythonlib.datetimefunc import format_datetime 

56from cardinal_pythonlib.logs import BraceStyleAdapter 

57from pandas import DataFrame 

58from pandas.errors import EmptyDataError 

59import redcap 

60 

61from camcops_server.cc_modules.cc_constants import ( 

62 ConfigParamExportRecipient, 

63 DateFormat, 

64) 

65from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient 

66 

67if TYPE_CHECKING: 

68 from camcops_server.cc_modules.cc_exportmodels import ExportedTaskRedcap 

69 from camcops_server.cc_modules.cc_request import CamcopsRequest 

70 from camcops_server.cc_modules.cc_task import Task 

71 

72log = BraceStyleAdapter(logging.getLogger(__name__)) 

73 

74MISSING_EVENT_TAG_OR_ATTRIBUTE = ( 

75 "The REDCap project has events but there is no 'event' tag " 

76 "in the fieldmap or an instrument is missing an 'event' " 

77 "attribute" 

78) 

79 

80 

81class RedcapExportException(Exception): 

82 pass 

83 

84 

85class RedcapFieldmap(object): 

86 """ 

87 Internal representation of the fieldmap XML file. 

88 This describes how the task fields should be translated to 

89 the REDCap record. 

90 """ 

91 

92 def __init__(self, filename: str) -> None: 

93 """ 

94 Args: 

95 filename: 

96 Name of an XML file telling CamCOPS how to map task fields 

97 to REDCap. See :ref:`REDCap export <redcap>`. 

98 """ 

99 self.filename = filename 

100 self.fields = {} # type: Dict[str, Dict[str, str]] 

101 # ... {task: {name: formula}} 

102 self.files = {} # type: Dict[str, Dict[str, str]] 

103 # ... {task: {name: formula}} 

104 self.instruments = {} # type: Dict[str, str] 

105 # ... {task: instrument_name} 

106 self.events = {} # type: Dict[str, str] 

107 # ... {task: event_name} 

108 

109 parser = ElementTree.XMLParser(encoding="UTF-8") 

110 try: 

111 tree = ElementTree.parse(filename, parser=parser) 

112 except FileNotFoundError: 

113 raise RedcapExportException( 

114 f"Unable to open fieldmap file '{filename}'" 

115 ) 

116 except ElementTree.ParseError as e: 

117 raise RedcapExportException( 

118 f"There was a problem parsing {filename}: {str(e)}" 

119 ) from e 

120 

121 root = tree.getroot() 

122 if root.tag != "fieldmap": 

123 raise RedcapExportException( 

124 ( 

125 f"Expected the root tag to be 'fieldmap' instead of " 

126 f"'{root.tag}' in {filename}" 

127 ) 

128 ) 

129 

130 patient_element = root.find("patient") 

131 if patient_element is None: 

132 raise RedcapExportException( 

133 f"'patient' is missing from {filename}" 

134 ) 

135 

136 self.patient = self._validate_and_return_attributes( 

137 patient_element, ("instrument", "redcap_field") 

138 ) 

139 

140 record_element = root.find("record") 

141 if record_element is None: 

142 raise RedcapExportException(f"'record' is missing from {filename}") 

143 

144 self.record = self._validate_and_return_attributes( 

145 record_element, ("instrument", "redcap_field") 

146 ) 

147 

148 default_event = None 

149 event_element = root.find("event") 

150 if event_element is not None: 

151 event_attributes = self._validate_and_return_attributes( 

152 event_element, ("name",) 

153 ) 

154 default_event = event_attributes["name"] 

155 

156 instrument_elements = root.find("instruments") 

157 if instrument_elements is None: 

158 raise RedcapExportException( 

159 f"'instruments' tag is missing from {filename}" 

160 ) 

161 

162 for instrument_element in instrument_elements: 

163 instrument_attributes = self._validate_and_return_attributes( 

164 instrument_element, ("name", "task") 

165 ) 

166 

167 task = instrument_attributes["task"] 

168 instrument_name = instrument_attributes["name"] 

169 self.fields[task] = {} 

170 self.files[task] = {} 

171 self.events[task] = instrument_attributes.get( 

172 "event", default_event 

173 ) 

174 self.instruments[task] = instrument_name 

175 

176 field_elements = instrument_element.find("fields") or [] 

177 

178 for field_element in field_elements: 

179 field_attributes = self._validate_and_return_attributes( 

180 field_element, ("name", "formula") 

181 ) 

182 name = field_attributes["name"] 

183 formula = field_attributes["formula"] 

184 

185 self.fields[task][name] = formula 

186 

187 file_elements = instrument_element.find("files") or [] 

188 for file_element in file_elements: 

189 file_attributes = self._validate_and_return_attributes( 

190 file_element, ("name", "formula") 

191 ) 

192 

193 name = file_attributes["name"] 

194 formula = file_attributes["formula"] 

195 self.files[task][name] = formula 

196 

197 def _validate_and_return_attributes( 

198 self, element: ElementTree.Element, expected_attributes: Iterable[str] 

199 ) -> Dict[str, str]: 

200 """ 

201 Checks that all the expected attributes are present in the XML element 

202 (from the fieldmap XML file), or raises :exc:`RedcapExportException`. 

203 """ 

204 attributes = element.attrib 

205 

206 if not all(a in attributes.keys() for a in expected_attributes): 

207 raise RedcapExportException( 

208 ( 

209 f"'{element.tag}' must have attributes: " 

210 f"{', '.join(expected_attributes)} in {self.filename}" 

211 ) 

212 ) 

213 

214 return attributes 

215 

216 def instrument_names(self) -> List[str]: 

217 """ 

218 Returns the names of all REDCap instruments. 

219 """ 

220 return list(self.instruments.values()) 

221 

222 

223class RedcapTaskExporter(object): 

224 """ 

225 Main entry point for task export to REDCap. Works out which record needs 

226 updating or creating. Creates the fieldmap and initiates upload. 

227 """ 

228 

229 def export_task( 

230 self, req: "CamcopsRequest", exported_task_redcap: "ExportedTaskRedcap" 

231 ) -> None: 

232 """ 

233 Exports a specific task. 

234 

235 Args: 

236 req: 

237 a :class:`camcops_server.cc_modules.cc_request.CamcopsRequest` 

238 exported_task_redcap: 

239 a :class:`camcops_server.cc_modules.cc_exportmodels.ExportedTaskRedcap` 

240 """ # noqa 

241 exported_task = exported_task_redcap.exported_task 

242 recipient = exported_task.recipient 

243 task = exported_task.task 

244 

245 if task.is_anonymous: 

246 raise RedcapExportException( 

247 f"Skipping anonymous task '{task.tablename}'" 

248 ) 

249 

250 which_idnum = recipient.primary_idnum 

251 idnum_object = task.patient.get_idnum_object(which_idnum) 

252 

253 project = self.get_project(recipient) 

254 fieldmap = self.get_fieldmap(recipient) 

255 

256 if project.is_longitudinal(): 

257 if not all(fieldmap.events.values()): 

258 raise RedcapExportException(MISSING_EVENT_TAG_OR_ATTRIBUTE) 

259 

260 existing_records = self._get_existing_records(project, fieldmap) 

261 existing_record_id = self._get_existing_record_id( 

262 existing_records, fieldmap, idnum_object.idnum_value 

263 ) 

264 

265 if existing_record_id is None: 

266 uploader_class = RedcapNewRecordUploader 

267 else: 

268 uploader_class = RedcapUpdatedRecordUploader # type: ignore[assignment] # noqa: E501 

269 

270 try: 

271 instrument_name = fieldmap.instruments[task.tablename] 

272 except KeyError: 

273 raise RedcapExportException( 

274 ( 

275 f"Instrument for task '{task.tablename}' is missing from " 

276 f"the fieldmap" 

277 ) 

278 ) 

279 

280 record_id_fieldname = fieldmap.record["redcap_field"] 

281 

282 next_instance_id = self._get_next_instance_id( 

283 existing_records, 

284 instrument_name, 

285 record_id_fieldname, 

286 existing_record_id, 

287 ) 

288 

289 uploader = uploader_class(req, project) 

290 

291 new_record_id = uploader.upload( 

292 task, 

293 existing_record_id, 

294 next_instance_id, 

295 fieldmap, 

296 idnum_object.idnum_value, 

297 ) 

298 

299 exported_task_redcap.redcap_record_id = new_record_id # type: ignore[assignment] # noqa: E501 

300 exported_task_redcap.redcap_instrument_name = instrument_name 

301 exported_task_redcap.redcap_instance_id = next_instance_id 

302 

303 @staticmethod 

304 def _get_existing_records( 

305 project: redcap.project.Project, fieldmap: RedcapFieldmap 

306 ) -> "DataFrame": 

307 """ 

308 Returns a Pandas data frame containing existing REDCap records for this 

309 project, for instruments we are interested in. 

310 

311 Args: 

312 project: 

313 a :class:`redcap.project.Project` 

314 fieldmap: 

315 a :class:`RedcapFieldmap` 

316 """ 

317 # Arguments to pandas read_csv() 

318 

319 type_dict = { 

320 # otherwise pandas may infer as int or str 

321 fieldmap.record["redcap_field"]: str 

322 } 

323 

324 df_kwargs = { 

325 "index_col": None, # don't index by record_id 

326 "dtype": type_dict, 

327 } 

328 

329 forms = ( 

330 fieldmap.instrument_names() 

331 + [fieldmap.patient["instrument"]] 

332 + [fieldmap.record["instrument"]] 

333 ) 

334 

335 try: 

336 records = project.export_records( 

337 format="df", forms=forms, df_kwargs=df_kwargs 

338 ) 

339 except EmptyDataError: 

340 # Should not happen, but in case of PyCap failing to catch this... 

341 return DataFrame() 

342 except redcap.RedcapError as e: 

343 raise RedcapExportException(str(e)) 

344 

345 return records 

346 

347 @staticmethod 

348 def _get_existing_record_id( 

349 records: "DataFrame", fieldmap: RedcapFieldmap, idnum_value: int 

350 ) -> Optional[str]: 

351 """ 

352 Returns the ID of an existing record that matches a specific 

353 patient, if one can be found. 

354 

355 Args: 

356 records: 

357 records retrieved from REDCap; Pandas data frame from 

358 :meth:`_get_existing_records` 

359 fieldmap: 

360 :class:`RedcapFieldmap` 

361 idnum_value: 

362 CamCOPS patient ID number 

363 

364 Returns: 

365 REDCap record ID or ``None`` 

366 """ 

367 

368 if records.empty: 

369 return None 

370 

371 patient_id_fieldname = fieldmap.patient["redcap_field"] 

372 

373 if patient_id_fieldname not in records: 

374 raise RedcapExportException( 

375 ( 

376 f"Field '{patient_id_fieldname}' does not exist in " 

377 f"REDCap. Is the 'patient' tag in the fieldmap correct?" 

378 ) 

379 ) 

380 

381 with_identifier = records[patient_id_fieldname] == idnum_value 

382 

383 if len(records[with_identifier]) == 0: 

384 return None 

385 

386 return records[with_identifier].iat[0, 0] 

387 

388 @staticmethod 

389 def _get_next_instance_id( 

390 records: "DataFrame", 

391 instrument: str, 

392 record_id_fieldname: str, 

393 existing_record_id: Optional[str], 

394 ) -> int: 

395 """ 

396 Returns the next REDCap record ID to use for a particular instrument, 

397 including for a repeating instrument (the previous highest ID plus 1, 

398 or 1 if none can be found). 

399 

400 Args: 

401 records: 

402 records retrieved from REDCap; Pandas data frame from 

403 :meth:`_get_existing_records` 

404 instrument: 

405 instrument name 

406 existing_record_id: 

407 ID of existing record 

408 """ 

409 if existing_record_id is None: 

410 return 1 

411 

412 if record_id_fieldname not in records: 

413 raise RedcapExportException( 

414 ( 

415 f"Field '{record_id_fieldname}' does not exist in REDCap. " 

416 f"Is the 'record' tag in the fieldmap correct?" 

417 ) 

418 ) 

419 

420 previous_instances = records[ 

421 (records["redcap_repeat_instrument"] == instrument) 

422 & (records[record_id_fieldname] == existing_record_id) 

423 ] 

424 

425 if len(previous_instances) == 0: 

426 return 1 

427 

428 return int(previous_instances.max()["redcap_repeat_instance"] + 1) 

429 

430 def get_fieldmap(self, recipient: ExportRecipient) -> RedcapFieldmap: 

431 """ 

432 Returns the relevant :class:`RedcapFieldmap`. 

433 

434 Args: 

435 recipient: 

436 an 

437 :class:`camcops_server.cc_modules.cc_exportmodels.ExportRecipient` 

438 """ 

439 fieldmap = RedcapFieldmap(self.get_fieldmap_filename(recipient)) 

440 

441 return fieldmap 

442 

443 @staticmethod 

444 def get_fieldmap_filename(recipient: ExportRecipient) -> str: 

445 """ 

446 Returns the name of the XML file containing our fieldmap details, or 

447 raises :exc:`RedcapExportException`. 

448 

449 Args: 

450 recipient: 

451 an 

452 :class:`camcops_server.cc_modules.cc_exportmodels.ExportRecipient` 

453 """ 

454 filename = recipient.redcap_fieldmap_filename 

455 if filename is None: 

456 raise RedcapExportException( 

457 f"{ConfigParamExportRecipient.REDCAP_FIELDMAP_FILENAME} " 

458 f"is not set in the config file" 

459 ) 

460 

461 if filename == "": 

462 raise RedcapExportException( 

463 f"{ConfigParamExportRecipient.REDCAP_FIELDMAP_FILENAME} " 

464 f"is empty in the config file" 

465 ) 

466 

467 return filename 

468 

469 @staticmethod 

470 def get_project(recipient: ExportRecipient) -> redcap.project.Project: 

471 """ 

472 Returns the :class:`redcap.project.Project`. 

473 

474 Args: 

475 recipient: 

476 an 

477 :class:`camcops_server.cc_modules.cc_exportmodels.ExportRecipient` 

478 """ 

479 try: 

480 project = redcap.project.Project( 

481 recipient.redcap_api_url, recipient.redcap_api_key 

482 ) 

483 except redcap.RedcapError as e: 

484 raise RedcapExportException(str(e)) 

485 

486 return project 

487 

488 

489class RedcapRecordStatus(Enum): 

490 """ 

491 Corresponds to valid values of Form Status -> Complete? field in REDCap 

492 """ 

493 

494 INCOMPLETE = 0 

495 UNVERIFIED = 1 

496 COMPLETE = 2 

497 

498 

499class RedcapUploader(object): 

500 """ 

501 Uploads records and files into REDCap, transforming the fields via the 

502 fieldmap. 

503 

504 Abstract base class. 

505 

506 Knows nothing about ExportedTaskRedcap, ExportedTask, ExportRecipient 

507 """ 

508 

509 def __init__( 

510 self, req: "CamcopsRequest", project: "redcap.project.Project" 

511 ) -> None: 

512 """ 

513 

514 Args: 

515 req: 

516 a :class:`camcops_server.cc_modules.cc_request.CamcopsRequest` 

517 project: 

518 a :class:`redcap.project.Project` 

519 """ 

520 self.req = req 

521 self.project = project 

522 self.project_info = project.export_project_info() 

523 

524 def get_record_id(self, existing_record_id: Optional[str]) -> str: 

525 """ 

526 Returns the REDCap record ID to use. 

527 

528 Args: 

529 existing_record_id: highest existing record ID, if known 

530 """ 

531 raise NotImplementedError("implement in subclass") 

532 

533 @property 

534 def return_content(self) -> str: 

535 """ 

536 The ``return_content`` argument to be passed to 

537 :meth:`redcap.project.Project.import_records`. Can be: 

538 

539 - ``count`` [default] - the number of records imported 

540 - ``ids`` - a list of all record IDs that were imported 

541 - ``auto_ids`` = (used only when ``forceAutoNumber=true``) a list of 

542 pairs of all record IDs that were imported, includes the new ID 

543 created and the ID value that was sent in the API request 

544 (e.g., 323,10). 

545 

546 Note (2020-01-27) that it can return e.g. ``15-30,0``, i.e. the ID 

547 values can be non-integer. 

548 """ 

549 raise NotImplementedError("implement in subclass") 

550 

551 @property 

552 def force_auto_number(self) -> bool: 

553 """ 

554 Should we force auto-numbering of records in REDCap? 

555 """ 

556 raise NotImplementedError("implement in subclass") 

557 

558 def get_new_record_id(self, record_id: str, response: List[str]) -> str: 

559 """ 

560 Returns the ID of the new (or updated) REDCap record. 

561 

562 Args: 

563 record_id: 

564 existing record ID 

565 response: 

566 response from :meth:`redcap.project.Project.import_records` 

567 """ 

568 raise NotImplementedError("implement in subclass") 

569 

570 @staticmethod 

571 def log_success(record_id: str) -> None: 

572 """ 

573 Report upload success to the Python log. 

574 

575 Args: 

576 record_id: REDCap record ID 

577 """ 

578 raise NotImplementedError("implement in subclass") 

579 

580 @property 

581 def autonumbering_enabled(self) -> bool: 

582 """ 

583 Does this REDCap project have record autonumbering enabled? 

584 """ 

585 return self.project_info["record_autonumbering_enabled"] 

586 

587 def upload( 

588 self, 

589 task: "Task", 

590 existing_record_id: Optional[str], 

591 next_instance_id: int, 

592 fieldmap: RedcapFieldmap, 

593 idnum_value: int, 

594 ) -> str: 

595 """ 

596 Uploads a CamCOPS task to REDCap. 

597 

598 Args: 

599 task: 

600 :class:`camcops_server.cc_modules.cc_task.Task` to be uploaded 

601 existing_record_id: 

602 REDCap ID of the existing record, if there is one 

603 next_instance_id: 

604 REDCap instance ID to be used for a repeating instrument 

605 fieldmap: 

606 :class:`RedcapFieldmap` 

607 idnum_value: 

608 CamCOPS patient ID number 

609 

610 Returns: 

611 str: REDCap record ID of the record that was created or updated 

612 

613 """ 

614 complete_status = RedcapRecordStatus.INCOMPLETE 

615 

616 if task.is_complete(): 

617 complete_status = RedcapRecordStatus.COMPLETE 

618 instrument_name = fieldmap.instruments[task.tablename] 

619 record_id_fieldname = fieldmap.record["redcap_field"] 

620 

621 record_id = self.get_record_id(existing_record_id) 

622 

623 record = { 

624 record_id_fieldname: record_id, 

625 "redcap_repeat_instrument": instrument_name, 

626 # https://community.projectredcap.org/questions/74561/unexpected-behaviour-with-import-records-repeat-in.html # noqa 

627 # REDCap won't create instance IDs automatically so we have to 

628 # assume no one else is writing to this record 

629 "redcap_repeat_instance": next_instance_id, 

630 f"{instrument_name}_complete": complete_status.value, 

631 "redcap_event_name": fieldmap.events[task.tablename], 

632 } 

633 

634 self.transform_fields(record, task, fieldmap.fields[task.tablename]) 

635 

636 import_kwargs = { 

637 "return_content": self.return_content, 

638 "force_auto_number": self.force_auto_number, 

639 } 

640 

641 response = self.upload_record(record, **import_kwargs) 

642 

643 new_record_id = self.get_new_record_id(record_id, response) # type: ignore[arg-type] # noqa: E501 

644 

645 # We don't mark the patient record as complete - it could be part of 

646 # a larger form. We don't require it to be complete. 

647 patient_record = { 

648 record_id_fieldname: new_record_id, 

649 fieldmap.patient["redcap_field"]: idnum_value, 

650 } 

651 self.upload_record(patient_record) 

652 

653 file_dict: dict[str, Any] = {} 

654 self.transform_fields(file_dict, task, fieldmap.files[task.tablename]) 

655 

656 self.upload_files( 

657 task, 

658 new_record_id, 

659 next_instance_id, 

660 file_dict, 

661 event=fieldmap.events[task.tablename], 

662 ) 

663 

664 self.log_success(new_record_id) 

665 

666 return new_record_id 

667 

668 def upload_record( 

669 self, record: Dict[str, Any], **kwargs: Any 

670 ) -> Union[Dict, List, str]: 

671 """ 

672 Uploads a REDCap record via the pycap 

673 :func:`redcap.project.Project.import_record` function. Returns its 

674 response. 

675 """ 

676 try: 

677 response = self.project.import_records([record], **kwargs) 

678 except redcap.RedcapError as e: 

679 raise RedcapExportException(str(e)) 

680 

681 return response 

682 

683 def upload_files( 

684 self, 

685 task: "Task", 

686 record_id: Union[int, str], 

687 repeat_instance: int, 

688 file_dict: Dict[str, bytes], 

689 event: Optional[str] = None, 

690 ) -> None: 

691 """ 

692 Uploads files attached to a task (e.g. a PDF of the CamCOPS task). 

693 

694 Args: 

695 task: 

696 the :class:`camcops_server.cc_modules.cc_task.Task` 

697 record_id: 

698 the REDCap record ID 

699 repeat_instance: 

700 instance number for repeating instruments 

701 file_dict: 

702 dictionary mapping filename to file contents 

703 event: 

704 for longitudinal projects, specify the unique event here 

705 

706 Raises: 

707 :exc:`RedcapExportException` 

708 """ 

709 for fieldname, value in file_dict.items(): 

710 with io.BytesIO(value) as file_obj: 

711 filename = f"{task.tablename}_{record_id}_{fieldname}" 

712 

713 try: 

714 self.project.import_file( 

715 record_id, 

716 fieldname, 

717 filename, 

718 file_obj, 

719 event=event, 

720 repeat_instance=repeat_instance, 

721 ) 

722 # ValueError if the field does not exist or is not 

723 # a file field 

724 except (redcap.RedcapError, ValueError) as e: 

725 raise RedcapExportException(str(e)) 

726 

727 def transform_fields( 

728 self, 

729 field_dict: Dict[str, Any], 

730 task: "Task", 

731 formula_dict: Dict[str, str], 

732 ) -> None: 

733 """ 

734 Uses the definitions from the fieldmap XML to set up field values to be 

735 exported to REDCap. 

736 

737 Args: 

738 field_dict: 

739 Exported field values go here (the dictionary is modified). 

740 task: 

741 the :class:`camcops_server.cc_modules.cc_task.Task` 

742 formula_dict: 

743 dictionary (from the XML information) mapping REDCap field 

744 name to a "formula". The formula is applied to extract data 

745 from the task in a flexible way. 

746 """ 

747 extra_symbols = self.get_extra_symbols() 

748 

749 symbol_table = make_symbol_table(task=task, **extra_symbols) 

750 interpreter = Interpreter(symtable=symbol_table) 

751 

752 for redcap_field, formula in formula_dict.items(): 

753 v = interpreter(f"{formula}", show_errors=True) 

754 if interpreter.error: 

755 message = "\n".join([e.msg for e in interpreter.error]) 

756 raise RedcapExportException( 

757 ( 

758 f"Fieldmap:\n" 

759 f"Error in formula '{formula}': {message}\n" 

760 f"Task: '{task.tablename}'\n" 

761 f"REDCap field: '{redcap_field}'\n" 

762 ) 

763 ) 

764 field_dict[redcap_field] = v 

765 

766 def get_extra_symbols(self) -> Dict[str, Any]: 

767 """ 

768 Returns a dictionary made available to the ``asteval`` interpreter. 

769 These become variables that the system administrator can refer to in 

770 their fieldmap XML; see :ref:`REDCap export <redcap>`. 

771 """ 

772 return dict( 

773 format_datetime=format_datetime, 

774 DateFormat=DateFormat, 

775 request=self.req, 

776 ) 

777 

778 

779class RedcapNewRecordUploader(RedcapUploader): 

780 """ 

781 Creates a new REDCap record. 

782 """ 

783 

784 @property 

785 def force_auto_number(self) -> bool: 

786 return self.autonumbering_enabled 

787 

788 @property 

789 def return_content(self) -> str: 

790 if self.autonumbering_enabled: 

791 # import_records returns ["<redcap record id>, 0"] 

792 return "auto_ids" 

793 

794 # import_records returns {'count': 1} 

795 return "count" 

796 

797 # noinspection PyUnusedLocal 

798 def get_record_id(self, existing_record_id: str) -> str: 

799 """ 

800 Get the record ID to send to REDCap when importing records 

801 """ 

802 if self.autonumbering_enabled: 

803 # Is ignored but we still need to set this to something 

804 return "0" 

805 

806 return self.project.generate_next_record_name() 

807 

808 def get_new_record_id(self, record_id: str, response: List[str]) -> str: 

809 """ 

810 For autonumbering, read the generated record ID from the 

811 response. Otherwise we already have it. 

812 """ 

813 if not self.autonumbering_enabled: 

814 return record_id 

815 

816 id_pair = response[0] 

817 

818 record_id = id_pair.rsplit(",")[0] 

819 

820 return record_id 

821 

822 @staticmethod 

823 def log_success(record_id: str) -> None: 

824 log.info(f"Created new REDCap record {record_id}") 

825 

826 

827class RedcapUpdatedRecordUploader(RedcapUploader): 

828 """ 

829 Updates an existing REDCap record. 

830 """ 

831 

832 force_auto_number = False 

833 # import_records returns {'count': 1} 

834 return_content = "count" 

835 

836 # noinspection PyMethodMayBeStatic 

837 def get_record_id(self, existing_record_id: str) -> str: 

838 return existing_record_id 

839 

840 # noinspection PyMethodMayBeStatic,PyUnusedLocal 

841 def get_new_record_id(self, old_record_id: str, response: Any) -> str: 

842 return old_record_id 

843 

844 @staticmethod 

845 def log_success(record_id: str) -> None: 

846 log.info(f"Updated REDCap record {record_id}")