Skip to main content

querier

Provides a high-level abstraction for extracting patient info from FHIR R4 APIs.

Classes​

FHIRR4GetPatientInfoError​

class FHIRR4GetPatientInfoError(*args, **kwargs):

Could not retrieve patient info.

Ancestors​

Subclasses​

FHIRR4MedicationsUnsupportedError​

class FHIRR4MedicationsUnsupportedError(*args, **kwargs):

No medication-related resource type is supported by this server.

A distinct subclass of FHIRR4GetPatientInfoError so a caller (or future monitoring/retry logic) can tell "this server structurally does not support any medication resource" apart from a genuine transient fetch failure — even though both currently degrade the same way in get_patient_code_states (medication_codes=None).

Ancestors​

FHIRR4PatientLister​

class FHIRR4PatientLister(    *,    fhir_client: FHIRClient,    list_resource_ids: list[str] | None = None,    ehr_provider: EHRProvider | None = None,):

Provides paginated patient ID listing from a FHIR R4 /Patient endpoint.

Also supports listing patients via one or more pre-configured FHIR List resources (see list_resource_ids), for EHRs (e.g. Epic) that reject an unfiltered /Patient search.

Ancestors​

Static methods​

from_ehr_backend​

def from_ehr_backend(    *,    fhir_client: FHIRClient | None = None,    list_resource_ids: list[str] | None = None,    ehr_provider: EHRProvider | None = None,    **kwargs: Any,) ‑> FHIRR4PatientLister:

Build a FHIRR4PatientLister from a unified set of EHR backend kwargs.

Arguments

  • fhir_client: FHIRClient instance wrapping the /Patient endpoint.
  • list_resource_ids: Optional FHIR List resource identifiers (system|value tokens). When set, patient enumeration is scoped to these pre-configured Lists instead of an unfiltered /Patient search.
  • ehr_provider: The active EHR provider, to handle quirks of different FHIR implementations.
  • **kwargs: Ignored. Accepted so callers (e.g. EHRDataResource) can pass a unified set of kwargs across NextGen and FHIR R4 backends without filtering by backend.

Raises

  • ValueError: If fhir_client is not supplied.

Methods​

get_patient_ids_page​

def get_patient_ids_page(self, page_size: int = 25) ‑> list[str]:

Return the next page of patient IDs.

If list_resource_ids was supplied at construction, patient IDs are served by resolving those FHIR List resources (see _get_patient_ids_page_from_lists). Otherwise, falls back to the unfiltered /Patient search-bundle pagination below, unchanged.

Raises

  • FHIRR4HTTPError: If the request fails with an HTTP error after the FHIR client has exhausted its retries.
  • FHIRR4APIError: If the request fails with a generic API error.
  • FHIRR4OperationOutcomeError: If the server returns an OperationOutcome.
  • FHIRR4AuthenticationError: If authentication fails.

Surfacing these (rather than returning an empty list) ensures a transient failure mid-listing fails the task instead of being mistaken for the end of pagination.

FHIRR4PatientQuerier​

class FHIRR4PatientQuerier(    patient_id: str,    *,    fhir_client: FHIRClient,    fhir_patient_info: RetrievedPatientDetailsJSON | None = None,    ehr_provider: str | None = None,    patient_dict: dict[str, Any] | None = None,):

Provides query/data extraction methods for a given patient.

This class is a higher-level abstraction than the direct API interactions, providing methods for extracting/munging data from the API responses.

NOTE: This querier is used by all FHIR R4 providers. The "generic r4" provider serves as the baseline FHIR R4 implementation that adheres strictly to the FHIR standard without any provider-specific customizations. If provider-specific elements (e.g., Epic-specific) are added to this class or its methods, ensure that "generic r4" continues to work as the baseline standard FHIR R4 provider.

Arguments

  • patient_id: The patient ID this querier corresponds to.
  • fhir_client: FHIRClient instance.
  • fhir_patient_info: FHIR Patient Info with contact details.
  • ehr_provider: The EHR provider e.g. "nextech", "epic", "generic r4"
  • patient_dict: Optional patient details. If we've created this class with from_patient_query, we'll store the obtained patient_dict for later reference.

Ancestors​

Static methods​

from_mrn​

