Skip to content

Pipelines

Pipeline/Step is the public builder API; both it and Pipeline.from_yaml converge on the same PipelineConfig before the streaming executor ever sees them. See Core → Configuration for PipelineConfig and Core → Provenance for the RunManifest a run returns.

from openbtk.pipelines import Pipeline, Step

pipeline = (
    Pipeline("clinical-rag")
    .add(Step("load", "loader.clinical_text.plain_text", path="./notes"))
    .add(Step("deid", "preprocessor.general.deidentify", mode="surrogate"))
    .add(Step("chunk", "chunker.clinical_text.section_aware", max_tokens=512))
)
manifest = pipeline.run()

openbtk.pipelines.pipeline.Pipeline 

Pipeline(name: str, *, version: int = 1, description: str | None = None, policy: PolicyConfig | None = None, provenance: ProvenanceConfig | None = None)

Build and run a streaming pipeline.

Example

pipeline = Pipeline("probe").add( ... Step("load", "loader.general.contract_reference") ... ) pipeline.to_config().name 'probe'

Source code in src/openbtk/pipelines/pipeline.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    name: str,
    *,
    version: int = 1,
    description: str | None = None,
    policy: PolicyConfig | None = None,
    provenance: ProvenanceConfig | None = None,
) -> None:
    self._name = name
    self._version = version
    self._description = description
    self._policy = policy if policy is not None else PolicyConfig()
    self._provenance = provenance if provenance is not None else ProvenanceConfig()
    self._steps: list[Step] = []
    self._guardrails: list[GuardrailConfig] = []

add 

add(step: Step) -> Pipeline

Append a step. A step with no explicit after chains onto the previously added step -- the common, linear case needs no boilerplate; a config loaded from YAML always states after explicitly instead (StepConfig.after defaults to [], not "infer from position").

Source code in src/openbtk/pipelines/pipeline.py 
75
76
77
78
79
80
81
82
83
84
85
86
def add(self, step: Step) -> Pipeline:
    """Append a step. A step with no explicit ``after`` chains onto the
    previously added step -- the common, linear case needs no
    boilerplate; a config loaded from YAML always states ``after``
    explicitly instead (``StepConfig.after`` defaults to ``[]``, not
    "infer from position")."""
    after = step.after
    if after is None:
        after = [self._steps[-1].id] if self._steps else []
    resolved = Step(step.id, step.type, after=after, **step.params)
    self._steps.append(resolved)
    return self

guard 

guard(guardrail_type: str, *, at: str | list[str], on_violation: Literal['block', 'warn'] = 'block') -> Pipeline

Attach a guardrail at one or more "after:<step_id>" points.

Source code in src/openbtk/pipelines/pipeline.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def guard(
    self,
    guardrail_type: str,
    *,
    at: str | list[str],
    on_violation: Literal["block", "warn"] = "block",
) -> Pipeline:
    """Attach a guardrail at one or more ``"after:<step_id>"`` points."""
    at_list = [at] if isinstance(at, str) else list(at)
    self._guardrails.append(
        GuardrailConfig(type=guardrail_type, at=at_list, on_violation=on_violation)
    )
    return self

run 

run(*, checkpoint_path: str | Path | None = None, checkpoint_interval: int = 1000) -> RunManifest

Execute this pipeline, streaming records through every step.

Always returns a RunManifest -- success or failure (ADR-0005: there is no manifest-off switch). Never raises: a GuardrailViolation or any OpenBTKError raised during wiring or execution is caught and reported as RunManifest.status == "failed" with a PHI-free RunManifest.error instead.

Parameters:

Name Type Description Default
checkpoint_path str | Path | None

When given (FR-L-05), resume from this file if it already holds a checkpoint for this pipeline, and periodically write how far each root loader has read. A successful run deletes the file; a failed one leaves it for the next attempt. See openbtk.pipelines.checkpoint for exactly what this guarantees -- at-least-once, not exactly-once or a real seek.

 None
checkpoint_interval int

Records between checkpoint saves.

 1000
Source code in src/openbtk/pipelines/pipeline.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def run(
    self,
    *,
    checkpoint_path: str | Path | None = None,
    checkpoint_interval: int = 1000,
) -> RunManifest:
    """Execute this pipeline, streaming records through every step.

    Always returns a ``RunManifest`` -- success or failure (ADR-0005:
    there is no manifest-off switch). Never raises: a
    ``GuardrailViolation`` or any ``OpenBTKError`` raised during wiring
    or execution is caught and reported as
    ``RunManifest.status == "failed"`` with a PHI-free
    ``RunManifest.error`` instead.

    Args:
        checkpoint_path: When given (FR-L-05), resume from this file if it
            already holds a checkpoint for this pipeline, and periodically
            write how far each root loader has read. A successful run
            deletes the file; a failed one leaves it for the next attempt.
            See ``openbtk.pipelines.checkpoint`` for exactly what this
            guarantees -- at-least-once, not exactly-once or a real seek.
        checkpoint_interval: Records between checkpoint saves.
    """
    return _Executor(
        self.to_config(),
        checkpoint_path=checkpoint_path,
        checkpoint_interval=checkpoint_interval,
    ).run()

