Skip to main content

ehr

EHR data resource protocol for step tasks.

Module​

Functions​

ehr_resource_identity​

def ehr_resource_identity(ehr_config: EHRConfig) ‑> str:

Return a stable identity string for the EHR ehr_config points at.

The EHR steps (ehr_patient_lister, ehr_query) carry no config field naming the system they read, so their task_hash would otherwise be blind to it — repointing a pod at a different EHR would serve the previous EHR's cached rows. This string is folded into those steps' hashes; see bitfount.flows.dag.hashing.

Four parts, because no one of them is sufficient:

  • provider selects the querier class in EHRDataResource.__init__, and so the shape of what lands in the cache.
  • fhir_url is the endpoint the FHIR R4 family actually queries — for those providers enterprise_url is never set.
  • enterprise_url is the NextGen Enterprise endpoint.
  • list_resource_ids selects the population: set, patient enumeration is scoped to those FHIR List resources' combined membership; unset, it is an unfiltered /Patient search. ehr_patient_lister has no config at all, so without this a narrowed or widened cohort would reuse the previous cohort's ehr_patient_ids rows. Sorted, since the entries are a set of Lists whose declaration order does not change the population.

The two URLs are resolved exactly as EHRDataResource.__init__ resolves them, defaults included, so a config that names an endpoint explicitly and one that falls back to the default identify the same EHR.

Deliberately excluded: smart_on_fhir_url and smart_on_fhir_resource_server_url. Both are auth endpoints — they decide how the querier obtains a token, not which records come back — so folding them in would re-partition the cache on a credentials move that changes no data.

Arguments

  • ehr_config: The pod's EHR configuration.

Returns A canonical provider|fhir_url|enterprise_url|list_resource_ids string.

Classes​

EHRDataResource​

class EHRDataResource(    hub: BitfountHub,    ehr_config: EHRConfig | None = None,    ehr_secrets: RefreshableJWT | None = None,):

Resource to allow connection to the EHR.

This is init with EHR configuration from the pod_config. Tasks can obtain a patient-specific querier either via Patient ID, or by providing first name + last name + DOB.

Construct the resource for a single EHR provider.

Picks one of two querier backends based on ehr_config: NextGenEHRConfig selects the NextGen path (which authenticates against the Hub session), any other config whose provider ends in "r4" selects the FHIR R4 path. Anything else raises ValueError.

Arguments

  • hub: Bitfount Hub instance, used by the NextGen path to obtain a SMART-on-FHIR session.
  • ehr_config: Provider configuration loaded from the pod config.
  • ehr_secrets: Externally-supplied JWT secrets for the FHIR R4 path. Not needed for SMARTBackendEHRConfig, which carries its own auth on the config object. May be None only when the config is SMART Backend, or when allow_no_ehr_secrets is set for a genuinely unauthenticated server.

Raises

  • ValueError: If ehr_config is None, if base_url is missing on a FHIR R4 config, if the provider value is unrecognised, or if a FHIR R4 config has no usable credentials and allow_no_ehr_secrets is not set.

Methods​

get_patient_info_by_id​

def get_patient_info_by_id(    self,    patient_id: str,    *,    fetch_options: EHRFetchOptions = EHRFetchOptions(fetch_appointments=False, observation_codes=None, observation_categories=None),) ‑> EHRPatientResource:

Returns EHR querier for this patient.

The querier class is selected by self.querier_type. Within each class, the construction method is chosen by the patient details variant: EHRIDPatientDetails builds via from_patient_id (no demographic search), and NameDOBPatientDetails builds via from_patient_query.

Arguments

  • patient_id: The EHR system's patient ID.
  • fetch_options: Forwarded to _get_patient_resource; see EHRFetchOptions.

get_patient_info_by_name_dob​

def get_patient_info_by_name_dob(    self,    given_name: str | None,    family_name: str | None,    patient_dob: str | date,    *,    fetch_options: EHRFetchOptions = EHRFetchOptions(fetch_appointments=False, observation_codes=None, observation_categories=None),) ‑> EHRPatientResource:

Look up a single patient by demographic search.

Routes the call to the active querier backend (NextGenPatientQuerier or FHIRR4PatientQuerier) and normalises the response into an EHRPatientResource.