def from_mrn(    mrn: str, *, fhir_client: FHIRClient, ehr_provider: EHRProvider | None = None,) ‑> FHIRR4PatientQuerier:

Build a FHIRR4PatientQuerier from MRN.

Arguments

  • mrn: Medical Record Number
  • fhir_client: FHIRClient instance
  • ehr_provider: The EHR provider

Returns FHIRR4PatientQuerier for the target patient.

Raises

  • NoMatchingFHIRR4PatientError: No patient matching the MRN could be found
  • NonSpecificFHIRR4PatientError: Multiple patients match the MRN
  • NoFHIRR4PatientIDError: Patient matching the MRN was found, but no patient ID was associated

from_patient_id​

def from_patient_id(    patient_id: str,    *,    fhir_client: FHIRClient | None = None,    ehr_provider: EHRProvider | None = None,    **kwargs: dict[str, Any],) ‑> FHIRR4PatientQuerier:

Build a FHIRR4PatientQuerier directly from a known patient ID.

Unlike from_patient_query/from_mrn, this does not search for the patient by demographics; the patient ID is already known. The /Patient endpoint is queried by ID to retrieve contact details; if that lookup returns nothing, NoMatchingFHIRR4PatientError is raised so the caller can skip the patient rather than issuing further queries with no confirmed patient context.

Arguments

  • patient_id: The known patient ID.
  • fhir_client: FHIRClient instance.
  • ehr_provider: The EHR provider.
  • **kwargs: Ignored. Accepted so callers (e.g. EHRDataResource) can pass a unified set of kwargs across NextGen and FHIR R4 backends without filtering by querier type.

Returns FHIRR4PatientQuerier for the target patient.

Raises

  • NoMatchingFHIRR4PatientError: No patient record could be retrieved for the given ID.

from_patient_query​

def from_patient_query(    patient_dob: str | date,    given_name: str | None = None,    family_name: str | None = None,    *,    fhir_client: FHIRClient | None = None,    ehr_provider: EHRProvider | None = None,    **kwargs: dict[str, Any],) ‑> FHIRR4PatientQuerier:

Build a FHIRR4PatientQuerier from patient query details.

Arguments

  • patient_dob: Patient date of birth.
  • given_name: Patient given name.
  • family_name: Patient family name.
  • fhir_client: FHIRClient instance
  • ehr_provider: The EHR provider
  • **kwargs: Ignored. Accepted so callers (e.g. EHRDataResource) can pass a unified set of kwargs across NextGen and FHIR R4 backends without filtering by querier type.

Returns NextGenPatientQuerier for the target patient.

Raises

  • NoMatchingFHIRR4PatientError: No patients matching the name/dob criteria could be found
  • NonSpecificFHIRR4PatientError: Multiple patients match the criteria, could not determine the correct one
  • NoFHIRR4PatientIDError: Patient matching the criteria was found, but no patient ID was associated

get_patient_response_by_mrn​

def get_patient_response_by_mrn(    fhir_client: FHIRClient, mrn: str,) ‑> dict[str, typing.Any] | None:

Get JSON response from /Patient endpoint given an MRN.

Arguments

  • fhir_client: FHIRClient instance
  • mrn: Medical Record Number

Returns Patient resource dict if found, None otherwise

get_patient_response_by_name​

def get_patient_response_by_name(    fhir_client: FHIRClient,    patient_dob: str | date,    given_name: str | None = None,    family_name: str | None = None,) ‑> dict[str, typing.Any] | None:

Get JSON response from /Patient endpoint given name and DOB.

Methods​

download_all_documents​

def download_all_documents(    self,    save_path: Path,) ‑> tuple[list[DownloadedEHRDocumentInfo], list[FailedEHRDocumentInfo]]:

Inherited from:

BaseEHRQuerier.download_all_documents :

Download PDF documents for the current patient.

Arguments

  • save_path: Documents path for the PDF documents to be saved.

download_documents_batch​

def download_documents_batch(    self,    save_path: Path,    batch_document_infos: list[EHRDocumentInfo],) ‑> tuple[list[DownloadedEHRDocumentInfo], list[FailedEHRDocumentInfo]]:

Downloads a batch of documents.