to_config 

to_config() -> PipelineConfig

The declarative PipelineConfig this builder currently describes.

Source code in src/openbtk/pipelines/pipeline.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def to_config(self) -> PipelineConfig:
    """The declarative ``PipelineConfig`` this builder currently describes."""
    return PipelineConfig(
        name=self._name,
        version=self._version,
        description=self._description,
        policy=self._policy,
        provenance=self._provenance,
        steps=[
            StepConfig(id=s.id, type=s.type, params=s.params, after=s.after or [])
            for s in self._steps
        ],
        guardrails=self._guardrails,
    )

validate 

validate() -> list[ValidationIssue]

Everything that can be checked without instantiating a component or touching data: the config's registry, parameter and policy checks (:meth:PipelineConfig.validate_registry) plus the executor's own shape rules (no repeated predecessor, loaders are roots). An empty list means run() will not be refused for any of these reasons -- not that it will succeed.

Source code in src/openbtk/pipelines/pipeline.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def validate(self) -> list[ValidationIssue]:
    """Everything that can be checked without instantiating a component or
    touching data: the config's registry, parameter and policy checks
    (:meth:`PipelineConfig.validate_registry`) plus the executor's own
    shape rules (no repeated predecessor, loaders are roots). An empty list means
    ``run()`` will not be refused for any of these reasons -- not that it will
    succeed.
    """
    config = self.to_config()
    issues = config.validate_registry()
    try:
        _validate_shape(config.steps)
    except ConfigError as e:
        issues.append(ValidationIssue(severity="error", message=str(e)))
    return issues

openbtk.pipelines.pipeline.Step 

Step(step_id: str, type: str, *, after: list[str] | None = None, **params: Any)

One pipeline step, for the programmatic builder API.

Example

step = Step("load", "loader.clinical_text.plain_text", path="./notes") step.params {'path': './notes'}

Source code in src/openbtk/pipelines/pipeline.py 
38
39
40
41
42
43
44
def __init__(
    self, step_id: str, type: str, *, after: list[str] | None = None, **params: Any
) -> None:
    self.id = step_id
    self.type = type
    self.after = after
    self.params: dict[str, Any] = params

Checkpoint and resume

Pipeline.run(checkpoint_path=...) resumes a long run instead of restarting it; see the CLI guide for what this does and does not guarantee before relying on it.

openbtk.pipelines.checkpoint.Checkpoint

Bases: BaseModel

Resume state for one pipeline: how far each root loader step has read.

Example

from datetime import UTC, datetime Checkpoint( ... pipeline_name="ingest", run_id="a1b2c3", ... loader_counts={"load": 4512003}, ... saved_at=datetime(2026, 1, 1, tzinfo=UTC), ... ).loader_counts["load"] 4512003

openbtk.pipelines.checkpoint.save_checkpoint 

save_checkpoint(path: str | Path, checkpoint: Checkpoint) -> None

Write checkpoint to path atomically (write to a temp file, then rename), so a crash mid-write never leaves a truncated, unreadable checkpoint file.

