Skip to main content

querier

Provides a high-level abstraction for extracting patient info from NextGen.

Module​

Functions​

def recursively_remove_links(dictionary: Any) ‑> Any:

Remove any references to _links in the patient data from Nextgen.

Classes​

FromPatientQueryError​

class FromPatientQueryError(*args, **kwargs):

No patient was returned when constructing from query.

Ancestors​

NextGenEncountersNotImplementedError​

class NextGenEncountersNotImplementedError(*args, **kwargs):

Encounter history is not yet implemented for NextGen.

Raised rather than answered with []: an empty list is a confirmed negative to every reader — appointment_history_filter (steps/criteria_matching/functions.py) FAILs a patient whose appointment columns are both empty — so reporting one here would exclude a patient whose history was never asked for.

Ancestors​

NextGenGetPatientInfoError​

class NextGenGetPatientInfoError(*args, **kwargs):

Could not retrieve patient info.

Ancestors​

Subclasses​

NextGenPatientLister​

class NextGenPatientLister(    *,    enterprise_api: NextGenEnterpriseAPI,    patient_ids_created_after: str | date | None = None,):

Provides paginated patient ID listing from the NextGen /persons endpoint.

Arguments

  • enterprise_api: NextGenEnterpriseAPI instance.
  • patient_ids_created_after: Lower bound for patient createTimestamp. Accepts a date or a "YYYY-MM-DD" string. Defaults to DEFAULT_PATIENT_IDS_CREATED_AFTER.

Ancestors​

Variables​

  • static DEFAULT_PATIENT_IDS_CREATED_AFTER

Static methods​

from_ehr_backend​

def from_ehr_backend(    *,    nextgen_session: NextGenAuthSession | None = None,    enterprise_url: str = 'https://nativeapi.nextgen.com/nge/prod/nge-api/api',    patient_ids_created_after: str | date | None = None,    **kwargs: Any,) ‑> NextGenPatientLister:

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

Arguments

  • nextgen_session: NextGenAuthSession used to construct the Enterprise API instance.
  • enterprise_url: Optional, the Enterprise API url to use.
  • patient_ids_created_after: Optional lower bound for patient createTimestamp.
  • **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 nextgen_session is not supplied.

from_nextgen_session​

def from_nextgen_session(    nextgen_session: NextGenAuthSession,    enterprise_url: str = 'https://nativeapi.nextgen.com/nge/prod/nge-api/api',    patient_ids_created_after: str | date | None = None,) ‑> NextGenPatientLister:

Build a NextGenPatientLister from a NextGenAuthSession.

Arguments

  • nextgen_session: NextGenAuthSession for constructing the API instance.
  • enterprise_url: Optional, the Enterprise API url to use.
  • patient_ids_created_after: Optional lower bound for patient createTimestamp.

Returns NextGenPatientLister configured against the given session.

Methods​

get_patient_ids_page​

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

Return the next page of patient IDs from /persons.

Raises

  • requests.HTTPError: If the /persons request fails after the HTTP client has exhausted its retries. Surfacing this (rather than returning an empty list) ensures a transient failure mid-listing fails the task instead of being mistaken for the end of pagination.

Note page_size is accepted for interface compatibility with BaseEHRPatientLister but is ignored. NextGen omits nextPageLink from the response whenever $top is supplied, which breaks server-driven pagination across the full patient list. We instead let NextGen choose the page size and follow its nextPageLink cursor, which is the only reliable way to walk the full list under a $filter without risking duplicates or skipped rows.

NextGenPatientQuerier​