Returns A tuple containing:

  • list of info of documents that were successfully downloaded.
  • list of info of documents that could not be downloaded.

get_document_infos​

def get_document_infos(    self,) ‑> collections.abc.Iterable[list[EHRDocumentInfo]]:

Yields document items related to this patient.

get_next_appointment​

def get_next_appointment(self) ‑> datetime.date | None:

Get the next appointment date for the patient.

Falls back to encounters if appointments are not available or empty.

Returns The next appointment date for the patient from today, or None if they have no future appointment. Any cancelled or errored appointments are ignored.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient information.

get_patient_allergies​

def get_patient_allergies(self) ‑> list[Allergy]:

Get allergy/intolerance information for this patient.

Fetched unfiltered (no code list to search by): allergy lists are small enough per patient to fetch whole and match against a criteria_tree leaf's own code/system afterwards. code (AllergyIntolerance's own top-level CodeableConcept, 0..1) and each reaction[].substance (also a CodeableConcept, 0..*) can each carry more than one coding. One Allergy is returned per FHIR resource, with codings holding every coding on its own code and each AllergyReaction.substance_codings holding every coding on that reaction's substance — not one entry per coding, so the same clinical fact is never stored more than once. A code with no coding at all still leaves code_system/code_code/ code_display and codings empty — AllergyIntolerance.code is 0..1, so a resource may carry no coded diagnosis and rely entirely on reaction[].substance instead.

Returns A list of Allergy objects relevant for the patient, sorted by date.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient allergy information.

get_patient_associated_medical_practitioner​

def get_patient_associated_medical_practitioner(self) ‑> str | None:

Retrieve an associated medical practitioner for the patient.

This prefers a practitioner referenced on the Patient resource via generalPractitioner. If none is available there, it falls back to the practitioner on the latest appointment or encounter.

Returns The practitioner's name, or None if no associated practitioner can be resolved.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient encounter information.

get_patient_code_states​

def get_patient_code_states(    self, *, include_medications: bool = False,) ‑> PatientCodeDetails:

Get information of Conditions, Procedures, and (optionally) Medications.

Sugar method that combines get_patient_conditions(), get_patient_procedures(), and — when requested — get_patient_medications(), returning a pre-constructed PatientCodeDetails container.

Arguments

  • include_medications: When True, also fetches medications (4 extra HTTP round-trips against MedicationRequest/Statement/ Dispense/Administration). Defaults to False so the many existing callers that only read condition_codes/ procedure_codes do not pay for a fetch whose result they never consume.

Returns A PatientCodeDetails instance detailing the presence or absence of the provided Conditions, Procedures, and Medications for the patient. medication_codes is None when include_medications is False, the same as a fetch that was never attempted.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve condition information.

get_patient_conditions​

def get_patient_conditions(    self,    statuses_filter: list[ClinicalStatus] | None = None,    code_types_filter: list[CodeSystems] | None = None,) ‑> list[Condition]:

Get conditions related to this patient.

Arguments

  • statuses_filter: If provided, returns only conditions of that status e.g. ['active','recurrence']
  • code_types_filter: If provided, returns only conditions with codes of a specific code system (ICD10, Snomed) e.g. ["icd10"]

Returns A list of Condition objects relevant for the patient, detailing its status and dates, sorted by condition onset date.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient condition information.

get_patient_devices​

def get_patient_devices(self) ‑> list[Device]:

Get device information for this patient.

Fetched unfiltered (no code list to search by): device lists are small enough per patient to fetch whole and match against a criteria_tree leaf's own code/system afterwards. type (Device's own top-level CodeableConcept, 0..1) can itself carry more than one coding; codings holds every one of them. One Device is returned per FHIR resource. A type with no coding at all (or absent entirely) still leaves code_system/code_code/ code_display and codings empty.

Returns A list of Device objects relevant for the patient, sorted by code for deterministic output (FHIR R4 Device has no field meaning "when this became relevant to the patient" to sort by).

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient device information.

get_patient_medications​

def get_patient_medications(self) ‑> list[Medication]:

Get medication-related information for this patient.

This method queries MedicationRequest, MedicationStatement, MedicationDispense, and MedicationAdministration in sequence. This matches every other resource type this querier fetches. A server may support only some of these four resource types. If so, that server still gives us whatever it has. The whole call does not fail just because one of the four resource types is absent.

