API — Authoring

Generated from package source. Hub: Python API Reference.

Authoring

Portable authoring and compilers

etlantic.transform, @Transformation.portable, symbolic DataFrame and Column objects, and functions as F normalize to published DTCS 3.0 dtcs.transform-plan/2 models. Official compilers ship in optional packages. See Portable Transformations and Portable Transform Compiler.

Core behavioral contracts

The generated signatures below are supplemented by these current guarantees:

API	Returns	Important failures / side effects
Transformation.step(**bindings)	A symbolic Step; no user code runs	Unknown bindings raise ModelDefinitionError; required ports are validated before execution
Transformation.implementation(engine)	A decorator returning the original callable	Registration replaces the implementation for the same class/engine in the current process
Transformation.portable	Decorator registering a symbolic definition	Authoring errors raise ModelDefinitionError (PMXFORM*) at registration; does not execute
Transformation.to_transform_plan()	Deep-copied dtcs.transform-plan/2 dict	Raises ModelDefinitionError if no portable definition is registered
Transformation.portable_fingerprint()	Hex fingerprint string	Same failure mode as to_transform_plan
Pipeline.validate(...)	ValidationReport	Does not execute transformation implementations; production empty allowlist fails closed (PMPLUG401)
Pipeline.plan(...)	Immutable, secret-free PipelinePlan	Missing plugins, bindings, trust, or capabilities produce planning/validation failure
Pipeline.run(...)	PipelineRunReport	Executes in-process; storage and plugin side effects follow the resolved plan
Pipeline.arun(...)	Awaitable PipelineRunReport	Uses the same validation and planning path as run
Pipeline.to_mermaid()	Mermaid flowchart string	Builds the logical graph but does not plan or execute

Minimal validation pattern:

report = CustomerPipeline.validate(profile="development")
report.raise_for_errors()
plan = CustomerPipeline.plan(profile="development")

Minimal execution pattern:

runtime = PipelineRuntime()
runtime.memory.seed("customer_source", records)
run_report = CustomerPipeline.run(
    profile="development",
    runtime=runtime,
)
if run_report.status.value != "succeeded":
    raise RuntimeError(run_report.to_text())

PipelineRuntime is application-owned. A new Python or CLI process receives a new process-local memory store and report store unless durable providers are configured.

Data contracts

etlantic.contracts

Data-contract integration boundary for ContractModel.

Data is ETLantic's thin public facade over ContractModel (DD-010A). ContractModel retains authority for data-contract semantics and ODCS. DataContractModel remains as a deprecated compatibility alias.

is_data_contract_type(obj)

Return True when obj is a ContractModel-compatible data-contract class.

resolve_contract_type(annotation)

Extract a data-contract class from a type annotation when possible.

Returns None when the annotation is not a concrete ContractModel subclass.

load_data_contract(path, *, root=None, class_name=None)

Load an ODCS YAML artifact into a :class:Data (ContractModel) subclass.

Parameters:

Name	Type	Description	Default
path	str | Path	ODCS YAML file path.	required
root	str | Path | None	Optional contract search root for relative imports.	None
class_name	str | None	Optional explicit generated class name.	None

Returns:

Type	Description
type[Data]	A concrete Data subclass reflecting the ODCS document.

Raises:

Type	Description
OSError	When the path cannot be read.
ValueError	When the document is not valid ODCS for ContractModel.

write_odcs(model, path, *, root=None)

Write a :class:Data class to an ODCS YAML file.

Parameters:

Name	Type	Description	Default
model	type[Data]	ContractModel-compatible class to serialize.	required
path	str | Path	Destination YAML path (created or overwritten via safe I/O when applicable).	required
root	str | Path | None	Optional contract search root for relative references.	None

Returns:

Type	Description
Path	Resolved output path.

Transformations

etlantic.transformation

Transformation contracts, steps, and implementation registration.

PortDefinition dataclass

Introspected port on a transformation class.

ImplementationRecord dataclass

Registered execution implementation for a transformation.

Step dataclass

A concrete use of a transformation inside a pipeline.

Created via Transformation.step(...). Attribute access to output port names yields :class:OutputRef values for downstream wiring.

output_refs property

Return all output references keyed by port name.

bind_name(name, *, pipeline_id=None)

Return a step with its node name and output refs bound.

as_output_ref(*, default_port='result')

Return the default or named output reference for this step.

Transformation

Base class for typed transformation contracts.

Subclasses declare :class:~etlantic.ports.Input, :class:~etlantic.ports.Output, and :class:~etlantic.ports.Parameter annotations. Execution backends are registered separately with :meth:implementation (native) or :meth:portable (symbolic DTCS plan compiled by engine plugins).

