Pipeline connector settings

The source: of an extract step and the target: of a load step are validated against the connector capability registry when a pipeline loads. Every built-in connector has a strict Pydantic schema. Inspect the accepted keys with pycharter capabilities --kind extractor or pycharter capabilities --kind loader, then use describe_capability for its config_schema. The MCP validate_pipeline_config tool applies the same validation as Pipeline.from_file and reports the step, field path, and a suggested spelling for close typos.

Custom connectors used in a pipeline must register a strict config_model with their capability. A plugin without one receives config_model_required from pipeline loading and MCP validation. Direct ExtractorFactory.create and LoaderFactory.create plugin calls keep their existing behavior.

Migration from settings with no runtime effect

This is a configuration semantics change: the class-based pipeline runtime previously accepted the following keys while ignoring them. Pipeline loading and MCP validation now return unsupported_field with a suggested action. Direct calls to connector factories remain compatible.

Previous setting	Action in a pipeline
HTTP max_calls_per_minute	Use rate_limit_delay in seconds between calls.
HTTP input_params / default_param_values	Put request values in params or pass declared pipeline variables.
Extractor incremental	Manage a checkpoint outside the connector and pass its value through variables or a SQL query. The class-based runtime has no persisted watermark.
File file_format	Rename to format on file extractors and loaders.
File extractor csv_options / json_options	Use the default parser or preprocess the file before the pipeline.
File loader csv_options / json_options / batch_size	Use the default writer; the loader writes the batch supplied by the pipeline.
Cloud loader mode / partition_by / batch_size	Choose the object path before loading; the loader writes the supplied batch.
SQL/Mongo write_method / primary_key	Rename to mode / match_on.

HTTP response_format: text and response_format: xml are also rejected in pipelines: the class-based extractor reads the text but emits no records. Use a JSON endpoint with response_format: json. Supplying both forms of a connector alias (for example path and file_path, or database.url and connection_string) returns conflicting_field because one value would be shadowed.

The HTTP paginator accepts nested page, offset, cursor, and next_url blocks, plus stop_conditions and page_delay. Older flat fields such as page_param, cursor_path, and max_pages do not reach the runtime; use the nested form in src/pycharter/data/templates/pipeline/snippets/extract_http_paginated.yaml. For Azure cloud sources and targets, put the container name in storage.bucket; the class-based connector forwards that key as the Azure container.

SQLite targets can use type: sqlite with path, which the loader converts to a SQLite connection URL. SQL target settings are parsed consistently for SQLite and Postgres; persistence backends for pipeline metadata are not changed by connector validation.

For a generic type: database target, update_columns, skip_schema_check, and jsonb_columns require a PostgreSQL URL. The generic loader ignores them with SQLite, MySQL, and MSSQL URLs, so pipeline loading rejects those combinations. SQLite also ignores db_schema; remove it or use a backend with schema support.

DAG semantic validation

Native DAG loading validates settings, step contract blocks, transform options, and legacy load-step quality checks before constructing configured connectors. File, dict, stored-pipeline, CLI, and MCP loading use shared semantic checks. Diagnostics identify the step and field path without including rejected values. Validation does not import custom transform functions or resolve contract references; unresolved secret placeholders remain valid for offline authoring. The published JSON Schema checks manifest shape; these shared validators also check options inside the retained dict/list fields. Run pipeline validation alongside schema validation before execution.

This tightens configuration acceptance. Earlier versions could accept these values and then execute a different operation, skip a check, or report success:

Previously accepted value	Migration
Custom Python or JSONata mode: batc	Choose batch or record; omission retains the batch default.
Filter operator: equ	Use a supported operator such as eq.
Conversion target integre	Use integer or int.
Unknown legacy quality check type or severity	Choose a supported load-step check and severity.
Unknown settings or step contract fields	Remove the field or correct its spelling.

Valid operation ordering, supported aliases, and runtime registered connectors remain supported. Canonical file/artifact loading still restricts connector types to its built-in union; runtime dict/raw-MCP/store paths can use registered extensions with strict models. Unifying those extension authoring paths remains unfinished. These checks do not establish executable semantics for every quality assertion, engine, phase, or operational policy.

Retained settings awaiting native integration

settings.state_store and settings.plugins remain accepted for compatibility and future integration. The native batch DAG currently does not use either setting:

Setting	Existing capability	Native DAG integration still needed
state_store: {backend: file, path: ./.state}	Standalone file/SQLite helpers persist watermarks and run metadata when called explicitly.	Reading/injecting checkpoints and deciding when to advance them after partial or failed delivery.
plugins: {extractors: {}, loaders: {}}	An explicit helper imports and registers connector classes. Installed entry-point plugins and explicit factory registration are separate working paths.	Loading manifest plugins with registration scoped to the run and clear failure behavior.

Adding these fields does not enable incremental extraction or load the named Python modules. Keep checkpoint ownership in the scheduled job; see incremental extraction. The explicit plugin helper changes process-wide factory registrations and imports Python with the process's permissions. Neither helper provides an execution sandbox.

Step contract quarantine policy

Every DAG step contract with on_error: quarantine requires a non-null dlq with a declared destination. Pipeline parsing, direct step execution, and the published JSON Schema enforce this requirement. A preview also requires a valid policy, although it does not construct or write the quarantine connector.

contract:
  ref: {type: file, path: contracts/orders.yaml}
  validate: true
  on_error: quarantine
  dlq:
    target:
      type: file
      path: ./out/orders.dlq.jsonl

This is a breaking configuration change. Earlier versions accepted quarantine without a destination and dropped invalid records. Migrate existing manifests by declaring a DLQ, or select on_error: skip when discarding invalid records is intentional. The separate validation APIs that return invalid-record envelopes to their caller retain their behavior.

Record outcome accounting

Each DAG StepResult exposes rows_in, rows_before_contract, rows_out, and record_outcomes. rows_before_contract is the operation's output before its contract runs; it is None until that output is known. This separates filtering and join/union cardinality from contract rejection. A source that fails after yielding batches retains the known count in this field.

record_outcomes reports processing events at that contract boundary:

Field	Meaning
observed, accepted	Records encountered and retained by contract processing.
validation_evaluated, validation_passed, validation_failed	Record validation outcomes.
warned, skipped, rejected	Invalid records retained by warn, discarded by skip, or stopped by fail.
quarantined	Invalid records whose DLQ write returned successfully.
would_quarantine	Invalid records selected for quarantine during a preview; no DLQ write occurs.
quarantine_failed, processing_failed	Failed DLQ writes and other contract-processing failures.
complete	Whether processing completed; failed steps can retain partial counts.

Run-level record_outcomes sums these events across steps. A record validated at both extract and load contributes at both boundaries. These totals do not count unique source records, and differences between input and output rows do not imply quarantine: transforms can filter, joins can expand, and unions can repeat records. A replay starts fresh counters.

Existing rows_failed continues to count loader-reported failures. Use record_outcomes.validation_failed for contract validation failures and total_quarantined for acknowledged DLQ writes across extract, load, transform, join, and union steps. These reports describe observed execution; they do not establish atomic writes, exactly-once delivery, or durable operational history. A loader that raises before returning a LoadResult can have written rows without reporting them. Zero acknowledged loads in that failed run does not prove the destination is unchanged; inspect the destination before rerunning.