Coverage for cc_modules/cc_fhir.py : 86%

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
3"""camcops_server/cc_modules/cc_fhir.py
5===============================================================================
7 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com).
9 This file is part of CamCOPS.
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.
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.
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/>.
24===============================================================================
26**Implements communication with a FHIR server.**
27"""
29from typing import Dict, TYPE_CHECKING
31from fhirclient import client
32from fhirclient.models.bundle import Bundle
33from requests.exceptions import HTTPError
35if TYPE_CHECKING:
36 from camcops_server.cc_modules.cc_exportmodels import ExportedTaskFhir
37 from camcops_server.cc_modules.cc_request import CamcopsRequest
40class FhirExportException(Exception):
41 pass
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
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 }
61 self.client = client.FHIRClient(settings=settings)
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
69 # TODO: Version of questionnaire?
71 patient_entry = self.task.patient.get_fhir_bundle_entry(
72 self.request,
73 self.exported_task.recipient
74 )
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")
84 bundle_entries = [patient_entry] + task_entries
86 bundle = Bundle(jsondict={
87 "type": "transaction",
88 "entry": bundle_entries,
89 })
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")
102 self.parse_response(response)
103 except HTTPError as e:
104 raise FhirExportException(
105 f"The server returned an error: {e.response.text}")
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)
148 if bundle.entry is not None:
149 self._save_exported_entries(bundle)
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
164 self.request.dbsession.add(saved_entry)