Wire instances in pipelines via :meth:step, which returns a symbolic :class:Step without running user code.

identity() classmethod

Stable transformation identity.

inputs() classmethod

Return declared input ports in definition order.

outputs() classmethod

Return declared output ports in definition order.

parameters() classmethod

Return declared parameters in definition order.

implementations() classmethod

Return registered implementations keyed by engine name.

validate_definition() classmethod

Return definition problems for this transformation class.

implementation(engine) classmethod

Register a native callable for one execution engine.

Parameters:

Name	Type	Description	Default
engine	str	Registry engine name such as "local", "polars", "pandas", "sql", or "pyspark".	required

Returns:

Type	Description
Callable[[F], F]	A decorator that records the callable and returns it unchanged.

Note

Registration is process-local. Registering the same engine again replaces the earlier callable for this transformation class.

portable(fn) classmethod

Register a portable symbolic definition for this transformation.

The callable is invoked with symbolic FrameExpr inputs and ParameterRef parameters during trusted import. It must return a FrameExpr or a mapping of declared output names to FrameExpr.

Native :meth:implementation callables remain available as escape hatches. Portable definitions emit dtcs.transform-plan/2 and do not execute data.

portable_definition() classmethod

Return the cached :class:~etlantic.transform.PortableDefinition, if any.

to_transform_plan() classmethod

Return the portable dtcs.transform-plan/2 document.

Raises:

Type	Description
ModelDefinitionError	If no portable definition is registered.

portable_fingerprint() classmethod

Return the portable plan fingerprint.

to_dtcs() classmethod

Generate a DTCS document dict for this transformation.

from_dtcs(source, *, contracts=None, root=None, class_name=None) classmethod

Load a Transformation subclass from a DTCS artifact.

step(**kwargs) classmethod

Bind this transformation into a pipeline graph without running it.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Port bindings (Input names to extracts, steps, or OutputRef values) and optional parameter values named exactly as the subclass declarations.	{}

Returns:

Type	Description
Step	A symbolic Step whose output attributes can be wired downstream.

Raises:

Type	Description
ModelDefinitionError	If a binding name is unknown or required ports are missing.

Examples:

>>> normalized = NormalizeCustomers.step(customers=raw)

This method does not execute a registered implementation.

Portable transform authoring (etlantic.transform)

etlantic.transform

Portable PySpark-inspired transformation authoring (etlantic.transform/1).

ColumnExpr dataclass

Symbolic column or scalar expression that lowers to a DTCS structured node.

eqNullSafe(other)

Null-safe equality (dtcs:null_safe_eq).

alias(name)

Attach an output field name for projection / with_fields.

cast(data_type)

Strict conversion requiring the conversion profile when not try_cast.

over(window)

Attach a window specification for analytic functions.

ParameterRef dataclass

Symbolic reference to a transformation Parameter port.

CompiledTransform dataclass

In-memory compiled artifact (never serialize backend objects into plans).

PortableTransformCompiler

Bases: Protocol

Plugin protocol for analyzing, compiling, and executing portable IR.

TransformCapabilities dataclass

Advertised compiler capabilities over DTCS profiles and operators.

TransformCompileContext dataclass

Caller identity for compile() (no data access).

TransformCompilerInfo dataclass

Installed transform compiler metadata.

TransformExecutionContext dataclass

Runtime identity for execute().

TransformOutputBundle dataclass

Normalized compiler execution outputs.

TransformPlanningContext dataclass

Caller identity for analyze() (no data access).

TransformSupportFinding dataclass

One unsupported or conditional requirement from analyze().

TransformSupportReport dataclass

Deterministic support analysis for a transformation plan.

FrameExpr dataclass

Symbolic relation built from an input port or prior actions.

GroupedData dataclass

Result of groupBy awaiting aggregation.

PortableDefinition dataclass

Built portable transformation plan and authoring metadata.

TransformBudgets dataclass

Security and resource budgets for portable definition building.

Window

Factory for window specifications.

WindowSpec dataclass

Immutable window specification.

input_frame(name, *, schema_fields=None)

Create a FrameExpr bound to a transformation input port.

build_portable_definition(*, transformation, produced, budgets=DEFAULT_BUDGETS)

Validate outputs, export dtcs.transform-plan/2, and fingerprint.

invoke_portable(transformation, fn, *, budgets=DEFAULT_BUDGETS)

Call a portable authoring function with symbolic bindings and build IR.

lambda_(*parameters, body=None)

Build a bounded DTCS lambda Expression node.

Prefer lambda_("element", body=lambda element: element > 0) (callable form binds parameters with scope: "lambda"). A pre-built ColumnExpr body is accepted only when it already uses lambda-scoped fieldRefs.

transform(collection, fn)

Higher-order transform over a collection using a lambda ColumnExpr.

