API reference¶

The public Python API of dlt-ops — the names exported from the top-level dlt_ops package (its __all__). Everything else is importable but internal, with no stability promise; the public-vs-internal convention and the stability contract live in versioning. Import any name directly, for example from dlt_ops import with_checkpoints, DestinationAdapter, register.

The schema-drift names (reconcile_all, reconcile_source, detect_removal, ReconcileResult, DriftFinding, AlertSink) are re-exported lazily so that import dlt_ops never pulls the reconciler's backend dependencies at import time; they are documented below under their dlt_ops.reconciler module but resolve equally as from dlt_ops import reconcile_all.

Plugin registration¶

Register a plugin against one of the entry-point axes; see plugins.

dlt_ops.register ¶

register(axis: str, name: str) -> Callable[[_T], _T]

Register the decorated object as plugin <axis>/<name>.

Runtime twin of the dlt_ops.<axis> entry-point groups — both feed the same process-wide registry, so get/names behave identically regardless of how a plugin arrived::

@dlt_ops.register("destination", "duckdb")
class DuckDBAdapter: ...

Discovery and validation¶

Scan a project for sources and run the rule framework; see discovery and validation.

dlt_ops.discover_sources ¶

discover_sources(
    project_root: Path,
) -> dict[str, SourceInfo]

Discover dlt sources and attach their imported callables.

Supports multiple sources per directory. Each source is keyed by its config_section (source name), not the directory name.

Parameters:

Name	Type	Description	Default
`project_root`	`Path`	Path to the project root (holds .dlt/config.toml and one subdirectory per pipeline)	required

Returns:

Type	Description
`dict[str, SourceInfo]`	Dict mapping source name (config_section) to a fully-introspected
`dict[str, SourceInfo]`	SourceInfo. Sources whose module could not be imported are excluded
`dict[str, SourceInfo]`	(with a warning); Phase 1 still lists them via `phase1.discover`.

Raises:

Type	Description
`ProjectConfigParseError`	.dlt/config.toml exists but is broken TOML.

dlt_ops.validate_sources ¶

validate_sources(
    project_root: Path,
    *,
    validators: list[Validator] | None = None,
    strict: bool = False,
) -> list[ValidationError]

Run the resolved rule set against discovered sources.

Discovery runs both phases: Phase 1 (AST) lists everything — including a placeholder per source module that does not parse — and Phase 2 (sandboxed import) attaches callables and records import failures / Rule 15 findings. Validators that instantiate sources see only the import-OK subset (ctx.sources); the import-health validators see the full Phase-2 output (ctx.introspected).

Rule assembly + resolution happen once per run: registry defaults overlaid by [dlt_ops.rules] decide which rules execute, and rule_exemptions findings are filtered per (source, rule) pair. Config problems — unknown rule IDs, non-bool knob values, unjustified exemptions — surface as errors in the returned list, as do rule providers that contributed no rules (:func:rule_provider_errors): a run that quietly checks less than it claims must not report success. Import-error surfacing (a module that cannot import cannot run — it fails to parse, raises, or is withheld for violating Rule 15) is infrastructure, not a rule: always on, no knob.

Every finding is returned in both modes — warnings included, so callers can render them — and is_warning carries the finding's severity for this run: strict promotes warnings to errors before returning, so a caller decides pass/fail with one question in either mode, "is any finding not a warning?".

Parameters:

Name	Type	Description	Default
`project_root`	`Path`	Path to the project root	required
`validators`	`list[Validator] \| None`	Escape hatch — run exactly these callables instead of the resolved rule set. Rule knobs and exemptions key on rule IDs and therefore do not apply to a custom list.	`None`
`strict`	`bool`	If True, promote warnings to errors (`is_warning=False`) so they fail the run.	`False`

Returns:

Type	Description
`list[ValidationError]`	Every finding, errors and warnings alike.

dlt_ops.SourceInfo ¶

Discovered dlt source with metadata.

Two-phase population:

Phase 1 (discovery.phase1.discover) fills the static fields from a pure AST scan and never imports project code. resources is then a static approximation: @dlt.resource declarations in the source's own module plus the pipeline's resource/*.py siblings (dynamic resource factories only resolve in Phase 2).
Phase 2 (discovery.phase2.introspect) enriches: attaches the imported source_fn, replaces resources with the authoritative instantiated list, and records import failures / import-safety findings.

source_fn `property` ¶

source_fn: Callable[..., Any]

The imported @dlt.source callable (Phase-2 enrichment).

Raises:

Type	Description
`RuntimeError`	the record is Phase-1-only (not introspected, or its module failed to import — see `import_error`).

is_introspected `property` ¶

is_introspected: bool

True iff Phase 2 attached the imported source callable.

config_section `property` ¶

config_section: str

Config section name (same as name).

dlt_ops.SourceConfig ¶

Config from config.toml for a source.

All custom keys are under [sources.X.dlt_ops]: - schedule: Schedule enum value - destination: per-source destination override; falls back to [dlt_ops].default_destination (see dlt_ops.config) - dataset: per-source dataset override; falls back to [dlt_ops].default_dataset - airflow_var: Airflow Variable name, surfaced by pipeline list / pipeline resources. Parsed for display only — the Airflow secret backend reads its own trigger keys straight off the raw [sources.X.dlt_ops] table, so core never acts on this value. - schema_contract_evolve_reason: opt-in for the evolve schema contract literal. Non-empty string = the source may declare {"tables": "evolve", "columns": "evolve", "data_type": "freeze"} on its @dlt.resource calls. Absence / empty string = default freeze. - injected_columns: infrastructure keys stamped per-row inside a resource rather than coming from the upstream payload. Consumed by the reconciler so it doesn't flag them as unknown drift. loaded_at is always ignored — no need to list it.

is_schema_contract_evolve `property` ¶

is_schema_contract_evolve: bool

True iff a non-empty justification is present.

Empty string is treated as absence so opt-in requires actual justification, not just presence of the key. Non-string TOML values (int/bool/list) are treated as absence rather than raising — the scanner passes the value through untouched so a hand-authored schema_contract_evolve_reason = 42 reaches here as an int.

dlt_ops.Schedule ¶

Bases: str, Enum

Valid schedule tags for dlt pipelines.

from_string `classmethod` ¶

from_string(value: str) -> Schedule

Parse schedule from string, with helpful error message.

dlt_ops.RuleSpec ¶

One registered validation rule: stable identity + validator + provenance.

rule_id is the rule's public identity: the [dlt_ops.rules] knob and rule_exemptions tables key on it, so it never renames within a major version. plugin names the provider that registered the rule ("core" for the package's own rules, the plugin's entry-point name otherwise) — shown as the rule's origin in validate --show-resolved-rules. default_on=False ships a rule opt-in.

description `property` ¶

description: str

One-line rule summary: the validator docstring's first line.

dlt_ops.Validator ¶

Bases: Protocol

Protocol for validator functions.

dlt_ops.ValidationContext ¶

Context passed to all validators.

sources holds only import-OK sources (source_fn attached) so rules that instantiate sources keep working. introspected is the full Phase-2 output — including sources whose module failed to import or violated Rule 15 — for the import-health validators.

resolved_rules is the per-run rule resolution (registry defaults overlaid by [dlt_ops.rules]) and exemptions the parsed [sources.<X>.dlt_ops.rule_exemptions] view ({source: {rule_id: reason}}). Both are populated once per validate_sources run so validators consult a single resolution instead of re-reading raw config.

rule_enabled ¶

rule_enabled(rule_id: str) -> bool

Resolved on/off state for a rule; unresolved (hand-built context) = on.

is_exempt ¶

is_exempt(source_name: str, rule_id: str) -> bool

True iff the source carries a justified exemption for the rule.

dlt_ops.ValidationError ¶

Validation error for a source.

Checkpoints¶

Persist pagination progress to the destination and resume mid-run; see checkpoints.

dlt_ops.with_checkpoints ¶

with_checkpoints(
    cursor_field: str,
    frequency: int = 10,
    offset_seconds: int = 1,
    checkpoint_table: str = DEFAULT_CHECKPOINT_TABLE,
    cleanup_days: int = 7,
    value_parser: Callable[[str], Any] | None = None,
) -> Callable

Decorator to add transparent checkpointing to dlt resources.

Saves progress to destination during pagination and resumes from last checkpoint on failure. Works with any incremental cursor field.

Concurrent Run Isolation: Automatically isolates checkpoints for different intervals (e.g., hourly vs backfill) using the incremental's initial_value. Runs with different initial_value get separate checkpoint namespaces. Scheduler-agnostic (works locally and in Airflow).

Decorator order: @with_checkpoints must sit UNDER @dlt.resource (closest to the generator function). Applied on top of @dlt.resource it would replace the DltResource with a plain generator function, silently dropping the resource's name, write disposition, and hints — that order raises TypeError at decoration time.

Parameters:

Name	Type	Description	Default
`cursor_field`	`str`	Name of incremental field (e.g., "last_updated_utc")	required
`frequency`	`int`	Checkpoint every N pages (default: 10)	`10`
`offset_seconds`	`int`	Safety overlap on resume in seconds (default: 1)	`1`
`checkpoint_table`	`str`	Name of checkpoint table (default: _dlt_custom_checkpoints)	`DEFAULT_CHECKPOINT_TABLE`
`cleanup_days`	`int`	Days to keep completed checkpoints (default: 7)	`7`
`value_parser`	`Callable[[str], Any] \| None`	Optional function to parse checkpoint string back to original type. If not provided, uses default parser for datetime/timestamp fields.	`None`

Raises:

Type	Description
`ValueError`	If frequency <= 0 or cleanup_days < 0
`TypeError`	If applied on top of `@dlt.resource` instead of under it

Usage

@dlt.resource(...) @with_checkpoints(cursor_field="last_updated_utc") def my_resource(last_updated_utc=dlt.sources.incremental(...)): for page in paginate(...): yield page

Example with custom parser

@dlt.resource(...) @with_checkpoints( cursor_field="cursor_id", value_parser=lambda s: int(s) ) def my_resource(cursor_id=dlt.sources.incremental(...)): for page in paginate(...): yield page

dlt_ops.list_checkpoints ¶

list_checkpoints(
    pipeline_name: str | None = None,
    checkpoint_table: str = DEFAULT_CHECKPOINT_TABLE,
    pipeline=None,
) -> list[dict[str, Any]]

List all checkpoints for debugging.

Parameters:

Name	Type	Description	Default
`pipeline_name`	`str \| None`	Pipeline name to filter (required if pipeline not provided)	`None`
`checkpoint_table`	`str`	Name of checkpoint table (default: _dlt_custom_checkpoints)	`DEFAULT_CHECKPOINT_TABLE`
`pipeline`		Optional pipeline object (avoids dlt.attach() call)	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	List of checkpoint records as dicts keyed by column name, newest first

dlt_ops.cleanup_checkpoints ¶

cleanup_checkpoints(
    pipeline_name: str | None = None,
    resource_name: str | None = None,
    checkpoint_table: str = DEFAULT_CHECKPOINT_TABLE,
    pipeline=None,
    *,
    include_active: bool = False,
)

Delete completed checkpoint rows for a pipeline or resource.

Retention housekeeping by default: only rows a successful run already marked completed are deleted. active rows are live resume state — the row a crashed or still-running extract resumes from — so deleting one restarts that resource at its window start and silently re-extracts everything the previous run already loaded. Active rows found in scope are kept and reported at WARNING.

Pass include_active=True for the destructive form: every checkpoint row in scope regardless of status. That is the surgical escape hatch — abandoning a poisoned resume point, or clearing state for a pipeline dropped outside dlt-ops — and the next run of an affected resource restarts its window from the beginning.

Parameters:

Name	Type	Description	Default
`pipeline_name`	`str \| None`	Pipeline name to clean (required if pipeline not provided)	`None`
`resource_name`	`str \| None`	Specific resource to clean (None = all resources in pipeline)	`None`
`checkpoint_table`	`str`	Name of checkpoint table (default: _dlt_custom_checkpoints)	`DEFAULT_CHECKPOINT_TABLE`
`pipeline`		Optional pipeline object (avoids dlt.attach() call)	`None`
`include_active`	`bool`	Also delete `active` rows, destroying resume state (default: False)	`False`

Examples:

Prune completed checkpoints for a pipeline¶

cleanup_checkpoints(pipeline_name="my_pipeline")

Prune completed checkpoints for one resource¶

cleanup_checkpoints(pipeline_name="my_pipeline", resource_name="companies_bulk")

Wipe every row, resume state included¶

cleanup_checkpoints(pipeline_name="my_pipeline", include_active=True)

Destination adapters¶

The contract a destination implements to reach full tier; see destinations and capability tiers and the adapter guide.

dlt_ops.DestinationAdapter ¶

Bases: Protocol

Everything checkpoint/cleanup/reconciler code may know about a destination.

Implementations register as dlt_ops.destination entry points (or via dlt_ops.register("destination", name)) and are resolved by name through the plugin registry.

name `instance-attribute` ¶

name: str

Registry key: the destination's dlt engine name — "bigquery", "duckdb", ...

Not the SQL dialect. Which dialect an adapter transpiles into is its own business and stays out of this port; engines can share a dialect, and an engine name is not always one any transpiler knows.

placeholder_style `instance-attribute` ¶

placeholder_style: str

Native positional placeholder token of the destination's dlt sql_client.

Deliberately an open string: the set of placeholder styles a DB-API driver can use is not the package's to close, so an adapter may declare any token its client binds against. Informational for diagnostics; conversion from canonical ? happens inside execute_sql / execute_query, never in caller code.

supports_if_exists `instance-attribute` ¶

supports_if_exists: bool

CREATE TABLE IF NOT EXISTS / DROP TABLE IF EXISTS are valid DDL.

Consumers: checkpoint table DDL and drop_table_if_exists, which falls back to probe-then-drop when False.

supports_create_schema_if_not_exists `instance-attribute` ¶

supports_create_schema_if_not_exists: bool

The adapter may create the schema/dataset via ensure_schema.

Consumer: CheckpointManager setup, which needs the schema to exist before the checkpoint DDL runs. False when schema creation is not the adapter's to make — it can carry placement or access-control decisions owned by dlt or the surrounding infrastructure — and ensure_schema is then a no-op.

timestamp_now_sql `instance-attribute` ¶

timestamp_now_sql: str

Canonical-dialect fragment for "now" that survives this adapter's transpile.

Consumers: checkpoint _mark_completed, cleanup retention.

timestamp_sub_days_sql `instance-attribute` ¶

timestamp_sub_days_sql: Callable[[int], str]

days -> canonical-dialect fragment for "now minus N days".

Same consumers as timestamp_now_sql; kept per-adapter because interval arithmetic is the idiom sqlglot most often mistranslates.

render_identifier ¶

render_identifier(ident: str) -> str

Validate ident against this destination's identifier grammar and quote it.

Returns the canonically-quoted form for embedding in canonical SQL — native quoting is applied by the transpile step inside execute_sql. Defaults to :func:render_canonical_identifier; an adapter whose destination accepts less may tighten the grammar, never loosen it. Raises ValueError for names outside the grammar.

render_table_ref ¶

render_table_ref(dataset: str, table: str) -> str

dataset.table reference in canonical form; both parts validated.

execute_sql ¶

execute_sql(
    client: Any, canonical_sql: str, *params: Any
) -> None

Transpile canonical SQL, bind *params, execute via client.

Params bind natively by default. An adapter whose driver cannot type a given bound value — a rejected bound None is the usual case — may inline it as a literal instead: the None becomes a NULL literal in the SQL, never injected text (the value still goes through the sqlglot AST, not string interpolation).

execute_query ¶

execute_query(
    client: Any, canonical_sql: str, *params: Any
) -> Cursor

Like execute_sql but returns a Cursor over the result rows.

ensure_schema ¶

ensure_schema(client: Any, dataset: str) -> None

Create the schema/dataset when the destination supports and needs it.

No-op when supports_create_schema_if_not_exists is False, so callers may invoke it unconditionally.

fetch_columns ¶

fetch_columns(
    client: Any, dataset: str, table: str
) -> list[ColumnInfo] | None

Columns of dataset.table via one canonical information_schema.columns SELECT.

Returns None when the table (or its dataset) is absent. Consumer: the reconciler's drift detection.

Capability-derived adapters¶

These names live on the dlt_ops.destinations subpackage rather than the top level — import them as from dlt_ops.destinations import register_derived_adapter. They build an adapter from the facts dlt already publishes about a destination, for engines nobody hand-wrote one for. Derivation shapes the SQL for the declared dialect; it does not verify that anyone has run it there, so registering is opt-in and logs a warning. Derived is not the same as tested is the framing to read before relying on one.

dlt_ops.destinations.derived.register_derived_adapter ¶

register_derived_adapter(
    engine: str, *, registry: Any = _default_registry
) -> DerivedAdapter

Register a capability-derived adapter for engine on the destination axis.

The runtime twin of shipping an adapter under the dlt_ops.destination entry-point group: after this call has_adapter answers True for engine and the adapter-gated features run against it. Registering the same engine twice is harmless.

Raises:

Type	Description
`UnderivableDestinationError`	see :func:`derived_adapter`.

dlt_ops.destinations.derived.derived_adapter ¶

derived_adapter(engine: str) -> DerivedAdapter

Build (without registering) a capability-derived adapter for engine.

Raises:

Type	Description
`UnderivableDestinationError`	dlt publishes no basis for one — an unresolvable destination, no declared dialect, or a dialect whose placeholder syntax sqlglot cannot render as a token.

dlt_ops.destinations._capabilities.derivable_destinations ¶

derivable_destinations() -> tuple[str, ...]

Engine names dlt publishes enough capabilities to derive an adapter for.

Diagnostic surface: what a user could opt into, not what is supported. Derivation says the SQL will be shaped for the right dialect; it says nothing about whether anyone has run it there.

dlt_ops.destinations.is_capability_derived ¶

is_capability_derived(adapter: DestinationAdapter) -> bool

Whether adapter was derived from dlt's capabilities rather than written and tested.

Takes a resolved adapter rather than a name so it never triggers a load of its own. Reporting surfaces use it to qualify full tier honestly: derived is usable, but it says the SQL was shaped for the declared dialect, not that anyone ran it there.

dlt_ops.destinations._capabilities.UnderivableDestinationError ¶

Bases: LookupError

Capabilities for this destination give no basis for an adapter.

Raised instead of falling back to a plausible-looking dialect: guessing one would produce SQL that parses and silently means something else.

dlt_ops.destinations.CI_VERIFIED_DESTINATIONS `module-attribute` ¶

CI_VERIFIED_DESTINATIONS: tuple[str, ...] = (
    "duckdb",
    "postgres",
)

Destinations whose adapter is exercised against a live instance in CI.

The honest floor under every "supported" claim. A destination absent here may still work — bigquery has a live lane that is credential-gated and therefore non-blocking, and a capability-derived adapter may be perfectly correct — but nothing has proven it on every commit. Kept separate from the registry on purpose: registration answers "can this run", this answers "has anyone checked", and conflating them is how a package starts overclaiming.

Assertions¶

Pre-load data-quality gates and the error they raise; see assertions.

dlt_ops.AssertionType ¶

Bases: Protocol

One assertion type: a named accumulator with a static config check.

Implementations register under the dlt_ops.assertion entry-point group (or the runtime twin dlt_ops.register("assertion", name)); entry points conventionally register the class, instantiated with no arguments.

name `instance-attribute` ¶

name: str

Registry name == TOML key. Frozen once released (config compatibility).

row_scoped `instance-attribute` ¶

row_scoped: bool

True → observe emits per-row verdicts (quarantine-compatible). False → batch-scope only; on_failure = "quarantine" is a config error.

check_config ¶

check_config(
    params: Mapping[str, Any], ctx: AssertionContext
) -> list[str]

Statically checkable config errors (Tier 1). Empty list = valid.

This is the enforced-opinions declaration mechanism: everything a type CAN check without data (param types/domains, column references against ctx.declared_columns) it MUST check here. Never touches data, network, or clients.

start ¶

start(params: Mapping[str, Any]) -> Any

Fresh accumulator state for one (resource, run) batch.

observe ¶

observe(
    state: Any,
    row: Mapping[str, Any],
    params: Mapping[str, Any],
) -> str | None

Per-row verdict: None = pass; message = THIS ROW fails.

Batch-scope types accumulate and return None.

finalize ¶

finalize(
    state: Any, params: Mapping[str, Any]
) -> str | None

Batch verdict after the last row: None = pass; message = batch fails.

dlt_ops.AssertionContext ¶

Static facts check_config may validate against. No data, no clients.

declared_columns `instance-attribute` ¶

declared_columns: tuple[str, ...] | None

Column names from the resource's columns= Pydantic model; None when the model is unresolvable (e.g. pydantic_columns_required exempted). Types MUST skip column-existence checks when this is None.

dlt_ops.AssertionFailedError ¶

Bases: Exception

A configured assertion failed with on_failure = "fail".

Raised by the engine (row verdicts inside the extract gate, batch verdicts at finalize); the run aborts, nothing loads, and the runner drops the pending extracted package so the rejected batch cannot auto-load on the next run.

Secret backends¶

The contract a secret backend implements; see plugins.

dlt_ops.SecretBackend ¶

Bases: Protocol

Everything the runtime may know about a secret backend.

Raising :class:SecretNotFoundError from get on a missing secret is part of the contract. The optional secret_requests selection hook is documented in the module docstring — it is not a required member.

name `instance-attribute` ¶

name: str

Registry name: "secrets_toml", "airflow", ...

get ¶

get(key: str) -> str

Fetch one secret by backend-native reference.

Raises:

Type	Description
`SecretNotFoundError`	the referenced secret does not exist.

Alert sinks¶

The contract a drift-alert sink implements; see reconciler.

dlt_ops.reconciler.AlertSink ¶

Bases: Protocol

Destination for reconciler events: drift findings + internal errors.

The contract of the alert_sink plugin axis (entry-point group dlt_ops.alert_sink): emit_drift(finding) / emit_error(exc, *, source_name, resource_name, context) / flush(timeout). Sinks are selected per project via [dlt_ops] alert_sinks = ["logging", ...] (default: the core logging sink) and constructed with their [dlt_ops.alert_sink.<name>] options as keyword arguments; all configured sinks receive every event. --dry-run suppresses all emission.

flush(timeout) is called by every public reconciler entry point on the way out — sinks with a background transport queue must drain it there (deterministic, bounded by timeout) because a short-lived CLI or orchestrator task may exit immediately after.

Schema-drift reconciliation¶

Diff declared models against the live destination schema; see reconciler.

dlt_ops.reconciler.reconcile_all ¶

reconcile_all(
    *,
    dry_run: bool = False,
    fetcher: "SchemaFetcher | None" = None,
    runner: "QueryRunner | None" = None,
    dataset: str | None = None,
    project_root: "Any | None" = None,
    project_config: ProjectConfig | None = None,
    sink: "AlertSink | None" = None,
) -> list[ReconcileResult]

Iterate every discovered source and reconcile each.

Each source resolves — and reconciles against — its own destination + dataset from the config chain, so a multi-destination project sweeps per destination with no cross-destination auth. A failure on one source never blocks the others — each source produces its own ReconcileResult (potentially with error=<message>). One sink flush at the end covers the whole sweep, not one per source.

dlt_ops.reconciler.reconcile_source ¶

reconcile_source(
    source_name: str,
    *,
    dry_run: bool = False,
    fetcher: "SchemaFetcher | None" = None,
    runner: "QueryRunner | None" = None,
    dataset: str | None = None,
    sources: dict[str, SourceInfo] | None = None,
    project_root: "Any | None" = None,
    project_config: ProjectConfig | None = None,
    sink: "AlertSink | None" = None,
) -> ReconcileResult

Run additive-drift detection against one discovered source.

dry_run=True suppresses all sink emission (both drift and reconciler-error paths); the returned ReconcileResult still carries the findings tuple so local verification can inspect them.

Injection points (fetcher, runner, dataset, sources, project_root, project_config, sink) exist for tests — prod callers pass nothing and everything resolves from the project config chain + discover_sources + the DestinationAdapter-backed defaults. project_root lets the CLI's project-root option reach discovery without an env-var round-trip.

dlt_ops.reconciler.detect_removal ¶

detect_removal(
    source_name: str,
    *,
    dry_run: bool = False,
    runner: "QueryRunner | None" = None,
    dataset: str | None = None,
    sources: dict[str, SourceInfo] | None = None,
    project_root: "Any | None" = None,
    project_config: ProjectConfig | None = None,
    sink: "AlertSink | None" = None,
    baseline_threshold: float = DEFAULT_BASELINE_THRESHOLD,
    recent_threshold: float = DEFAULT_RECENT_THRESHOLD,
    recent_window_hours: int = DEFAULT_RECENT_WINDOW_HOURS,
    baseline_window_days: int = DEFAULT_BASELINE_WINDOW_DAYS,
) -> ReconcileResult

Run removal-drift detection against one discovered source.

Contract matches additive.reconcile_source — dry_run=True suppresses all sink emission; source-level failures land in result.error; per-resource failures are trapped and reported through the sink's error path but don't stop the sweep. When no load_timestamp_column is configured, detection is skipped and the result carries a warning instead.

Callers can widen or tighten the windows and thresholds without patching this module — every knob is a keyword-only parameter with the canonical default exported as DEFAULT_*. Prod callers pass runner=None and the reconciler opens the source's own resolved destination through the DestinationAdapter boundary.

dlt_ops.reconciler.ReconcileResult ¶

Outcome of one reconcile_source / detect_removal call.

findings is the tuple of per-resource findings surfaced during the run. duration_ms is wall-clock time including destination round-trips and alert emission (or 0 when the runner skipped everything on an early failure). error is populated only when the top-level source-scan itself failed — per-resource failures are wrapped and reported through the alert sink's error path, and do not surface here so an orchestrated sweep can succeed with partial coverage. warnings carries non-fatal degradations (e.g. removal detection skipped because no load-timestamp column is configured).

dlt_ops.reconciler.DriftFinding ¶

One drifted resource surfaced by the reconciler.

kind disambiguates additive (the destination has a column the model doesn't) from removal (a model column's non-null coverage collapsed to ~0). columns is the full drifted set — additive detection emits one finding per drifted RESOURCE with all its drifted columns, not one finding per column, so downstream alert issues collapse cleanly.

sample_values is a mapping keyed by column name → up to 5 recent values from the resource's destination table. Empty for removal findings (there are no recent non-null values to sample by definition). inferred_types is the destination-reported data_type string per column, positionally aligned to columns.

reproduce_sql is a copy-pasteable canonical-dialect SELECT computed at detection time (where the resolved dataset and load-timestamp column are known) so alert sinks stay pure serializers with no SQL knowledge.

Pydantic model helpers¶

Derive column facts from your declared Pydantic models; see add a source.

dlt_ops.drop_unknown_nulls ¶

drop_unknown_nulls(
    model: type[BaseModel],
) -> Callable[[dict[str, Any]], dict[str, Any]]

Strip None-valued keys not in the model's known-fields set.

Prevents the seen-null-first freeze trap: dlt would otherwise register an incomplete column on the first null observation; when the first non-null arrives, dlt tries to complete-with-type and data_type: "freeze" rejects it. Stripping unknown nulls means the column is born typed on its first real value.

dlt_ops.extract_model_column_names ¶

extract_model_column_names(
    model: type[BaseModel],
) -> set[str]

Return the set of keys a raw payload can carry for a known model field.

Includes attribute names + aliases + validation_aliases so callers do not need to know Pydantic-v2 alias forms. Shared by drop_unknown_nulls (below), reconciler/additive.py (known-set for live_columns - known diff), and reconciler/removal.py (columns-to-check-for-coverage-drop).

API reference¶

Plugin registration¶

dlt_ops.register ¶

Discovery and validation¶

dlt_ops.discover_sources ¶

dlt_ops.validate_sources ¶

dlt_ops.SourceInfo ¶

source_fn property ¶

is_introspected property ¶

config_section property ¶

dlt_ops.SourceConfig ¶

is_schema_contract_evolve property ¶

dlt_ops.Schedule ¶

from_string classmethod ¶

dlt_ops.RuleSpec ¶

description property ¶

dlt_ops.Validator ¶

dlt_ops.ValidationContext ¶

rule_enabled ¶

is_exempt ¶

dlt_ops.ValidationError ¶

Checkpoints¶

dlt_ops.with_checkpoints ¶

dlt_ops.list_checkpoints ¶

dlt_ops.cleanup_checkpoints ¶

Prune completed checkpoints for a pipeline¶

Prune completed checkpoints for one resource¶

Wipe every row, resume state included¶

Destination adapters¶

dlt_ops.DestinationAdapter ¶

name instance-attribute ¶

placeholder_style instance-attribute ¶

supports_if_exists instance-attribute ¶

supports_create_schema_if_not_exists instance-attribute ¶

timestamp_now_sql instance-attribute ¶

timestamp_sub_days_sql instance-attribute ¶

render_identifier ¶

render_table_ref ¶

execute_sql ¶

execute_query ¶

ensure_schema ¶

fetch_columns ¶

Capability-derived adapters¶

dlt_ops.destinations.derived.register_derived_adapter ¶

dlt_ops.destinations.derived.derived_adapter ¶

dlt_ops.destinations._capabilities.derivable_destinations ¶

dlt_ops.destinations.is_capability_derived ¶

dlt_ops.destinations._capabilities.UnderivableDestinationError ¶

dlt_ops.destinations.CI_VERIFIED_DESTINATIONS module-attribute ¶

Assertions¶

dlt_ops.AssertionType ¶

name instance-attribute ¶

row_scoped instance-attribute ¶

check_config ¶

start ¶

observe ¶

finalize ¶

dlt_ops.AssertionContext ¶

declared_columns instance-attribute ¶

dlt_ops.AssertionFailedError ¶

Secret backends¶

dlt_ops.SecretBackend ¶

name instance-attribute ¶

get ¶

Alert sinks¶

dlt_ops.reconciler.AlertSink ¶

Schema-drift reconciliation¶

dlt_ops.reconciler.reconcile_all ¶

dlt_ops.reconciler.reconcile_source ¶

dlt_ops.reconciler.detect_removal ¶

dlt_ops.reconciler.ReconcileResult ¶

dlt_ops.reconciler.DriftFinding ¶

Pydantic model helpers¶

dlt_ops.drop_unknown_nulls ¶

dlt_ops.extract_model_column_names ¶

source_fn `property` ¶

is_introspected `property` ¶

config_section `property` ¶

is_schema_contract_evolve `property` ¶

from_string `classmethod` ¶

description `property` ¶

name `instance-attribute` ¶

placeholder_style `instance-attribute` ¶

supports_if_exists `instance-attribute` ¶

supports_create_schema_if_not_exists `instance-attribute` ¶

timestamp_now_sql `instance-attribute` ¶

timestamp_sub_days_sql `instance-attribute` ¶

dlt_ops.destinations.CI_VERIFIED_DESTINATIONS `module-attribute` ¶

name `instance-attribute` ¶

row_scoped `instance-attribute` ¶

declared_columns `instance-attribute` ¶

name `instance-attribute` ¶