Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1#!/usr/bin/env python 

2 

3"""camcops_server/cc_modules/cc_fhir.py 

4 

5=============================================================================== 

6 

7 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com). 

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 <http://www.gnu.org/licenses/>. 

23 

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

25 

26**Implements communication with a FHIR server.** 

27""" 

28 

29from typing import Dict, TYPE_CHECKING 

30 

31from fhirclient import client 

32from fhirclient.models.bundle import Bundle 

33from requests.exceptions import HTTPError 

34 

35if TYPE_CHECKING: 

36 from camcops_server.cc_modules.cc_exportmodels import ExportedTaskFhir 

37 from camcops_server.cc_modules.cc_request import CamcopsRequest 

38 

39 

40class FhirExportException(Exception): 

41 pass 

42 

43 

44class FhirTaskExporter(object): 

45 def __init__(self, 

46 request: "CamcopsRequest", 

47 exported_task_fhir: "ExportedTaskFhir") -> None: 

48 self.request = request 

49 self.exported_task = exported_task_fhir.exported_task 

50 self.exported_task_fhir = exported_task_fhir 

51 

52 self.recipient = self.exported_task.recipient 

53 self.task = self.exported_task.task 

54 settings = { 

55 "app_id": "camcops", 

56 "api_base": self.recipient.fhir_api_url, 

57 "app_secret": self.recipient.fhir_app_secret, 

58 "launch_token": self.recipient.fhir_launch_token, 

59 } 

60 

61 self.client = client.FHIRClient(settings=settings) 

62 

63 def export_task(self) -> None: 

64 # TODO: Authentication 

65 # TODO: Server capability statement 

66 # TODO: Anonymous tasks 

67 # TODO: Missing API URL in config 

68 

69 # TODO: Version of questionnaire? 

70 

71 patient_entry = self.task.patient.get_fhir_bundle_entry( 

72 self.request, 

73 self.exported_task.recipient 

74 ) 

75 

76 try: 

77 task_entries = self.task.get_fhir_bundle_entries( 

78 self.request, 

79 self.exported_task.recipient 

80 ) 

81 except NotImplementedError: 

82 raise FhirExportException("No FHIR support for this task") 

83 

84 bundle_entries = [patient_entry] + task_entries 

85 

86 bundle = Bundle(jsondict={ 

87 "type": "transaction", 

88 "entry": bundle_entries, 

89 }) 

90 

91 # TODO: Can raise Exception 

92 try: 

93 response = bundle.create(self.client.server) 

94 if response is None: 

95 # Not sure this will ever happen. 

96 # fhirabstractresource.py create() says it returns 

97 # "None or the response JSON on success" but an exception will 

98 # already have been raised if there was a failure 

99 raise FhirExportException( 

100 "The server unexpectedly returned an OK, empty response") 

101 

102 self.parse_response(response) 

103 except HTTPError as e: 

104 raise FhirExportException( 

105 f"The server returned an error: {e.response.text}") 

106 

107 def parse_response(self, response: Dict) -> None: 

108 """ 

109 Response looks something like this: 

110 { 

111 'resourceType': 'Bundle', 

112 'id': 'cae48957-e7e6-4649-97f8-0a882076ad0a', 

113 'type': 'transaction-response', 

114 'link': [ 

115 { 

116 'relation': 'self', 

117 'url': 'http://localhost:8080/fhir' 

118 } 

119 ], 

120 'entry': [ 

121 { 

122 'response': { 

123 'status': '200 OK', 

124 'location': 'Patient/1/_history/1', 

125 'etag': '1' 

126 } 

127 }, 

128 { 

129 'response': { 

130 'status': '200 OK', 

131 'location': 'Questionnaire/26/_history/1', 

132 'etag': '1' 

133 } 

134 }, 

135 { 

136 'response': { 

137 'status': '201 Created', 

138 'location': 'QuestionnaireResponse/42/_history/1', 

139 'etag': '1', 

140 'lastModified': '2021-05-24T09:30:11.098+00:00' 

141 } 

142 } 

143 ] 

144 } 

145 """ 

146 bundle = Bundle(jsondict=response) 

147 

148 if bundle.entry is not None: 

149 self._save_exported_entries(bundle) 

150 

151 def _save_exported_entries(self, bundle: Bundle) -> None: 

152 from camcops_server.cc_modules.cc_exportmodels import ( 

153 ExportedTaskFhirEntry 

154 ) 

155 for entry in bundle.entry: 

156 saved_entry = ExportedTaskFhirEntry() 

157 saved_entry.exported_task_fhir_id = self.exported_task_fhir.id 

158 saved_entry.status = entry.response.status 

159 saved_entry.location = entry.response.location 

160 saved_entry.etag = entry.response.etag 

161 if entry.response.lastModified is not None: 

162 saved_entry.last_modified = entry.response.lastModified.date 

163 

164 self.request.dbsession.add(saved_entry)