Pipeline

Pipeline builds and runs a configured DAG. Extract, transform, join, union, and load steps are declared in the manifest; connector classes remain available separately for direct use.

Build and execute

from pycharter import Pipeline

pipeline = Pipeline.from_dict({
    "name": "users",
    "steps": [
        {
            "id": "source",
            "type": "extract",
            "source": {"type": "file", "path": "input.json", "format": "json"},
        },
        {
            "id": "sink",
            "type": "load",
            "input": "source",
            "target": {
                "type": "file", "path": "output.json", "format": "json",
                "mode": "overwrite",
            },
        },
    ],
})
result = pipeline.run()  # Synchronous; call outside a running event loop.
print(f"Loaded {result.rows_loaded} rows")

Inside an async application, use result = await pipeline.arun() instead. Load a manifest with Pipeline.from_file("pipelines/users/pipeline.yaml"), or its directory with Pipeline.from_dir("pipelines/users"). Pass declared variable overrides with variables={"OUTPUT_PATH": "/tmp/users.json"}.

API reference

Pipeline

Pipeline(
    config: PipelineConfig,
    steps: list[StepConfig],
    *,
    base_dir: Path | None = None,
)

DAG-based ETL pipeline.

Pipelines are constructed from YAML config files or dicts and consist of a directed acyclic graph of steps (extract, transform, join, union, load).

Example::

pipeline = Pipeline.from_file("pipeline.yaml")
result = pipeline.run()            # sync
result = await pipeline.arun()     # inside a running event loop

Source code in src/pycharter/pipeline_generator/pipeline.py

def __init__(
    self,
    config: PipelineConfig,
    steps: list[StepConfig],
    *,
    base_dir: Path | None = None,
) -> None:
    self._config = config
    self._steps = steps
    self._base_dir = base_dir

from_dict classmethod

from_dict(
    raw: dict[str, Any],
    *,
    variables: dict[str, str] | None = None,
) -> Pipeline

Create a pipeline from a config dict.

Accepts both step-based (steps: [...]) and section-based (extract/transform/load) configs.

Parameters:

Name	Type	Description	Default
raw	dict[str, Any]	Raw config dictionary.	required
variables	dict[str, str] | None	Additional variables.	None

Returns:

Type	Description
Pipeline	A validated, ready-to-run pipeline.

Source code in src/pycharter/pipeline_generator/pipeline.py

@classmethod
def from_dict(
    cls,
    raw: dict[str, Any],
    *,
    variables: dict[str, str] | None = None,
) -> Pipeline:
    """Create a pipeline from a config dict.

    Accepts both step-based (``steps: [...]``) and section-based
    (``extract/transform/load``) configs.

    Args:
        raw: Raw config dictionary.
        variables: Additional variables.

    Returns:
        A validated, ready-to-run pipeline.
    """
    config = parse_pipeline_config(raw, variables=variables)
    sorted_steps = resolve_dag(config.steps)
    return cls(config=config, steps=sorted_steps)

from_file classmethod

from_file(
    path: str | Path,
    *,
    variables: dict[str, str] | None = None,
) -> Pipeline

Create a pipeline from a YAML config file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the pipeline YAML file.	required
variables	dict[str, str] | None	Additional variables merged on top of the file's variables section (caller wins).	None

Returns:

Type	Description
Pipeline	A validated, ready-to-run pipeline.

Source code in src/pycharter/pipeline_generator/pipeline.py

@classmethod
def from_file(
    cls,
    path: str | Path,
    *,
    variables: dict[str, str] | None = None,
) -> Pipeline:
    """Create a pipeline from a YAML config file.

    Args:
        path: Path to the pipeline YAML file.
        variables: Additional variables merged on top of the
            file's ``variables`` section (caller wins).

    Returns:
        A validated, ready-to-run pipeline.
    """
    config = load_pipeline_config(path, variables=variables)
    sorted_steps = resolve_dag(config.steps)
    return cls(
        config=config,
        steps=sorted_steps,
        base_dir=Path(path).parent.resolve(),
    )

from_dir classmethod

from_dir(
    directory: str | Path,
    *,
    variables: dict[str, str] | None = None,
) -> Pipeline

Create a pipeline from a directory containing pipeline.yaml.

Parameters:

Name	Type	Description	Default
directory	str | Path	Path to the config directory.	required
variables	dict[str, str] | None	Additional variables.	None

