Skip to main content

base_querier

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.

Classes​

BaseEHRPatientLister​

class BaseEHRPatientLister():

Base class for listing patient IDs from an EHR.

Ancestors​

Subclasses​

Static methods​

from_ehr_backend​

def from_ehr_backend(    **kwargs: Any,) ‑> BaseEHRPatientLister:

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

Concrete implementations cherry-pick the kwargs they need (e.g. nextgen_session / enterprise_url for NextGen, fhir_client for FHIR R4) and ignore the rest, so callers (e.g. EHRDataResource) can pass a single set of kwargs without filtering by backend.

Methods​

get_patient_ids_page​

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

Get the next page of patient IDs from the EHR.

Implementations should return patient IDs for the current page on the first call, then advance internal pagination state so subsequent calls return the next page. Once exhausted, they should return an empty list.

Transient HTTP/API failures are retried at the HTTP client layer; if they still fail, implementations should raise rather than return an empty list, so callers do not mistake a failure for end of pagination.

BaseEHRQuerier​

class BaseEHRQuerier(patient_id: str):

Base class for patient-scoped EHR queriers.

Arguments

  • patient_id: The patient ID this querier corresponds to.

Ancestors​

Subclasses​

Static methods​

from_patient_id​

def from_patient_id(    patient_id: str, **kwargs: Any,) ‑> BaseEHRQuerier:

Build a querier directly from a known patient ID.

from_patient_query​

def from_patient_query(    patient_dob: str | date,    given_name: str | None = None,    family_name: str | None = None,    **kwargs: Any,) ‑> BaseEHRQuerier:

Look up a single patient by demographic search.

Methods​

download_all_documents​

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

Download PDF documents for the current patient.

Arguments

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

get_next_appointment​

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

Get the next appointment date for the patient.

Returns The next appointment date for the patient from today, or None if they have no future appointment.

Raises

  • NextGenGetPatientInfoError: 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 codes parameter): an allergy list is small enough per patient to fetch whole and match against AllergyCriterion.code/code_system afterwards, the same as conditions/procedures/medications.

Returns A list of Allergy objects relevant for the patient.

Raises

  • GetPatientInfoError: If unable to retrieve allergy information — including when the backend does not support allergy search at all (e.g. NextGen, not yet implemented). That case must raise rather than return []: this result feeds allergy_codes_json, which eligibility criteria read, and "backend can't search this" is "could not determine" (UNKNOWN), not "confirmed no allergies".

get_patient_associated_medical_practitioner​

def get_patient_associated_medical_practitioner(self) ‑> str | None:

Retrieves an associated medical practitioner for the patient.

The exact source depends on the EHR implementation. For example, this may be derived from the patient's last encounter or from a practitioner reference stored on the patient record.

Returns The name of an associated medical practitioner for the patient, or None if no practitioner can be resolved.

Raises

  • GetPatientInfoError: If unable to retrieve patient information.

get_patient_code_states​

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

Get Condition, Procedure, and (optionally) Medication code information.

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. Defaults to False so 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 Condition and Procedure codes for the patient. medication_codes is None when include_medications is False.

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.

get_patient_devices​

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

Get device information for this patient.

Fetched unfiltered (no codes parameter): a device list is small enough per patient to fetch whole and match against DeviceCriterion.code/code_system afterwards, the same as conditions/procedures/medications/allergies.

Returns A list of Device objects relevant for the patient.

Raises

  • GetPatientInfoError: If unable to retrieve device information — including when the backend does not support device search at all (e.g. NextGen, not yet implemented). That case must raise rather than return []: this result feeds device_codes_json, which eligibility criteria read, and "backend can't search this" is "could not determine" (UNKNOWN), not "confirmed no devices".

get_patient_medications​

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

Get medication-related information for this patient.

Implementations that query more than one medication-related resource (e.g. FHIR R4's MedicationRequest/Statement/Dispense/Administration) should return whatever a server supports rather than failing outright when only some are available — see the FHIR R4 implementation for the "where available" contract this establishes.

Returns A list of Medication objects relevant for the patient.

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.

Unlike get_patient_conditions/get_patient_procedures (which fetch every code unfiltered), Observations must be searched by an explicit code list. At least one major FHIR backend enforces this server-side: Epic's Observation.Search rejects a query with neither code nor category (error 59108, "either the category or code parameter must be specified" — https://fhir.epic.com/Specifications?api=10297). This method supplies code; see get_patient_observations_by_category for the category alternative (a coarser, less precise fetch — Epic's own docs note category assignments are "subjective and may differ across organizations" and recommend code search when consistency matters).

Arguments

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

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

Raises

  • GetPatientInfoError: If unable to retrieve Observation information — including when the backend does not support Observation search at all (e.g. NextGen). That case must raise rather than return []: this result feeds observation_codes_json, which eligibility criteria read, and "backend can't search this" is "could not determine" (UNKNOWN), not "confirmed no observations".

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.

Searches the core FHIR observation-category system only (values like "laboratory", "vital-signs", "social-history" — see http://hl7.org/fhir/R4/valueset-observation-category.html) — not vendor-specific category mechanisms (e.g. Epic also lets category take a SNOMED code repurposed per-organization, or an org-specific Order Type OID/abbreviation pair; neither is supported here, since both require configuration this generic querier has no way to know). Category is a coarser, less precise net than code (get_patient_observations) — it returns every Observation in the category, not just ones matching a specific measurement, and Epic's own docs warn category assignment is "subjective and may differ across organizations."

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

  • GetPatientInfoError: If unable to retrieve Observation information — including when the backend does not support Observation search at all (e.g. NextGen); see get_patient_observations for why that must raise rather than return [].

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.

Returns A list of procedure codes relevant for the patient.

get_previous_appointment_details​

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

Get the details of previous appointments for the patient.

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

get_previous_encounter_details​

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

Get the details of previous encounters for the patient.

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

get_visual_acuity​

def get_visual_acuity(self) ‑> Observation | None:

Get Visual Acuity observation for a patient.

BCVA = Best Corrected Visual Acuity.

produce_json_dump​

def produce_json_dump(    self,    save_path: Path,    elements_to_dump: Container[str] = frozenset({'procedures', 'medications', 'conditions', 'chart', 'appointments', 'patientInfo', 'encounters'}),) ‑> collections.abc.Mapping[str, typing.Any] | None:

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 options that are recognised depends on the querier.

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, or None if the querier does not return one.