15.2.119. camcops_server.cc_modules.cc_fhir

camcops_server/cc_modules/cc_fhir.py

---

Copyright (C) 2012, University of Cambridge, Department of Psychiatry. Created by Rudolf Cardinal (rnc1001@cam.ac.uk).

This file is part of CamCOPS.

CamCOPS is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

CamCOPS is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.

---

Implements communication with a FHIR server.

Fast Healthcare Interoperability Resources

https://www.hl7.org/fhir/

Our implementation exports:

• patients as FHIR Patient resources;

• task concepts as FHIR Questionnaire resources;

• task instances as FHIR QuestionnaireResponse resources.

Currently PHQ9 and APEQPT (anonymous) are supported. Each task and patient (if appropriate is sent to the FHIR server in a single “transaction” Bundle). The resources are given a unique identifier based on the URL of the CamCOPS server.

We use the Python client https://github.com/smart-on-fhir/client-py/. This only supports one version of the FHIR specification (currently 4.0.1).

Testing: HAPI FHIR server locally

To test with a HAPI FHIR server locally, which was installed from instructions at https://github.com/hapifhir/hapi-fhir-jpaserver-starter (Docker). Most simply:

docker run -p 8080:8080 hapiproject/hapi:latest

with the following entry in the CamCOPS export recipient configuration:

FHIR_API_URL = http://localhost:8080/fhir

To inspect it while it’s running (apart from via its log):

• Browse to (by default) http://localhost:8080/

  • then e.g. Patient –> Search, which is a pretty version of http://localhost:8080/fhir/Patient?_pretty=true;

  • Questionnaire –> Search, which is a pretty version of http://localhost:8080/fhir/Questionnaire?_pretty=true.

• Can also browse to (by default) http://localhost:8080/fhir/metadata

Testing: Other

There are also public sandboxes at:

• http://hapi.fhir.org/baseR4

• https://r4.smarthealthit.org (errors when exporting questionnaire responses)

Intermittent problem with If-None-Exist

This problem occurs intermittently:

