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​

NextGenAllergiesNotImplementedError​

class NextGenAllergiesNotImplementedError(*args, **kwargs):

Allergy/intolerance extraction is not yet implemented for NextGen.

A "not yet" — same framing as NextGenMedicationsNotImplementedError, not NextGenObservationsNotImplementedError's "never will" — since AllergyCriterion is currently scoped to FHIR only, not because NextGen's API has no AllergyIntolerance equivalent to search.

Ancestors​

NextGenDevicesNotImplementedError​

class NextGenDevicesNotImplementedError(*args, **kwargs):

Device extraction is not yet implemented for NextGen.

A "not yet" — same framing as NextGenAllergiesNotImplementedError — since DeviceCriterion is currently scoped to FHIR only, not because NextGen's API has no Device equivalent to search.

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​

NextGenMedicationsNotImplementedError​

class NextGenMedicationsNotImplementedError(*args, **kwargs):

Medication extraction is not yet implemented for NextGen.

A distinct subclass of NextGenGetPatientInfoError so a caller (or future monitoring/retry logic) can tell "this will never succeed until implemented" apart from a genuine transient fetch failure — even though both currently degrade the same way in get_patient_code_states (medication_codes=None).

Ancestors​

NextGenObservationsNotImplementedError​

class NextGenObservationsNotImplementedError(*args, **kwargs):

NextGen has no equivalent of FHIR's Observation search.

Unlike NextGenMedicationsNotImplementedError this is not a "not yet": the Enterprise API exposes nothing to search, so no implementation is pending. A distinct subclass all the same, so "this will never succeed" is legible in the exception type rather than only in a message string.

Ancestors​

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.

Not yet implemented: AllergyCriterion is currently scoped to FHIR only. Implement this the same way get_patient_conditions/ get_patient_procedures do, against whatever NextGen Enterprise API endpoint/JSON shape carries allergy/intolerance data, before removing this raise.

Raises

  • NextGenAllergiesNotImplementedError: Always, until the above is done (a NextGenGetPatientInfoError subclass).

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 attempts to fetch medications. get_patient_medications is not yet implemented for NextGen, so this always yields medication_codes=None today; the flag still exists so callers write the same call shape as the FHIR R4 backend.

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 device information for this patient.

Not yet implemented: DeviceCriterion is currently scoped to FHIR only. Implement this the same way get_patient_conditions/ get_patient_procedures do, against whatever NextGen Enterprise API endpoint/JSON shape carries device data, before removing this raise.

Raises

  • NextGenDevicesNotImplementedError: Always, until the above is done (a NextGenGetPatientInfoError subclass).

get_patient_medications​

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

Get medication-related information for this patient.

Not yet implemented: NextGenEnterpriseMedicationJSON (the entry shape enterprise_api.get_medications returns) is currently an empty stub — "only containing the elements we care about," and today that is nothing, because no example response of this type has ever been captured (see medications_response in test_api.py: "DEV: We have no example JSONs of this type of response"). Extracting real Medication fields here would mean guessing field names against an unconfirmed shape, which risks silently corrupting the cache rather than surfacing a clear gap. Confirm the real shape against a live response or NextGen's API docs, fill in NextGenEnterpriseMedicationJSON, and implement extraction the same way get_patient_conditions/ get_patient_procedures do, before removing this raise.

Raises

  • NextGenMedicationsNotImplementedError: Always, until the above is done (a NextGenGetPatientInfoError subclass).

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.

Raises rather than returning []: this feeds observation_codes_json, which eligibility criteria read, and NextGen not supporting Observation search at all is "could not determine" (caught by _fetch_or_none in ehr.py and reported as None/UNKNOWN), not "confirmed no observations" — returning [] here would make that structural limitation look identical to a successful-but-empty search, and eligibility criteria could then FAIL a patient instead of reporting UNKNOWN.

Raises

  • NextGenGetPatientInfoError: Always — NextGen has no equivalent of FHIR's Observation search.

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.

Raises rather than returning [] — see get_patient_observations.

Raises

  • NextGenGetPatientInfoError: Always — NextGen has no equivalent of FHIR's Observation search.

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({'procedures', 'medications', 'conditions', 'chart', 'appointments', 'patientInfo', 'encounters'}),) ‑> 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.