Arguments

  • given_name: Patient's first/given name.
  • family_name: Patient's last/family name.
  • patient_dob: Date of birth, either as a date or an ISO-format string the backend can parse.
  • fetch_options: Forwarded to _get_patient_resource; see EHRFetchOptions.

Returns The materialised EHRPatientResource. Code lists may be empty (or None on a GetPatientInfoError from the backend).

Raises

  • NoMatchingPatientError: Propagated from the backend when no patient matches the supplied demographics.
  • ValueError: If self.querier_type is set to an unsupported value.

get_patient_lister​

def get_patient_lister(self) ‑> BaseEHRPatientLister:

Build a paginated patient-ID lister for the active EHR backend.

Routes the call to the active lister backend (NextGenPatientLister or FHIRR4PatientLister) via the unified from_ehr_backend entrypoint — each implementation cherry-picks the kwargs it needs and ignores the rest, mirroring the dispatch shape of get_patient_info_by_name_dob.

EHRFetchOptions​

class EHRFetchOptions(    fetch_appointments: bool = False,    observation_codes: Sequence[tuple[str | None, str]] | None = None,    observation_categories: Sequence[str] | None = None,):

Which extra per-patient EHR data to fetch, beyond the unfiltered code lists.

Condition/Procedure codes are always fetched unfiltered and matched against criteria locally afterwards — cheap, and every backend supports an unfiltered fetch. These three don't work that way, each for its own reason, which is why they must be requested up front rather than filtered afterwards:

  • fetch_appointments gates an extra round-trip (previous appointments/encounters, next appointment) most flows don't need.
  • observation_codes/observation_categories: Observations must be searched by an explicit code or category — at least one major FHIR backend (Epic) rejects an Observation search with neither.

Variables​

  • static fetch_appointments : bool

EHRPatientResource​

class EHRPatientResource(**data: Any):

In-memory representation of a single patient pulled from the EHR.

Built by EHRDataResource._get_patient_resource from the raw querier response, and consumed by _build_ehr_record when assembling cache rows. patient_id is the EHR system's ID — it is not the same as the imaging-side Patient ID column. condition_codes_json / procedure_codes_json / observation_codes_json / medication_codes_json / allergy_codes_json default to None so that "lookup failed" (None) stays distinguishable from "no codes for this patient" ([]). observation_codes_json is only populated when observation_codes and/or observation_categories was passed in — unlike conditions/procedures/medications/allergies, Observations are fetched by explicit code or category, not unfiltered; when both are passed, the two fetches' results are merged and deduplicated. medication_codes_json and allergy_codes_json are both fetched unfiltered, the same as conditions/procedures. Every one of the five preserves None distinctly from [] at this layer; see _get_patient_resource.

fetch_outcomes carries which FetchGroups that lookup reached and, for those it did not, whether the endpoint is unsupported by the server or was merely unreachable this time. FetchGroup.DEMOGRAPHICS never appears there — it has no group-scope reason, so "were demographics fetched" is answered by the lookup having succeeded at all. It is in-memory only: the cache row builders copy the fields above across one at a time and do not copy it, so a group outcome never lands in ehr_data. device_codes_json is fetched unfiltered, the same as allergies, but has no date field of its own (see DeviceCriterion).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static allergy_codes_json : list[typing.Any] | None
  • static cell_numbers : list[str]
  • static condition_codes_json : list[dict[str, str | None | datetime.datetime | dict[str, str] | list[dict[str, typing.Any]]]] | None
  • static country : str | None
  • static device_codes_json : list[typing.Any] | None
  • static emails : list[str]
  • static family_name : str | None
  • static gender : str | None
  • static given_name : str | None
  • static home_numbers : list[str]
  • static mailing_address : str | None
  • static medical_record_number : list[str] | None
  • static medication_codes_json : list[typing.Any] | None
  • static model_config
  • static patient_id : str
  • static postal_code : str | None
  • static previous_appointments : list[dict[str, str]] | None
  • static previous_encounters : list[dict[str, str]] | None
  • static procedure_codes_json : list[dict[str, str | None | datetime.datetime | dict[str, str] | list[dict[str, typing.Any]]]] | None

QuerierType​

class QuerierType(*args, **kwds):

EHR Querier for use in algorithm.

Ancestors​

Variables​

  • static FHIR_R4
  • static NEXTGEN