• “Failed to CREATE resource with match URL … because this search matched 2 resources” – an OperationOutcome error.

  At https://groups.google.com/g/hapi-fhir/c/8OolMOpf8SU, it says (for an error with 40 resources) “You can only do a conditional create if there are 0..1 existing resources on the server that match the criteria, and in this case there are 40.” But I think that is an error in the explanation.

  Proper documentation for ifNoneExist (Python client) or If-None-Exist (FHIR itself) is at https://www.hl7.org/fhir/http.html#ccreate.

  I suspect that “0..1” comment relates to “cardinality” (https://www.hl7.org/fhir/bundle.html#resource), which is how many times the attribute can appear in a resource type (https://www.hl7.org/fhir/conformance-rules.html#cardinality); that is, this statement is optional. It would clearly be silly if it meant “create if no more than 1 exist”!

  However, the “Failed to CREATE” problem seemed to go away. It does work fine, and you get status messages of “200 OK” rather than “201 Created” if you try to insert the same information again (SELECT * FROM _exported_task_fhir_entry;).

• This is a concurrency problem (they dispute “bug”) in the HAPI FHIR implementation. See our bug report at https://github.com/hapifhir/hapi-fhir/issues/3141.

• The suggested fix is a “unique combo search index parameter”, as per https://smilecdr.com/docs/fhir_repository/custom_search_parameters.html#uniqueness.

• However, that seems implementation-specific (e.g. HAPI FHIR, SmileCDR). A specific value of http://hapifhir.io/fhir/StructureDefinition/sp-unique must be used. Specimen code is https://hapifhir.io/hapi-fhir/apidocs/hapi-fhir-base/src-html/ca/uhn/fhir/util/HapiExtensions.html.

• Instead, we could force a FHIR export for a given recipient to occur in serial (particularly as other FHIR implementations may have this bug).

  Celery doesn’t allow you to send multiple jobs and enforce that they all happen via the same worker (https://docs.celeryproject.org/en/stable/userguide/calling.html). However, our exports already start (mostly!) as “one recipient, one job”, via camcops_server.cc_modules.celery.export_to_recipient_backend() (see camcops_server.cc_modules.celery.get_celery_settings_dict()).

  The tricky bit is that push exports require back-end single-task jobs, so they are hard to de-parallelize.

  So we use a carefully sequenced file lock; see camcops_server.cc_modules.cc_export.export_task().

class camcops_server.cc_modules.cc_fhir.FHIRAnswerType(value)

An enum for value type keys of QuestionnaireResponseItemAnswer.

class camcops_server.cc_modules.cc_fhir.FHIRAnsweredQuestion(qname: str, qtext: str, qtype: camcops_server.cc_modules.cc_fhir.FHIRQuestionType, answer_type: camcops_server.cc_modules.cc_fhir.FHIRAnswerType, answer: Any, answer_options: Optional[Dict[Any, str]] = None)

Represents a question in a questionnaire-based task. That includes both the abstract aspects:

• What kind of question is it (e.g. multiple-choice, real-value answer, text)? That can go into some detail, e.g. possible responses for a multiple-choice question. (Thus, the FHIR Questionnaire.)

and the concrete aspects:

• what is the response/answer for a specific task instance? (Thus, the FHIR QuestionnaireResponse.)

Used for autodiscovery.

__init__(qname: str, qtext: str, qtype: camcops_server.cc_modules.cc_fhir.FHIRQuestionType, answer_type: camcops_server.cc_modules.cc_fhir.FHIRAnswerType, answer: Any, answer_options: Optional[Dict[Any, str]] = None) → None

Parameters

• qname – Name (task attribute name) of the question, e.g. “q1”.

• qtext – Question text (e.g. “How was your day?”).

• qtype – Question type, e.g. multiple-choice.

• answer_type – Answer type, e.g. integer.

• answer – Actual answer.

• answer_options – For multiple-choice questions (MCQs), a dictionary mapping answer codes to human-legible display text.

property is_mcq: bool

Is this a multiple-choice question?

questionnaire_item() → Dict

Returns a JSON/dict representation of a FHIR QuestionnaireItem.

questionnaire_response_item() → Dict

Returns a JSON/dict representation of a FHIR QuestionnaireResponseItem.

class camcops_server.cc_modules.cc_fhir.FHIRQuestionType(value)

An enum for value type keys of QuestionnaireResponseItemAnswer.

class camcops_server.cc_modules.cc_fhir.FhirTaskExporter(request: CamcopsRequest, exported_task_fhir: ExportedTaskFhir)

Class that knows how to export a single task to FHIR.

__init__(request: CamcopsRequest, exported_task_fhir: ExportedTaskFhir) → None

export_task() → None

Export a single task to the server, with associated patient information if the task has an associated patient.

parse_response(response: Dict) → None

Parse the response from the FHIR server to which we have sent our task. The response looks something like this:

{
  "resourceType": "Bundle",
  "id": "cae48957-e7e6-4649-97f8-0a882076ad0a",
  "type": "transaction-response",
  "link": [
    {
      "relation": "self",
      "url": "http://localhost:8080/fhir"
    }
  ],
  "entry": [
    {
      "response": {
        "status": "200 OK",
        "location": "Patient/1/_history/1",
        "etag": "1"
      }
    },
    {
      "response": {
        "status": "200 OK",
        "location": "Questionnaire/26/_history/1",
        "etag": "1"
      }
    },
    {
      "response": {
        "status": "201 Created",
        "location": "QuestionnaireResponse/42/_history/1",
        "etag": "1",
        "lastModified": "2021-05-24T09:30:11.098+00:00"
      }
    }
  ]
}

The server’s reply contains a Bundle (https://www.hl7.org/fhir/bundle.html), which is a container for resources. Here, the bundle contains entry objects (https://www.hl7.org/fhir/bundle-definitions.html#Bundle.entry).

camcops_server.cc_modules.cc_fhir.fhir_observation_component_from_snomed(req: CamcopsRequest, expr: camcops_server.cc_modules.cc_snomed.SnomedExpression) → Dict

Returns a FHIR ObservationComponent (as a dict in JSON format) for a SNOMED CT expression.

camcops_server.cc_modules.cc_fhir.fhir_pk_identifier(req: CamcopsRequest, tablename: str, pk: int, value_within_task: str) → fhirclient.models.identifier.Identifier

Creates a “fallback” identifier – this is poor, but allows unique identification of anything (such as a patient with no proper ID numbers) based on its CamCOPS table name and server PK.

camcops_server.cc_modules.cc_fhir.fhir_reference_from_identifier(identifier: fhirclient.models.identifier.Identifier) → str

Returns a reference to a specific FHIR identifier.

camcops_server.cc_modules.cc_fhir.fhir_system_value(system: str, value: str) → str

How FHIR expresses system/value pairs.

camcops_server.cc_modules.cc_fhir.fhir_sysval_from_id(identifier: fhirclient.models.identifier.Identifier) → str

How FHIR expresses system/value pairs.

camcops_server.cc_modules.cc_fhir.make_fhir_bundle_entry(resource_type_url: str, identifier: fhirclient.models.identifier.Identifier, resource: Dict, identifier_is_list: bool = True) → Dict

Builds a FHIR BundleEntry, as a JSON dict.

This also takes care of the identifier, by ensuring (a) that the resource is labelled with the identifier, and (b) that the BundleEntryRequest has an ifNoneExist condition referring to that identifier.