mutcleaner.core.pipeline

Functions

create_pipeline(data[, name])	Create a new pipeline with initial data
multiout_step(**outputs)	Decorator factory for multi-output pipeline functions.
pipeline_step([_func, name])

Classes

Pipeline([data, name, logging_level])

Pipeline for processing data with pandas-style method chaining

class mutcleaner.core.pipeline.Pipeline(data=None, name=None, logging_level='INFO')

Bases: object

Pipeline for processing data with pandas-style method chaining

Attributes:

artifacts

Always return the artifacts dictionary.

data

Always return the actual data, never PipelineOutput.

has_pending_steps

Check if there are delayed steps waiting to be executed

structured_data

Return PipelineOutput object with both data and artifacts.

Methods

add_delayed_step(func[, index])	Add a delayed step before a specific position in the delayed execution queue.
apply(func, *args, **kwargs)	Apply function and return new Pipeline (functional style)
assign(**kwargs)	Add attributes or computed values to data
copy()	Create a deep copy of this pipeline
delayed_then(func, *args, **kwargs)	Add a function to the delayed execution queue without running it immediately
execute([steps])	Execute delayed steps.
filter(condition)	Filter data based on condition
get_all_artifacts()	Get all stored artifacts
get_artifact(name)	Get a specific artifact by name
get_data()	Get current data (same as .data property).
get_delayed_steps_info()	Get information about delayed steps
get_execution_summary()	Get summary of pipeline execution
get_step_result(step_index)	Get result from a specific step by index or name
load(filepath[, format, name])	Load data from file and create new pipeline
load_structured_data(filepath[, format, name])	Load structured data from file and create new pipeline
peek([func, prefix])	Inspect data without modifying it (for debugging)
remove_delayed_step(index_or_name)	Remove a delayed step at the specified index.
save(filepath[, format])	Save current data to file
save_artifacts(filepath[, format])	Save all artifacts to file
save_structured_data(filepath[, format])	Save structured data (data + artifacts) to file
store(name[, extractor])	Store current data or extracted value as artifact
then(func, *args, **kwargs)	Apply a function to the current data (pandas.pipe style)
transform(transformer, *args, **kwargs)	Alias of then, used to define format transformations.
validate(validator[, error_msg])	Validate data and raise error if invalid
visualize_pipeline()	Generate a text visualization of the pipeline

add_delayed_step(func, index=None, *args, **kwargs)

Add a delayed step before a specific position in the delayed execution queue.

Performs a similar action to the list.insert() method.

Parameters:

• func (Callable) – Function to add as delayed step

• index (Optional[int]) – Position to insert the step. If None, appends to the end. Supports negative indexing.

• **kwargs (*args,) –

  Arguments to pass to the function

Return type:

Pipeline

Returns:

Self for method chaining

Examples

>>> # Add step at the beginning
>>> pipeline.add_delayed_step(func1, 0)

>>> # Add step at the end (same as delayed_then)
>>> pipeline.add_delayed_step(func2)

>>> # Insert step at position 2
>>> pipeline.add_delayed_step(func3, 2)

>>> # Insert step before the last one
>>> pipeline.add_delayed_step(func4, -1)

apply(func, *args, **kwargs)

Apply function and return new Pipeline (functional style)

Return type:

Pipeline

property artifacts: Dict[str, Any]

Always return the artifacts dictionary.

This provides direct access to all stored artifacts from pipeline steps.

assign(**kwargs)

Add attributes or computed values to data

Return type:

Pipeline

copy()

Create a deep copy of this pipeline

Return type:

Pipeline

property data: Any

Always return the actual data, never PipelineOutput.

This ensures consistent user experience - pipeline.data can always be used with methods like .copy(), .append(), etc.

delayed_then(func, *args, **kwargs)

Add a function to the delayed execution queue without running it immediately

Return type:

Pipeline

execute(steps=None)

Execute delayed steps.

Parameters:

steps (Union[int, List[int], None]) – Which delayed steps to execute: - None: execute all delayed steps - int: execute the first N delayed steps - List[int]: execute specific delayed steps by index

Return type:

Pipeline

Returns:

Self for method chaining

filter(condition)

Filter data based on condition

Return type:

Pipeline

get_all_artifacts()

Get all stored artifacts

Return type:

Dict[str, Any]

get_artifact(name)

Get a specific artifact by name

Return type:

Any

get_data()

Get current data (same as .data property).

Kept for backward compatibility.

Return type:

Any

get_delayed_steps_info()

Get information about delayed steps

Return type:

List[Dict[str, Any]]

get_execution_summary()

Get summary of pipeline execution

Return type:

Dict[str, Any]

get_step_result(step_index)

Get result from a specific step by index or name

Return type:

Any

property has_pending_steps: bool

Check if there are delayed steps waiting to be executed

classmethod load(filepath, format='pickle', name=None)

Load data from file and create new pipeline

Return type:

Pipeline

classmethod load_structured_data(filepath, format='pickle', name=None)

Load structured data from file and create new pipeline

Return type:

Pipeline

peek(func=None, prefix='')

Inspect data without modifying it (for debugging)

Return type:

Pipeline

remove_delayed_step(index_or_name)

Remove a delayed step at the specified index.

Parameters:

index (int) – Index of the delayed step to remove

Return type:

Pipeline

Returns:

Self for method chaining

Raises:

ValueError – If no delayed step is found with the specified index or name

save(filepath, format='pickle')

Save current data to file

Return type:

Pipeline

save_artifacts(filepath, format='pickle')

Save all artifacts to file

Return type:

Pipeline

save_structured_data(filepath, format='pickle')

Save structured data (data + artifacts) to file

Return type:

Pipeline

store(name, extractor=None)

Store current data or extracted value as artifact

Return type:

Pipeline

property structured_data: PipelineOutput

Return PipelineOutput object with both data and artifacts.

Use this when you need the complete pipeline state for serialization, passing to other systems, or when working with structured data flows.

then(func, *args, **kwargs)

Apply a function to the current data (pandas.pipe style)

Return type:

Pipeline

transform(transformer, *args, **kwargs)

Alias of then, used to define format transformations.

Return type:

Pipeline

validate(validator, error_msg='Validation failed')

Validate data and raise error if invalid

Return type:

Pipeline

visualize_pipeline()

Generate a text visualization of the pipeline

Return type:

str

mutcleaner.core.pipeline.create_pipeline(data, name=None, **kwargs)

Create a new pipeline with initial data

Return type:

Pipeline

mutcleaner.core.pipeline.multiout_step(**outputs)

Decorator factory for multi-output pipeline functions.

Parameters:

**outputs (str) – Named outputs. Use ‘main’ to specify which output is the main data flow. If ‘main’ is not specified, the first return value is treated as main.

Return type:

Callable[[Callable[[ParamSpec(P, bound= None)], Union[Tuple[Any, ...], Any]]], Callable[[ParamSpec(P, bound= None)], MultiOutput]]

Examples

>>> @multiout_step(stats="statistics", plot="visualization")
... def analyze_data(data):
...     # return (main, stats, plot)
...     return processed, stats_dict, plot_obj

mutcleaner.core.pipeline.pipeline_step(_func=None, *, name=None)

Overloads:

• _func (Callable[P, R]) → Callable[P, R]

• name (Optional[str]) → Callable[[Callable[P, R]], Callable[P, R]]

Decorator for single-output pipeline functions.

Use this for functions that return a single value (including tuples as single values). For multiple outputs, use @multiout_step instead.

Parameters:

name (Optional[str]) – Custom name for the step. If None, uses function name.

Examples

>>> @pipeline_step
... def process(data):
...     return processed_data  # Single output

>>> @pipeline_step("process_data")
... def process(data):
...     return processed_data  # Single output

>>> @pipeline_step()
... def get_coordinates():
...     return (10, 20)  # Single tuple output