Returns:

Type	Description
Pipeline	A validated, ready-to-run pipeline.

Raises:

Type	Description
FileNotFoundError	If pipeline.yaml (or .yml) is missing.

Source code in src/pycharter/pipeline_generator/pipeline.py

@classmethod
def from_dir(
    cls,
    directory: str | Path,
    *,
    variables: dict[str, str] | None = None,
) -> Pipeline:
    """Create a pipeline from a directory containing ``pipeline.yaml``.

    Args:
        directory: Path to the config directory.
        variables: Additional variables.

    Returns:
        A validated, ready-to-run pipeline.

    Raises:
        FileNotFoundError: If ``pipeline.yaml`` (or ``.yml``) is missing.
    """
    dirpath = Path(directory)
    if not dirpath.is_dir():
        raise FileNotFoundError(f"Config directory not found: {dirpath}")

    pipeline_file = _find_yaml(dirpath, "pipeline")
    if not pipeline_file:
        raise FileNotFoundError(
            f"Config directory must contain pipeline.yaml: {dirpath}"
        )
    return cls.from_file(pipeline_file, variables=variables)

from_store classmethod

from_store(
    ref: StoreRef | str,
    *,
    store: PipelineStoreClient | None = None,
    variables: dict[str, str] | None = None,
) -> Pipeline

Create a pipeline from a config stored in a pipeline store.

Parameters:

Name	Type	Description	Default
ref	StoreRef | str	A :class:~pycharter.shared.refs.StoreRef (or "name" / "name@version" string) addressing the stored pipeline config. A None version resolves to the latest version by numeric-aware ordering.	required
store	PipelineStoreClient | None	An open :class:PipelineStoreClient. When omitted, a default store is opened via :func:pycharter.open_store ("pipelines").	None
variables	dict[str, str] | None	Additional variables merged on top of the stored config's variables section (caller wins).	None

Returns:

Type	Description
Pipeline	A validated, ready-to-run pipeline.

Raises:

Type	Description
FileNotFoundError	If no config exists for ref.

Source code in src/pycharter/pipeline_generator/pipeline.py

@classmethod
def from_store(
    cls,
    ref: StoreRef | str,
    *,
    store: PipelineStoreClient | None = None,
    variables: dict[str, str] | None = None,
) -> Pipeline:
    """Create a pipeline from a config stored in a pipeline store.

    Args:
        ref: A :class:`~pycharter.shared.refs.StoreRef` (or
            ``"name"`` / ``"name@version"`` string) addressing the
            stored pipeline config. A ``None`` version resolves to the
            latest version by numeric-aware ordering.
        store: An open :class:`PipelineStoreClient`. When omitted, a
            default store is opened via
            :func:`pycharter.open_store` (``"pipelines"``).
        variables: Additional variables merged on top of the
            stored config's ``variables`` section (caller wins).

    Returns:
        A validated, ready-to-run pipeline.

    Raises:
        FileNotFoundError: If no config exists for ``ref``.
    """
    from pycharter.shared.refs import StoreRef

    ref = StoreRef.parse(ref)
    owns_store = store is None
    if store is None:
        from pycharter.stores import open_store

        store = open_store("pipelines")

    try:
        env = store.get(ref.name, ref.version)
        if env is None:
            raise FileNotFoundError(f"No pipeline config found in store for {ref}")
        doc = env["document"]
        config_dict: dict[str, Any] = {
            "name": env["name"],
            "version": env["version"],
            "steps": doc["steps"],
        }
        if doc.get("settings"):
            config_dict["settings"] = doc["settings"]
        if doc.get("variables"):
            config_dict["variables"] = doc["variables"]
        if doc.get("description"):
            config_dict["description"] = doc["description"]
    finally:
        if owns_store:
            with suppress(Exception):
                store.disconnect()

    return cls.from_dict(config_dict, variables=variables)

preflight

preflight(
    *, privacy_secret: str | None = None
) -> PreflightResult

Run checks without source or destination connector I/O.

Walks every step's optional contract: block, resolves the referenced contract (file path or contract-store ref), and collects any resolution errors into a :class:PreflightResult. The resolved contexts are then checked for privacy-clearance problems. Extractors and loaders are not constructed; no records flow.

Contract files and configured contract stores are read. For offline validation, use file refs without settings.contract_store.

Use as a CI / orchestrator gate to fail before starting an actual run when contract refs are broken — typical signal:

result = pipeline.preflight()
if not result.success:
    for issue in result.issues:
        print(f"{issue.step_id}: {issue.message}")
    sys.exit(1)

:meth:arun runs the same pass before the first step, so a pipeline never starts in a state preflight would reject.

Parameters:

Name	Type	Description	Default
privacy_secret	str | None	The org secret the run will use, if the caller passes one explicitly. Honoured by the [missing_secret] privacy rule so a secret supplied to :meth:run is not reported as absent.	None

Returns:

Name	Type	Description
A	PreflightResult	class:PreflightResult with success and the list of
	PreflightResult	per-step :class:PreflightIssue entries.

Source code in src/pycharter/pipeline_generator/pipeline.py

def preflight(self, *, privacy_secret: str | None = None) -> PreflightResult:
    """Run checks without source or destination connector I/O.

    Walks every step's optional ``contract:`` block, resolves the
    referenced contract (file path or contract-store ref), and
    collects any resolution errors into a :class:`PreflightResult`.
    The resolved contexts are then checked for privacy-clearance
    problems. Extractors and loaders are not constructed; no records
    flow.

    Contract files and configured contract stores are read. For offline
    validation, use file refs without ``settings.contract_store``.

    Use as a CI / orchestrator gate to fail before starting an
    actual run when contract refs are broken — typical signal:

        result = pipeline.preflight()
        if not result.success:
            for issue in result.issues:
                print(f"{issue.step_id}: {issue.message}")
            sys.exit(1)

    :meth:`arun` runs the same pass before the first step, so a
    pipeline never starts in a state preflight would reject.

    Args:
        privacy_secret: The org secret the run will use, if the
            caller passes one explicitly. Honoured by the
            ``[missing_secret]`` privacy rule so a secret supplied
            to :meth:`run` is not reported as absent.

    Returns:
        A :class:`PreflightResult` with ``success`` and the list of
        per-step :class:`PreflightIssue` entries.
    """
    if execution_issues(self._steps):
        return preflight_steps(self._steps)
    contract_store = create_contract_store(self._config.settings)
    if contract_store is None:
        return preflight_steps(
            self._steps,
            base_dir=self._base_dir,
            contract_store=None,
            privacy_secret=privacy_secret,
        )

    # The store may itself fail to connect (bad URL, missing
    # credentials, unreachable backend). That is itself a preflight
    # finding — surface it as a pipeline-level issue rather than
    # raising, so the caller still gets a structured result.
    try:
        contract_store.connect()
    except Exception as exc:  # noqa: BLE001 — preflight collects, never crashes
        result = PreflightResult()
        result.issues.append(
            PreflightIssue(
                step_id="",
                category="contract_store",
                message=f"contract store not reachable: {exc}",
            )
        )
        return result

    try:
        return preflight_steps(
            self._steps,
            base_dir=self._base_dir,
            contract_store=contract_store,
            privacy_secret=privacy_secret,
        )
    finally:
        _safe_disconnect(contract_store)

run

run(
    *,
    run_id: str | None = None,
    fail_fast: bool | None = None,
    on_error: str | None = None,
    privacy_secret: str | None = None,
) -> PipelineResult

Execute the pipeline synchronously.

Convenience wrapper that calls asyncio.run() on the async :meth:arun method (see it for the on_error / fail_fast / privacy_secret semantics). Unusable inside a running event loop — await :meth:arun there instead.

Source code in src/pycharter/pipeline_generator/pipeline.py

def run(
    self,
    *,
    run_id: str | None = None,
    fail_fast: bool | None = None,
    on_error: str | None = None,
    privacy_secret: str | None = None,
) -> PipelineResult:
    """Execute the pipeline synchronously.

    Convenience wrapper that calls ``asyncio.run()`` on the async
    :meth:`arun` method (see it for the ``on_error`` / ``fail_fast`` /
    ``privacy_secret`` semantics). Unusable inside a running event
    loop — await :meth:`arun` there instead.
    """
    return asyncio.run(
        self.arun(
            run_id=run_id,
            fail_fast=fail_fast,
            on_error=on_error,
            privacy_secret=privacy_secret,
        )
    )

arun async

arun(
    *,
    run_id: str | None = None,
    fail_fast: bool | None = None,
    on_error: str | None = None,
    progress: RunProgressEmitter | None = None,
    extraction_sink_factory: Any | None = None,
    grounded_emitter_factory: Any | None = None,
    dry_run: bool = False,
    io: PipelineIO | None = None,
    privacy_secret: str | None = None,
) -> PipelineResult