Returns A list of Medication objects for the patient, sorted by date.

Raises

  • FHIRR4MedicationsUnsupportedError: Every resource type came back UNSUPPORTED. None processed, and none errored. This exception is a FHIRR4GetPatientInfoError subclass. It is the only case that means "this server structurally does not support medications".
  • FHIRR4GetPatientInfoError: At least one resource type genuinely failed (a network error or a deserialisation error), and none processed successfully. This result is too uncertain to report as the more specific "unsupported" signal above.

get_patient_observations​

def get_patient_observations(    self, codes: Sequence[tuple[str | None, str]],) ‑> list[Observation]:

Get Observation resources matching any of the given codes.

Arguments

  • codes: (code_system, code) pairs to search for. code_system is a short name from CODE_SYSTEM_TO_IDENTIFIER (e.g. "loinc", "snomed"); None matches the code value across any system.

Returns Observations for the patient matching any of the given codes, should be sorted by date (newest first).

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve Observation information for the patient.

get_patient_observations_by_category​

def get_patient_observations_by_category(    self, categories: Sequence[str],) ‑> list[Observation]:

Get Observation resources matching any of the given categories.

Arguments

  • categories: Core FHIR observation-category values to search for (e.g. ["laboratory", "vital-signs"]).

Returns Observations for the patient matching any of the given categories, sorted by date (newest first).

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve Observation information for the patient.

get_patient_procedures​

def get_patient_procedures(    self,    statuses_filter: list[ProcedureStatus] | None = None,    code_types_filter: list[CodeSystems] | None = None,) ‑> list[Procedure]:

Get information of procedure codes this patient has.

Arguments

  • statuses_filter: If provided, returns only procedures of that status e.g. ['completed', 'in-progress']
  • code_types_filter: If provided, returns only conditions with codes of a specific code system (CPT4, Snomed) e.g. ["cpt4", "snomed"]

Returns A list of Procedure objects relevant for the patient, detailing its status and dates, sorted by procedure date.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient procedures information.

get_patient_response_by_id​

def get_patient_response_by_id(self) ‑> dict[str, typing.Any] | None:

Get JSON response from /Patient endpoint given a Patient ID.

get_previous_appointment_details​

def get_previous_appointment_details(    self, include_maybe_attended: bool = True,) ‑> list[EHRAppointmentEncounter]:

Get list of previous appointments for the patient.

Returns A list of EHRAppointmentEncounter for the patient, sorted chronologically (oldest first), or an empty list if they have no previous appointments.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient information.

get_previous_encounter_details​

def get_previous_encounter_details(    self, include_maybe_attended: bool = True,) ‑> list[EHRAppointmentEncounter]:

Get list of previous encounters for the patient.

Returns A list of EHRAppointmentEncounter for the patient, sorted chronologically (oldest first), or an empty list if they have no previous appointments.

Raises

  • FHIRR4GetPatientInfoError: If unable to retrieve patient information.

get_visual_acuity​

def get_visual_acuity(self) ‑> Observation | None:

Get Visual Acuity observation for a patient.

produce_json_dump​

def produce_json_dump(    self,    save_path: Path,    elements_to_dump: Container[str] = frozenset({'procedures', 'medications', 'conditions', 'chart', 'appointments', 'patientInfo', 'encounters'}),) ‑> bitfount.externals.ehr.fhir_r4.querier._FHIRR4PatientJSONDump:

Produce a JSON dump of patient information for the target patient.

Saves the JSON dump out to file and the contents can be controlled by elements_to_dump.

The following options are recognised:

  • "patientInfo": /Patient
  • "appointments": /Appointment (not always available)
  • "conditions": /Conditions
  • "encounters": /Encounters
  • "medications": /MedicationRequest
  • "procedures": /Procedures

Arguments

  • save_path: The file location to save the JSON dump to.
  • elements_to_dump: Collection of elements to include in the dump. See above for what options can be included.

Returns The assembled JSON dump. Elements that could not be retrieved (or that the server does not support) are omitted, so callers can inspect the result to decide whether the dump contains anything meaningful.