Steps — Base

Step protocol: the contract every pipeline step must follow.

Step

Base class for all pipeline steps.

Every step declares what it reads from and writes to the Results object. Pipeline uses these declarations for validation before running.

Subclasses must: 1. Set reads and writes class attributes. 2. Implement __call__(results) -> results.

Example:

• class MyStep(Step):
• reads = [“dataset”]
• writes = [“record”] def call(self, results: Results) -> Results: … return results

expected_read_types

def expected_read_types(self, results: Results | None = None, available_types: Mapping[str, ExpectedType] | None = None) -> Mapping[str, ExpectedType]:

Return the expected types for keys this step reads.

Default implementation returns the static read_types class attribute. Override this when the expected types depend on instance state or on the types written by upstream steps.

Args:

• results: Live Results object, when called by validate.
• available_types: Type map accumulated by the Pipeline so far,
• when called during dry-run validation.

Returns:

• Mapping from read-key to one or more allowed types.

expected_write_types

def expected_write_types(self, results: Results | None = None, available_types: Mapping[str, ExpectedType] | None = None) -> Mapping[str, ExpectedType]:

Return the expected types for keys this step writes.

Default implementation returns the static write_types class attribute. Override this when the written type depends on instance state (e.g., on the type passed to the constructor).

Args:

• results: Live Results object, when known.
• available_types: Type map accumulated by the Pipeline so far,
• when called during dry-run validation.

Returns:

• Mapping from write-key to the type(s) this step will produce.

validate

def validate(self, results: Results) -> None:

Check that all required keys are present in results.

Raises:

• KeyError: If a required key is missing.