Write a result

Find the case a specimen belongs to and write the result event to it.

This page follows Setup in the Lab results series. It describes the service that writes a result to DHIS2. The services on the HL7 v2 and FHIR pages invoke it after translating the laboratory's codes to DHIS2 option codes.

The service accepts four values: the specimen ID, the test and result as DHIS2 option codes, and the result date in ISO 8601 format.

Find the case

The specimen ID is stored on the lab request event, so the lookup queries the lab request stage of the program and filters on the specimen ID data element. The enrollment and organisation unit of the matching event identify the case:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class WriteLabResult(Service):
    name = 'dhis2.lab.write-result'
    input = 'specimen_id', 'test', 'result', 'result_date'

    def find_case(self, specimen_id:'str') -> 'dict | None':
        program = self.config.dhis2.program
        element = self.config.dhis2.element

        # Query the lab request stage for the event that carries this specimen ID ..
        params = {
            'program': program.id,
            'programStage': program.stage_lab_request,
            'filter': f'{element.specimen_id}:eq:{specimen_id}',
            'orgUnitMode': 'ACCESSIBLE',
            'fields': 'enrollment,orgUnit',
        }

        response = self.rest['DHIS2 Events'].get(params=params)
        found = response.data['events']

        # .. there is no case for this specimen if nothing matched.
        if not found:
            return None
        else:
            return found[0]

orgUnitMode=ACCESSIBLE searches every organisation unit the API user has access to, so the service does not need to know where the case was registered. DHIS2 2.41 and later return the matching events under events. DHIS2 2.40 returns them under instances.

Build the event

The result is one event on the lab result stage of the enrollment the case belongs to. The specimen ID is written to the result event as well, so that a corrected result can be matched to the event it replaces:

    def build_event(self, case:'dict') -> 'dict':

        request = self.request.input
        program = self.config.dhis2.program
        element = self.config.dhis2.element

        data_values = [
            {'dataElement': element.specimen_id, 'value': request.specimen_id},
            {'dataElement': element.test, 'value': request.test},
            {'dataElement': element.result, 'value': request.result},
        ]

        event = {
            'program': program.id,
            'programStage': program.stage_lab_result,
            'enrollment': case['enrollment'],
            'orgUnit': case['orgUnit'],
            'status': 'COMPLETED',
            'occurredAt': request.result_date,
            'dataValues': data_values,
        }

        return event

Write the event

The event is posted to the tracker importer and the import status is returned to the caller:

    def handle(self) -> 'None':

        request = self.request.input
        case = self.find_case(request.specimen_id)

        if case is None:
            self.response.payload = {'is_found': False}
            return

        event = self.build_event(case)
        payload = {'events': [event]}
        params = {'async': 'false', 'importStrategy': 'CREATE_AND_UPDATE'}

        conn = self.rest['DHIS2 Tracker']
        response = conn.post(payload, params)

        status = response.data['status']
        self.response.payload = {'is_found': True, 'status': status}

async=false makes DHIS2 process the request synchronously and return the import summary in the response. importStrategy=CREATE_AND_UPDATE creates the event when it is new and updates it when the payload carries the event UID of an existing one.

Correcting a result

A corrected result updates the event the original was written to. The lookup in find_case is repeated with program.stage_lab_result in place of program.stage_lab_request and event added to fields. The returned event UID is included in the payload under the event key, and CREATE_AND_UPDATE updates that event instead of creating a new one.

The import summary

DHIS2 returns the import summary with HTTP 200 when the event was accepted and HTTP 409 when it was rejected. The status field is OK or ERROR, stats counts what was created, updated and ignored, and validationReport.errorReports lists the reasons for a rejection:

if not response.ok:
    validation_report = response.data['validationReport']
    for report in validation_report['errorReports']:
        self.logger.warning('DHIS2 rejected the result: %s', report['message'])

The most frequent rejections are an option code that is not in the option set of the data element, a date in the future, and an enrollment the API user cannot write to.

Results over HL7 v2 receives ORU messages from a laboratory information system and invokes this service for each one.

Feature	What it does
Setup	The connections and configuration files this service reads
Calling REST APIs	Query parameters, payloads and response objects
Config files	Reading identifiers from config/user-conf