Symbolic only: FrameExpr / ColumnExpr trees lower to DTCS plans. They are not Polars/Pandas/Spark objects. Polars and PySpark relational compilation shipped in 0.13; eager Pandas relational compilation shipped in 0.14.

Portable transform compiler protocol (etlantic.transform.compiler)

etlantic.transform.compiler

Portable transform compiler protocol (etlantic.transform-compiler/1).

TransformCapabilities dataclass

Advertised compiler capabilities over DTCS profiles and operators.

TransformCompilerInfo dataclass

Installed transform compiler metadata.

TransformSupportFinding dataclass

One unsupported or conditional requirement from analyze().

TransformSupportReport dataclass

Deterministic support analysis for a transformation plan.

TransformPlanningContext dataclass

Caller identity for analyze() (no data access).

TransformCompileContext dataclass

Caller identity for compile() (no data access).

TransformExecutionContext dataclass

Runtime identity for execute().

CompiledTransform dataclass

In-memory compiled artifact (never serialize backend objects into plans).

TransformOutputBundle dataclass

Normalized compiler execution outputs.

PortableTransformCompiler

Bases: Protocol

Plugin protocol for analyzing, compiling, and executing portable IR.

Discovery helpers:

etlantic.transform.discovery

Entry-point discovery for portable transform compilers.

discover_transform_compilers()

Discover compilers (open allowlist) with authorize-before-load.

Parameterless for monkeypatch compatibility in unit tests. Profile-aware discovery uses :func:discover_transform_compilers_for_profile.

register_discovered_compilers(registry, *, compilers=None, profile=None)

Register discovered transform compilers into a planning registry.

load_transform_compiler(engine, *, profile=None)

Return a discovered compiler for engine, or None.

compiler_registry_snapshot(*, profile=None)

Return serializable descriptors for discovered compilers.

discover_transform_compilers_for_profile(profile)

Discover compilers applying profile.plugin_allowlist before load.

Honors monkeypatches of :func:discover_transform_compilers (tests).

Optional package factories (install etlantic-polars): etlantic_polars.create_plugin, etlantic_polars.create_transform_compiler, etlantic_polars.PolarsTransformCompiler.

Pipelines

etlantic.pipeline

Pipeline authoring: Extract, Load, Pipeline, and subpipelines.

Extract

A typed logical entry boundary that introduces data into a pipeline.

Constructing an Extract never reads data. Profiles resolve the logical asset name to an environment-specific provider at plan/runtime time.

result property

Default output reference for this extract.

as_output_ref(*, default_port='result')

Return an OutputRef for this extract's produced dataset.

bind(name, *, pipeline_id=None)

Return an extract bound to a node name within a pipeline.

Load

A typed logical publication boundary that receives data from a pipeline.

Constructing a Load never writes data. Profiles resolve the logical asset name to an environment-specific provider at plan/runtime time.

bind(name, *, pipeline_id=None)

Return a load bound to a node name within a pipeline.

SubpipelineInstance dataclass

An embedded child pipeline with parent-side bindings.

bind_name(name, *, pipeline_id=None)

Bind this subpipeline instance to a parent node name.

Pipeline

Declarative typed pipeline graph.

Subclasses declare Extract, transformation Step, Load, and optional subpipeline members. Importing a pipeline does not execute it.

identity() classmethod

Stable pipeline identity.

build_graph() classmethod

Build (and cache) the immutable logical graph for this pipeline.

inspect() classmethod

Return the read-only logical graph for this pipeline.

validate(*, profile=None, policy=None, context=None) classmethod

Validate the complete graph without executing transformation code.

Parameters:

Name	Type	Description	Default
profile	str | Any	Built-in profile name or explicit Profile.	None
policy	str | Any	Validation policy name or object.	None
context	Any	Optional planning context with registries and capabilities.	None

Returns:

Type	Description
Any	A ValidationReport containing phase results and diagnostics.

Examples:

>>> report = CustomerPipeline.validate(profile="development")
>>> report.raise_for_errors()

plan(profile=None, *, context=None, selection=None) classmethod

Resolve an immutable, secret-free execution plan.

Parameters:

Name	Type	Description	Default
profile	str | Any	Built-in profile name or explicit Profile.	None
context	Any	Optional planning context with bindings and plugins.	None
selection	dict[str, Any] | None	Optional partial-run selection mapping.	None

Returns:

Type	Description
Any	A deterministic PipelinePlan. Planning does not execute user
Any	transformation code or resolve secret values.

explain_plan(profile=None, *, context=None, selection=None) classmethod

Return a structured explanation of the planned pipeline.

Parameters:

