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 try: 

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

63 except Exception as e: 

64 raise FhirExportException(str(e)) 

65 

66 def export_task(self) -> None: 

67 # TODO: Server capability statement 

68 # TODO: Question codes 

69 # TODO: Version of questionnaire? 

70 

71 bundle_entries = [] 

72 

73 if self.task.has_patient: 

74 patient_entry = self.task.patient.get_fhir_bundle_entry( 

75 self.request, 

76 self.exported_task.recipient 

77 ) 

78 bundle_entries.append(patient_entry) 

79 

80 try: 

81 task_entries = self.task.get_fhir_bundle_entries( 

82 self.request, 

83 self.exported_task.recipient 

84 ) 

85 bundle_entries += task_entries 

86 

87 except NotImplementedError as e: 

88 raise FhirExportException(str(e)) 

89 

90 bundle = Bundle(jsondict={ 

91 "type": "transaction", 

92 "entry": bundle_entries, 

93 }) 

94 

95 try: 

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

97 if response is None: 

98 # Not sure this will ever happen. 

99 # fhirabstractresource.py create() says it returns 

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

101 # already have been raised if there was a failure 

102 raise FhirExportException( 

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

104 

105 self.parse_response(response) 

106 except HTTPError as e: 

107 raise FhirExportException( 

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

109 

110 except Exception as e: 

111 # Unfortunate that fhirclient doesn't give us anything more 

112 # specific 

113 raise FhirExportException(e) 

114 

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

116 """ 

117 Response looks something like this: 

118 { 

119 'resourceType': 'Bundle', 

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

121 'type': 'transaction-response', 

122 'link': [ 

123 { 

124 'relation': 'self', 

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

126 } 

127 ], 

128 'entry': [ 

129 { 

130 'response': { 

131 'status': '200 OK', 

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

133 'etag': '1' 

134 } 

135 }, 

136 { 

137 'response': { 

138 'status': '200 OK', 

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

140 'etag': '1' 

141 } 

142 }, 

143 { 

144 'response': { 

145 'status': '201 Created', 

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

147 'etag': '1', 

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

149 } 

150 } 

151 ] 

152 } 

153 """ 

154 bundle = Bundle(jsondict=response) 

155 

156 if bundle.entry is not None: 

157 self._save_exported_entries(bundle) 

158 

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

160 from camcops_server.cc_modules.cc_exportmodels import ( 

161 ExportedTaskFhirEntry 

162 ) 

163 for entry in bundle.entry: 

164 saved_entry = ExportedTaskFhirEntry() 

165 saved_entry.exported_task_fhir_id = self.exported_task_fhir.id 

166 saved_entry.status = entry.response.status 

167 saved_entry.location = entry.response.location 

168 saved_entry.etag = entry.response.etag 

169 if entry.response.lastModified is not None: 

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

171 

172 self.request.dbsession.add(saved_entry)