Execute the pipeline (async; awaitable inside a running loop).

Parameters:

Name	Type	Description	Default
run_id	str | None	Optional identifier for this run.	None
privacy_secret	str | None	Explicit org secret used to tokenize pii fields at load boundaries under privacy.mode: enforce (falls back to PYCHARTER_KB_ORG_SECRET). Never read from pipeline YAML.	None
fail_fast	bool | None	Abort on the first step failure. Soft-deprecated — prefer on_error; kept for backward compatibility.	None
on_error	str | None	Run-level failure behaviour using the shared vocabulary of step contract.on_error blocks: "fail" aborts on the first step failure (default), "warn" logs and continues with remaining runnable steps. Per-record quarantine routing is configured on the step's contract block, not here. Mutually exclusive with fail_fast.	None
progress	RunProgressEmitter | None	Optional Phase-4 live-progress emitter. When supplied, the runner publishes pipeline_start / step_start / step_complete / step_failed / pipeline_complete / pipeline_failed events to the configured ChangeBus; the SSE endpoint at GET /api/v1/pipeline/runs/{run_id}/stream relays them to UI / CLI clients.	None
extraction_sink_factory	Any | None	Optional Theme-D production hook. A callable that, given a :class:LoadStepConfig, returns an :class:AutoExtractionSink (or None to skip extraction for that step). Build via :func:pycharter.semantic.extraction.sink_factory.make_extraction_sink_factory. When omitted, the pipeline runs with no structural extraction (pre-Theme-D behaviour).	None
grounded_emitter_factory	Any | None	Optional KB-enrichment hook. A callable that, given a :class:LoadStepConfig, returns a :class:pycharter.semantic.kb.emitter.GroundedRecordEmitter. Build via :func:pycharter.semantic.kb.emitter.make_grounded_emitter_factory. Capture is opt-in per contract (kb_capture); when omitted, no GroundedRecords are emitted.	None
dry_run	bool	Read sources and run transforms and contract checks, suppressing normal and quarantine destination writes. This is an extraction preview, not offline planning or a sandbox for retained Python. Lineage/progress may still emit events.	False
io	PipelineIO | None	Explicit run-local connector substitutions keyed by step ID. Missing substitutions fail without configured fallbacks. Auxiliary configured I/O and callbacks are rejected. See :class:pycharter.pipeline_generator.io.PipelineIO.	None

Returns:

Type	Description
PipelineResult	Aggregated pipeline result with per-step results. When
PipelineResult	preflight reports errors, a failed result carrying those
PipelineResult	messages is returned and no step runs.

Source code in src/pycharter/pipeline_generator/pipeline.py