Name	Type	Description	Default
profile	str | Any	Built-in profile name or explicit Profile.	None
context	Any	Optional planning context with bindings and plugins.	None
selection	dict[str, Any] | None	Optional partial-run selection mapping.	None

Returns:

Type	Description
dict[str, Any]	A JSON-serializable explanation dict (regions, bindings,
dict[str, Any]	interchange boundaries, fingerprints). Does not execute user code.

Examples:

>>> explanation = CustomerPipeline.explain_plan(profile="development")
>>> "fingerprint" in explanation or "plan" in explanation
True

run(profile='development', *, request=None, runtime=None, context=None, workspace=None) classmethod

Validate, plan, and execute this pipeline in the current process.

Parameters:

Name	Type	Description	Default
profile	str | Any	Built-in profile name or explicit Profile.	'development'
request	Any	Optional run selection, intent, and policy request.	None
runtime	Any	Application-owned PipelineRuntime. A new runtime is created when omitted.	None
context	Any	Optional planning context.	None
workspace	str | Any	Optional durable workspace root.	None

Returns:

Type	Description
Any	A structured PipelineRunReport.

Raises:

Type	Description
PipelineValidationError	If validation fails before execution.
PipelineExecutionError	If execution cannot produce a run report.

Storage writes and plugin calls follow the resolved plan. Process-local memory and report stores do not survive a new CLI or Python process.

arun(profile='development', *, request=None, runtime=None, context=None, workspace=None) async classmethod

Validate, plan, and execute this pipeline locally (async).

Parameters:

Name	Type	Description	Default
profile	str | Any	Built-in profile name or explicit Profile.	'development'
request	Any	Optional run selection, intent, and policy request.	None
runtime	Any	Application-owned PipelineRuntime.	None
context	Any	Optional planning context.	None
workspace	str | Any	Optional durable workspace root.	None

Returns:

Type	Description
Any	A structured PipelineRunReport.

Raises:

Type	Description
PipelineValidationError	If validation fails before execution.
PipelineExecutionError	If execution cannot produce a run report.

Examples:

>>> import anyio
>>> report = anyio.run(CustomerPipeline.arun, "development")

debug(profile='development', *, runtime=None, context=None) classmethod

Open a stateful local debug session.

to_mermaid() classmethod

Generate a Mermaid flowchart from the logical graph.

to_dpcs() classmethod

Generate a DPCS document dict for this pipeline.

from_dpcs(source, *, registry=None, root=None, class_name=None) classmethod

Load a Pipeline subclass from a DPCS artifact.

generate_contracts() classmethod

Discover and build an in-memory contract bundle.

write_contracts(directory) classmethod

Generate and write ODCS/DTCS/DPCS artifacts under directory.

subpipeline(**bindings) classmethod

Embed this pipeline as a reusable subpipeline in a parent.

Ports and references

etlantic.ports

Port annotation markers: Input, Output, and Parameter.

Input

Bases: Generic[T]

Declares a typed input port consumed under contract T.

Use on :class:~etlantic.transformation.Transformation subclasses::

class Normalize(Transformation):
    rows: Input[RawRow]
    result: Output[CuratedRow]

Parameters:

Name	Type	Description	Default
contract_type	type[T] | None	:class:~etlantic.contracts.Data subclass (via subscript Input[MyData] or constructor argument).	None

Output

Bases: Generic[T]

Declares data produced under contract T.

role distinguishes primary valid outputs from invalid/side channels: "valid" (default), "invalid", or "side".

as_invalid()

Return a copy marked as an invalid-output channel.

as_side()

Return a copy marked as a side-output channel.

Parameter

Bases: Generic[T]

Declares typed configuration that is not a graph edge.

Parameters are bound at :meth:~etlantic.transformation.Transformation.step time and appear in plans separately from input/output ports.

Example::

class ThresholdFilter(Transformation):
    rows: Input[Row]
    min_score: Parameter[int]
    result: Output[Row]

with_default(value)

Return a copy of this parameter marker with a default value.

unwrap_port_marker(annotation)

Classify an annotation as Input/Output/Parameter and return (kind, contract).

Returns (None, None) when the annotation is not a port marker. The first element is the marker class (Input, Output, or Parameter).

etlantic.refs

Typed output references for pipeline wiring.

OutputRef dataclass

Bases: Generic[T]

A typed reference to a concrete producer port in a pipeline graph.

Records the producing node, named output port, and output contract. Does not hold runtime dataframe or storage objects during authoring.

identity property

Stable identity for this output reference.

bind_pipeline(pipeline_id)

Return a copy bound to a pipeline identity.

with_node(node_name, *, pipeline_id=None)

Return a copy with a concrete node name.

as_output_ref(value, *, default_port='result')

Normalize an Extract, Step attribute, or OutputRef into an OutputRef.