class NextGenPatientQuerier(    patient_id: str,    *,    fhir_api: NextGenFHIRAPI,    enterprise_api: NextGenEnterpriseAPI,    fhir_patient_info: RetrievedPatientDetailsJSON | 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.

Arguments

  • patient_id: The patient ID this querier corresponds to.
  • fhir_api: NextGenFHIRAPI instance.
  • enterprise_api: NextGenEnterpriseAPI instance.
  • fhir_patient_info: FHIR Patient Info with contact details.

Ancestors​

Static methods​

from_mrn​

def from_mrn(    mrn: str,    *,    fhir_api: NextGenFHIRAPI | None = None,    enterprise_api: NextGenEnterpriseAPI | None = None,    nextgen_session: NextGenAuthSession | None = None,    fhir_url: str = 'https://fhir.nextgen.com/nge/prod/fhir-api-r4/fhir/R4',    enterprise_url: str = 'https://nativeapi.nextgen.com/nge/prod/nge-api/api',) ‑> NextGenPatientQuerier:

Build a NextGenPatientQuerier from MRN.

Arguments

  • mrn: Medical Record Number
  • fhir_api: Optional, NextGenFHIRAPI instance. If not provided, nextgen_session must be.
  • enterprise_api: Optional, NextGenEnterpriseAPI instance. If not provided, nextgen_session must be.
  • nextgen_session: Optional, NextGenAuthSession instance. Only needed if fhir_api or enterprise_api are not provided.
  • fhir_url: Optional, FHIR API url. Only needed if fhir_api is not provided and a non-default URL is wanted.
  • enterprise_url: Optional, Enterprise API url. Only needed if fhir_api is not provided and a non-default URL is wanted.

Returns NextGenPatientQuerier for the target patient.

Raises

  • FromPatientQueryError: if patient ID could not be found (maybe because multiple patients match the MRN, or none do)
  • ValueError: if unable to construct the API instances because session information was not provided.

from_nextgen_session​

def from_nextgen_session(    patient_id: str,    nextgen_session: NextGenAuthSession,    fhir_url: str = 'https://fhir.nextgen.com/nge/prod/fhir-api-r4/fhir/R4',    enterprise_url: str = 'https://nativeapi.nextgen.com/nge/prod/nge-api/api',) ‑> NextGenPatientQuerier:

Build a NextGenPatientQuerier from a NextGenAuthSession.

Arguments

  • patient_id: The patient ID this querier will correspond to.
  • nextgen_session: NextGenAuthSession for constructing API instances against.
  • fhir_url: Optional, the FHIR API url to use.
  • enterprise_url: Optional, the Enterprise API url to use.

Returns NextGenPatientQuerier for the target patient.

from_patient_id​

def from_patient_id(    patient_id: str,    *,    fhir_api: NextGenFHIRAPI | None = None,    enterprise_api: NextGenEnterpriseAPI | None = None,    nextgen_session: NextGenAuthSession | None = None,    fhir_url: str = 'https://fhir.nextgen.com/nge/prod/fhir-api-r4/fhir/R4',    enterprise_url: str = 'https://nativeapi.nextgen.com/nge/prod/nge-api/api',    **kwargs: dict[str, Any],) ‑> NextGenPatientQuerier:

Build a NextGenPatientQuerier 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 FHIR API is queried by ID to retrieve contact details; if that lookup returns nothing, NoMatchingNextGenPatientError 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_api: Optional, NextGenFHIRAPI instance. If not provided, nextgen_session must be.
  • enterprise_api: Optional, NextGenEnterpriseAPI instance. If not provided, nextgen_session must be.
  • nextgen_session: Optional, NextGenAuthSession instance. Only needed if fhir_api or enterprise_api are not provided.
  • fhir_url: Optional, FHIR API url. Only needed if fhir_api is not provided and a non-default URL is wanted.
  • enterprise_url: Optional, Enterprise API url. Only needed if fhir_api is not provided and a non-default URL is wanted.
  • **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

  • ValueError: if unable to construct the API instances because session information was not provided.
  • NoMatchingNextGenPatientError: 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_api: NextGenFHIRAPI | None = None,    enterprise_api: NextGenEnterpriseAPI | None = None,    nextgen_session: NextGenAuthSession | None = None,    fhir_url: str = 'https://fhir.nextgen.com/nge/prod/fhir-api-r4/fhir/R4',    enterprise_url: str = 'https://nativeapi.nextgen.com/nge/prod/nge-api/api',    **kwargs: dict[str, Any],) ‑> NextGenPatientQuerier:

Build a NextGenPatientQuerier from patient query details.

Arguments

  • patient_dob: Patient date of birth.
  • given_name: Patient given name.
  • family_name: Patient family name.
  • fhir_api: Optional, NextGenFHIRAPI instance. If not provided, nextgen_session must be.
  • enterprise_api: Optional, NextGenEnterpriseAPI instance. If not provided, nextgen_session must be.
  • nextgen_session: Optional, NextGenAuthSession instance. Only needed if fhir_api or enterprise_api are not provided.
  • fhir_url: Optional, FHIR API url. Only needed if fhir_api is not provided and a non-default URL is wanted.
  • enterprise_url: Optional, Enterprise API url. Only needed if fhir_api is not provided and a non-default URL is wanted.
  • **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

  • NoMatchingNextGenPatientError: if patient ID could not be found (because multiple patients match the criteria, or none do)
  • ValueError: if unable to construct the API instances because session information was not provided.

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.