async def arun(
    self,
    *,
    run_id: str | None = None,
    fail_fast: bool | None = None,
    on_error: str | None = None,
    progress: RunProgressEmitter | None = None,
    extraction_sink_factory: Any | None = None,
    grounded_emitter_factory: Any | None = None,
    dry_run: bool = False,
    io: PipelineIO | None = None,
    privacy_secret: str | None = None,
) -> PipelineResult:
    """Execute the pipeline (async; awaitable inside a running loop).

    Args:
        run_id: Optional identifier for this run.
        privacy_secret: Explicit org secret used to tokenize ``pii``
            fields at load boundaries under ``privacy.mode: enforce``
            (falls back to ``PYCHARTER_KB_ORG_SECRET``). Never read
            from pipeline YAML.
        fail_fast: Abort on the first step failure. Soft-deprecated —
            prefer ``on_error``; kept for backward compatibility.
        on_error: Run-level failure behaviour using the shared
            vocabulary of step ``contract.on_error`` blocks:
            ``"fail"`` aborts on the first step failure (default),
            ``"warn"`` logs and continues with remaining runnable
            steps. Per-record ``quarantine`` routing is configured on
            the step's contract block, not here. Mutually exclusive
            with ``fail_fast``.
        progress: Optional Phase-4 live-progress emitter. When
            supplied, the runner publishes ``pipeline_start`` /
            ``step_start`` / ``step_complete`` / ``step_failed`` /
            ``pipeline_complete`` / ``pipeline_failed`` events to
            the configured ``ChangeBus``; the SSE endpoint at
            ``GET /api/v1/pipeline/runs/{run_id}/stream`` relays
            them to UI / CLI clients.
        extraction_sink_factory: Optional Theme-D production hook. A
            callable that, given a :class:`LoadStepConfig`, returns
            an :class:`AutoExtractionSink` (or ``None`` to skip
            extraction for that step). Build via
            :func:`pycharter.semantic.extraction.sink_factory.make_extraction_sink_factory`.
            When omitted, the pipeline runs with no structural
            extraction (pre-Theme-D behaviour).
        grounded_emitter_factory: Optional KB-enrichment hook. A
            callable that, given a :class:`LoadStepConfig`, returns a
            :class:`pycharter.semantic.kb.emitter.GroundedRecordEmitter`.
            Build via
            :func:`pycharter.semantic.kb.emitter.make_grounded_emitter_factory`.
            Capture is opt-in per contract (``kb_capture``); when
            omitted, no GroundedRecords are emitted.
        dry_run: Read sources and run transforms and contract checks,
            suppressing normal and quarantine destination writes. This
            is an extraction preview, not offline planning or a sandbox
            for retained Python. Lineage/progress may still emit events.
        io: Explicit run-local connector substitutions keyed by step ID.
            Missing substitutions fail without configured fallbacks.
            Auxiliary configured I/O and callbacks are rejected. See
            :class:`pycharter.pipeline_generator.io.PipelineIO`.

    Returns:
        Aggregated pipeline result with per-step results. When
        preflight reports errors, a failed result carrying those
        messages is returned and no step runs.
    """
    if execution_issues(self._steps):
        gate = preflight_steps(self._steps)
        return _preflight_failure(self._config.name, run_id, gate)
    effective_fail_fast = resolve_on_error(fail_fast, on_error)
    if io is not None:
        io.check_settings(self._config.settings)
        io.check_hooks(progress, extraction_sink_factory, grounded_emitter_factory)

    # Run-time variable resolution: re-resolve the retained raw steps
    # strictly against the environment + overrides, so ${VAR}
    # placeholders pick up the run environment and ${VAR:?} fails fast
    # here (not at offline validation). Falls back to the parse-time
    # steps for configs built without a retained raw form.
    if self._config.raw_steps:
        runtime_vars, runtime_steps = resolve_runtime_steps(self._config)
        runtime_steps = resolve_dag(runtime_steps)
    else:
        runtime_vars, runtime_steps = self._config.variables, self._steps

    if execution_issues(runtime_steps):
        gate = preflight_steps(runtime_steps)
        return _preflight_failure(self._config.name, run_id, gate)
    contract_store = create_contract_store(self._config.settings)
    if contract_store is not None:
        # Wrap connect() so a failure here doesn't leak the
        # half-initialised store — some drivers (e.g. SQLAlchemy
        # engines) hold resources even on a failed connect, and the
        # previous shape called disconnect() in `finally` only when
        # we entered the run_dag try block (after connect()
        # already succeeded).
        try:
            contract_store.connect()
        except Exception:
            _safe_disconnect(contract_store)
            raise

    # Build the OpenLineage emitter from `settings.lineage`. Returns
    # None when lineage is disabled, missing, or when the
    # `openlineage-python` package is not installed; in that case
    # `run_dag` runs with zero overhead.
    lineage_emitter = create_lineage_emitter(self._config.settings)

    try:
        # Config-side gate: a pipeline never starts in a state preflight
        # would reject (proposal §2 A4). It runs against the *runtime*
        # steps — after ``${VAR}`` re-resolution, so a contract ref that
        # resolves at run time is not refused for its placeholder form —
        # and reuses the contract store connection the run itself will
        # use. Warnings inform and the run proceeds; errors abort before
        # any extractor or loader exists.
        gate = preflight_steps(
            runtime_steps,
            base_dir=self._base_dir,
            contract_store=contract_store,
            privacy_secret=privacy_secret,
        )
        for issue in gate.warnings:
            logger.warning("Preflight %s", _format_issue(issue))
        if not gate.success:
            return _preflight_failure(self._config.name, run_id, gate)
        return await run_dag(
            runtime_steps,
            pipeline_name=self._config.name,
            run_id=run_id,
            variables=runtime_vars,
            fail_fast=effective_fail_fast,
            base_dir=self._base_dir,
            contract_store=contract_store,
            lineage_emitter=lineage_emitter,
            progress=progress,
            extraction_sink_factory=extraction_sink_factory,
            grounded_emitter_factory=grounded_emitter_factory,
            dry_run=dry_run,
            io=io,
            privacy_secret=privacy_secret,
        )
    finally:
        if contract_store is not None:
            _safe_disconnect(contract_store)

