Skip to main content

protocols

Structural protocols for objects passed into step tasks.

These Protocols describe the minimal interface that the DAG runner must satisfy when constructing and passing resource-like objects to Prefect tasks. Concrete implementations do not need to inherit from these — structural (duck-type) compatibility is sufficient.

Sub-modules​

model ModelResourceProtocol — any object that can load a Bitfount model. ehr EHRDataResource — concrete resource that queries an EHR provider (NextGen or FHIR R4) and returns EHRPatientResource records. EHRFetchOptions — which extra per-patient data to fetch alongside it. NO_EHR_RESOURCE_IDENTITY — the ehr hashing identity for a pod with no EHR configured.

Module​

Submodules​

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​

check_availability​

def check_availability(self) ‑> EHRPreflightResult:

Establish whether this EHR can be reached, before any patient is queried.

Both backends are probed the same way -- obtain a token, then ask the FHIR server for its CapabilityStatement -- because both speak FHIR and differ only in where the token comes from. The retry policy lives in externals.ehr.preflight; this method supplies the one attempt it repeats.

The probe is a statement about the run, not about any patient: a patient the EHR has no record of is an answer, and answering is what this checks for.

Returns The verdict. available is False only when every attempt failed, so a single blip does not cost a run its EHR data.

get_patient_info_by_id​

def get_patient_info_by_id(    self,    patient_id: str,    *,    fetch_options: EHRFetchOptions = EHRFetchOptions(fetch_appointments=False, fetch_practitioner=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, fetch_practitioner=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,    fetch_practitioner: 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 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.
  • fetch_practitioner gates one extra round-trip for the patient's associated medical practitioner, which only display-oriented flows 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
  • static fetch_practitioner : bool

ModelResourceProtocol​

class ModelResourceProtocol(*args, **kwargs):

An object that can load a Bitfount model ready for inference.

Ancestors​

Methods​

load_model​

def load_model(    self,    model_ref: Any,    *,    datastructure: DataStructure,    schema: BitfountSchema,    batch_size: int | None,) ‑> Any:

Load and return a model instance ready for inference.