Returns A tuple containing:

  • List of successfully downloaded EHRDocumentInfo objects with local_path.
  • List of failed download EHRDocumentInfo objects.

get_next_appointment​

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

Inherited from:

BaseEHRQuerier.get_next_appointment :

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.

Entries whose allergyVerificationStatus is refuted are dropped.

NextGen's allergyId is a First DataBank identifier, unique only within an allergyTypeId, so it is not used as a code. rxNormCode is emitted as the coding instead, under the canonical RXNORM_SYSTEM_IDENTIFIER URI, so that one AllergyCriterion pinned to a code_system matches the same allergy whether it came from here or from a FHIR R4 server.

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

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient allergy information.

get_patient_associated_medical_practitioner​

def get_patient_associated_medical_practitioner(self) ‑> str | None:

Retrieves an associated medical practitioner for the patient.

For NextGen, this is the rendering provider for the patient's last encounter.

Returns The name of the associated medical practitioner for the patient, or None if no practitioner name is listed on the latest encounter.

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient encounter 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. A failure leaves medication_codes=None and the other two groups intact, rather than failing the whole call. The group is recorded as unsupported only where the error says so, which separates "we could not read them" from "this backend cannot".

Returns A PatientCodeDetails instance detailing the presence or absence of the provided Condition, Procedure, and Medication codes for the patient.

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.

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

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient condition information.

get_patient_devices​

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

Get implantable device information for this patient.

A device built from the FDA GUDID catalogue carries no SNOMED concept, so codings is usually empty and code_text holds the Global Medical Device Nomenclature term instead — which is what DeviceCriterion.text matches against. A device imported from a C-CDA document can carry a SNOMED concept, which is emitted as a coding under the canonical SNOMED_SYSTEM_IDENTIFIER URI.

Returns A list of Device objects relevant for the patient, sorted by code then by text for determinism.

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient device information.

get_patient_medications​

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

Get medication-related information for this patient.

We map to FHIR vocabulary wherever possible to maintain parity and simplify writing of criteria.

Each entry is coded with rxNormCode under RxNorm and ndcId under NDC. We store both against a Medication entry. A UDM-prefixed ndcId indicates a User Defined Medication (i.e. free-text), not a code.

Inactive medications are returned and carry their own status.

Deleted ones never arrive, because NextGen drops them from the endpoint.

Medication.dosage carries the dosage text (Sig) as its text, the route, the dose and the rate. Where a dosage order resolves, the dose and the rate come from it as numbers. Where none resolves, the dose comes from the product's strength and the rate is read from the dosage text.

This is not a single request. Reading a patient's medications costs one call for the list, one for the patient's ICD descriptions, and two per medication: the detail record, which carries the linked diagnoses the list omits, and the dosage orders.

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

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient medication information.

get_patient_observations​

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

Get every observation recorded for this patient.

codes is accepted for the base querier's signature and ignored. NextGen has no server-side search over lab results or vitals, and narrowing here would restate the matcher's own code semantics — globs, code systems, CodeSpec alternatives — in a second place where the two could drift. The matcher already narrows by code, and by text for the results NextGen leaves uncoded.

Arguments

  • codes: Ignored. See above.

Returns Every lab result and vitals reading for the patient, sorted by date then code for determinism.

Raises

  • NextGenGetPatientInfoError: If unable to retrieve the patient's observations.

get_patient_observations_by_category​

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

Get every observation recorded for this patient.

categories is accepted for the base querier's signature and ignored, for the reason get_patient_observations gives, and one more: NextGen records no category at all, so any mapping onto FHIR's values would be invented here. Seven of the nine the default fetch scope asks for (social-history, imaging, procedure, survey, exam, therapy, activity) could never be satisfied by a lab result or a vitals reading in any case.

Arguments

  • categories: Ignored. See above.

Returns Every lab result and vitals reading for the patient, sorted by date then code for determinism.

Raises

  • NextGenGetPatientInfoError: If unable to retrieve the patient's observations.

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 objects relevant for the patient, detailing its status and dates, sorted by procedure date.

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient procedures information.

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.

Raises

  • NextGenGetPatientInfoError: If unable to retrieve patient information.

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.

Raises

  • NextGenEncountersNotImplementedError: Always.

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({'appointments', 'conditions', 'chart', 'procedures', 'patientInfo', 'encounters', 'medications'}),) ‑> bitfount.externals.ehr.nextgen.querier._NextGenPatientJSONDump:

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":

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 (also written to save_path), so callers can gate on whether it holds any meaningful content.