PipelineResult

PipelineResult dataclass

PipelineResult(
    success: bool = True,
    rows_extracted: int = 0,
    rows_transformed: int = 0,
    rows_loaded: int = 0,
    rows_failed: int = 0,
    rows_quarantined_extract: int = 0,
    rows_quarantined_load: int = 0,
    validation_errors_extract: list[str] = list(),
    validation_errors_load: list[str] = list(),
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    duration_seconds: float | None = None,
    batches_processed: int = 0,
    batch_results: list[BatchResult] = list(),
    errors: list[str] = list(),
    pipeline_name: str | None = None,
    run_id: str | None = None,
    quality_report: Any = None,
    privacy_report: Any = None,
    lineage_events: list[dict[str, Any]] = list(),
    step_results: list[StepResult] = list(),
    rows_quarantined_other: int = 0,
    record_outcomes: RecordOutcomes = RecordOutcomes(),
    quality_summary: QualitySummary = QualitySummary(),
)

Complete result from running an ETL pipeline.

success class-attribute instance-attribute

success: bool = True

rows_extracted class-attribute instance-attribute

rows_extracted: int = 0

rows_transformed class-attribute instance-attribute

rows_transformed: int = 0

rows_loaded class-attribute instance-attribute

rows_loaded: int = 0

rows_failed class-attribute instance-attribute

rows_failed: int = 0

record_outcomes class-attribute instance-attribute

record_outcomes: RecordOutcomes = field(
    default_factory=RecordOutcomes
)

quality_summary class-attribute instance-attribute

quality_summary: QualitySummary = field(
    default_factory=QualitySummary
)

total_quarantined property

total_quarantined: int

Acknowledged quarantine writes across all DAG step types.

step_results class-attribute instance-attribute

step_results: list[StepResult] = field(default_factory=list)

duration_seconds class-attribute instance-attribute

duration_seconds: float | None = None

errors class-attribute instance-attribute

errors: list[str] = field(default_factory=list)

rows_failed counts loader-reported failures. Contract validation and routing are reported separately in record_outcomes; total_quarantined counts acknowledged DLQ writes. Each step retains its input, pre-contract, and output cardinality, validation outcomes, and scoped quality reports, including partial results after a failure. See record outcome accounting and pipeline quality.

Failure behavior

The default on_error="fail" aborts on the first step failure. Use on_error="warn" to continue with remaining runnable steps. Inspect the returned result in either case:

result = await pipeline.arun(on_error="warn")
if not result.success:
    for error in result.errors:
        print(error)

This run-level policy is separate from per-record contract.on_error routing. Every step contract selecting quarantine requires a declared DLQ destination. A failed run can retain acknowledged destination writes; it does not imply a rollback. A loader exception can leave delivery unknown.

Preview and offline validation

pipeline.preflight() resolves and checks contracts without constructing source or destination connectors. A configured contract registry can still require network reads; use local contract files for offline checks.

await pipeline.arun(dry_run=True) is an extraction preview. It reads sources and executes transforms and contract checks while suppressing normal load and quarantine destination writes. It reports quarantine candidates as record_outcomes.would_quarantine. Retained Python runs with the process's permissions; connector substitution does not sandbox it.

Use PipelineTestHarness to substitute every configured DAG source, destination, and quarantine connector with fixture and capture implementations, with per-step captures and no configured fallback.

Checkpoints and queue consumers

The native batch DAG does not persist incremental watermarks or automatically acknowledge messaging extractor batches after loading. Keep checkpoint and consumer acknowledgment ownership explicit in the scheduled job or use the streaming runtime for its documented broker lifecycle. source.incremental is rejected; standalone state helpers do not establish checkpoint behavior in a DAG manifest.

settings.state_store and settings.plugins are retained configuration fields awaiting native DAG integration. They do not automatically persist checkpoints or import the specified connector classes. Standalone helpers remain available; see retained settings for their current scope. Installed entry-point connectors and explicit factory registration remain separate supported mechanisms.

See also

• ETL pipelines tutorial
• Connector settings
• Testing pipelines
• Extractors
• Transformers
• Loaders