Source code in src/openbtk/pipelines/checkpoint.py 
74
75
76
77
78
79
80
81
def save_checkpoint(path: str | Path, checkpoint: Checkpoint) -> None:
    """Write ``checkpoint`` to ``path`` atomically (write to a temp file, then rename),
    so a crash mid-write never leaves a truncated, unreadable checkpoint file."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = target.with_name(f"{target.name}.tmp-{os.getpid()}")
    tmp.write_text(checkpoint.model_dump_json(indent=2) + "\n", encoding="utf-8")
    tmp.replace(target)

openbtk.pipelines.checkpoint.load_checkpoint 

load_checkpoint(path: str | Path) -> Checkpoint

Read a checkpoint file.

Raises:

Type Description
ConfigError

If it cannot be read or parsed. Resuming from a checkpoint you cannot trust would risk silently skipping unprocessed records, so this refuses rather than falling back to "start from the beginning."
Source code in src/openbtk/pipelines/checkpoint.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def load_checkpoint(path: str | Path) -> Checkpoint:
    """Read a checkpoint file.

    Raises:
        ConfigError: If it cannot be read or parsed. Resuming from a checkpoint you
            cannot trust would risk silently skipping unprocessed records, so this
            refuses rather than falling back to "start from the beginning."
    """
    target = Path(path)
    try:
        raw = target.read_text(encoding="utf-8")
        return Checkpoint.model_validate_json(raw)
    except (OSError, ValueError) as e:
        raise ConfigError(
            f"Could not read checkpoint file {target}: {e}",
            context={"checkpoint": str(target)},
        ) from e

Cross-modal joins

Attach a patient's structured events to their notes; see the EHR guide.

openbtk.pipelines.join.join_notes_to_events 

join_notes_to_events(notes: Iterable[ClinicalTextRecord], patients: Mapping[str, PatientRecord] | Callable[[str], PatientRecord | None], *, on: JoinOn = 'either', before: timedelta = timedelta(hours=24), after: timedelta = timedelta(0)) -> Iterator[NoteWithEvents]

Join each note to its patient's events. Yields one result per note, in order.

Parameters:

Name Type Description Default
notes Iterable[ClinicalTextRecord]

The notes, streamed.

 required
patients Mapping[str, PatientRecord] | Callable[[str], PatientRecord | None]

A mapping of patient_id to record (see :func:index_patients), or a function returning the record for an id (or None).

 required
on JoinOn

"encounter", "window" or "either"; see the module docstring.

 'either'
before timedelta

How far before a note's timestamp an event may be and still join.

 timedelta(hours=24)
after timedelta

How far after it.

 timedelta(0)

Raises:

Type Description
ValueError

If before or after is negative.
Example

from datetime import UTC, datetime from openbtk.core.schemas import CodeSystem from openbtk.data.ehr.schemas import ( ... Demographics, Measurement, PatientRecord, ... ) noon = datetime(2024, 3, 14, 12, tzinfo=UTC) patient = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(), ... observations=[Measurement(code="718-7", value=13.5, timestamp=noon)], ... source_system="fhir-r4", ... ) note = ClinicalTextRecord( ... record_id="n1", source="s", text="Discharge summary.", ... patient_ref="pt-1", timestamp=noon.replace(hour=18), ... ) (joined,) = join_notes_to_events([note], index_patients([patient])) matches = [(e.kind, e.matched_by) for e in joined.events] matches [('observation', 'window')]

Source code in src/openbtk/pipelines/join.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def join_notes_to_events(
    notes: Iterable[ClinicalTextRecord],
    patients: Mapping[str, PatientRecord] | Callable[[str], PatientRecord | None],
    *,
    on: JoinOn = "either",
    before: timedelta = timedelta(hours=24),
    after: timedelta = timedelta(0),
) -> Iterator[NoteWithEvents]:
    """Join each note to its patient's events. Yields one result per note, in order.

    Args:
        notes: The notes, streamed.
        patients: A mapping of ``patient_id`` to record (see :func:`index_patients`), or
            a function returning the record for an id (or ``None``).
        on: ``"encounter"``, ``"window"`` or ``"either"``; see the module docstring.
        before: How far before a note's timestamp an event may be and still join.
        after: How far after it.

    Raises:
        ValueError: If ``before`` or ``after`` is negative.

    Example:
        >>> from datetime import UTC, datetime
        >>> from openbtk.core.schemas import CodeSystem
        >>> from openbtk.data.ehr.schemas import (
        ...     Demographics, Measurement, PatientRecord,
        ... )
        >>> noon = datetime(2024, 3, 14, 12, tzinfo=UTC)
        >>> patient = PatientRecord(
        ...     patient_id="pt-1",
        ...     demographics=Demographics(),
        ...     observations=[Measurement(code="718-7", value=13.5, timestamp=noon)],
        ...     source_system="fhir-r4",
        ... )
        >>> note = ClinicalTextRecord(
        ...     record_id="n1", source="s", text="Discharge summary.",
        ...     patient_ref="pt-1", timestamp=noon.replace(hour=18),
        ... )
        >>> (joined,) = join_notes_to_events([note], index_patients([patient]))
        >>> matches = [(e.kind, e.matched_by) for e in joined.events]
        >>> matches
        [('observation', 'window')]
    """
    if before < timedelta(0) or after < timedelta(0):
        raise ValueError("before and after must not be negative")
    return _join(notes, patients, on, before, after)

openbtk.pipelines.join.index_patients 

index_patients(patients: Iterable[PatientRecord]) -> dict[str, PatientRecord]

A lookup of patients by patient_id for join_notes_to_events.

Holds every patient record, so for a large cohort supply a function that fetches a patient on demand instead. A repeated patient_id keeps the last record.

Example

from openbtk.data.ehr.schemas import Demographics, PatientRecord record = PatientRecord( ... patient_id="pt-1", demographics=Demographics(), source_system="fhir-r4" ... ) list(index_patients([record])) ['pt-1']

Source code in src/openbtk/pipelines/join.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def index_patients(patients: Iterable[PatientRecord]) -> dict[str, PatientRecord]:
    """A lookup of ``patients`` by ``patient_id`` for ``join_notes_to_events``.

    Holds every patient record, so for a large cohort supply a function that fetches a
    patient on demand instead. A repeated ``patient_id`` keeps the last record.

    Example:
        >>> from openbtk.data.ehr.schemas import Demographics, PatientRecord
        >>> record = PatientRecord(
        ...     patient_id="pt-1", demographics=Demographics(), source_system="fhir-r4"
        ... )
        >>> list(index_patients([record]))
        ['pt-1']
    """
    return {patient.patient_id: patient for patient in patients}

openbtk.pipelines.join.NoteWithEvents

Bases: BaseModel

A note and the structured events joined to it, in time order.

openbtk.pipelines.join.MatchedEvent

Bases: BaseModel

One structured event joined to a note, and why it matched.

Example

from openbtk.core.schemas import CodeSystem matched = MatchedEvent( ... kind="condition", ... event=CodedEvent(code="385093006", system=CodeSystem.SNOMED), ... matched_by="encounter", ... ) matched.kind 'condition'