ml_grid.util.data_cache

Data caching module for ml_grid pipeline.

This module provides efficient data caching using joblib to cache preprocessed data, feature encoders/scalers, and transformed datasets for reuse between experiments. Implements smart cache invalidation based on file checksums/timestamps.

Exceptions

CacheValidationError

Initialize self. See help(type(self)) for accurate signature.

Classes

`DataCache`	Initialize the data cache system.
`CachedPipelineInstance`	Initialize cached pipeline instance.
`CachedDataPipeline`	Initialize CachedDataPipeline wrapper.

Functions

`with_cache_validation`(→ Callable)	Decorator to add cache validation before method execution.
`get_cache_stats`(→ Dict[str, Any])	Helper function to get cache statistics.
`clear_experiment_cache`(→ int)	Helper function to clear cache for an experiment.

Module Contents

exception ml_grid.util.data_cache.CacheValidationError[source]

Bases: Exception

Initialize self. See help(type(self)) for accurate signature.

class ml_grid.util.data_cache.DataCache(base_cache_dir: str | None = None)[source]

Initialize the data cache system.

global_params[source]

base_cache_dir[source]

property cache_dir: pathlib.Path[source]: Get the current experiment-specific cache directory.

get_cache_path(experiment_name: str, run_id: str) → pathlib.Path[source]

Get full path for a cached item.

Parameters:

experiment_name – Name of the experiment
run_id – Unique identifier for this specific run

Returns:

Full path to cache file

save(experiment_name: str, run_id: str, data_dict: Dict[str, Any], metadata: Dict[str, Any] | None = None) → bool[source]

Save data dictionary to cache.

Parameters:

experiment_name – Name of the experiment
run_id – Unique identifier for this run
data_dict – Dictionary of data to cache (X_train, X_test, scaler, etc.)
metadata – Optional metadata to store

Returns:

True if save was successful, False otherwise

load(experiment_name: str, run_id: str) → Dict[str, Any] | None[source]

Load cached data.

Parameters:

experiment_name – Name of the experiment
run_id – Unique identifier for this run

Returns:

Cached data dictionary or None if not found/invalid

invalidate(experiment_name: str, run_id: str) → bool[source]

Invalidate (delete) a specific cache entry.

Parameters:

experiment_name – Name of the experiment
run_id – Unique identifier for this run

Returns:

True if invalidated successfully

clear_experiment_cache(experiment_name: str) → int[source]

Clear all cached data for an experiment.

Parameters:: experiment_name – Name of the experiment
Returns:: Number of files removed

get_cache_stats() → Dict[str, Any][source]: Get statistics about the cache directory.

class ml_grid.util.data_cache.CachedPipelineInstance(experiment_name: str, run_id: str, cache: DataCache, file_path: str, local_param_dict: Dict[str, Any], original_kwargs: Dict[str, Any], original_class: Any)[source]

Initialize cached pipeline instance.

experiment_name[source]

run_id[source]

cache[source]

file_path[source]

local_param_dict[source]

original_kwargs[source]

original_class[source]

run_pipeline() → Any[source]

Run the pipeline, using cached data when available.

Returns:: Pipeline instance with data populated

class ml_grid.util.data_cache.CachedDataPipeline(original_pipeline_class: Any, cache: DataCache | None = None)[source]

Initialize CachedDataPipeline wrapper.

Parameters:

original_pipeline_class – The original pipe class to wrap
cache – DataCache instance (creates new if None)

cache[source]

cached_instances[source]

create_cached_pipeline(experiment_name: str, run_id: str, file_path: str, local_param_dict: Dict[str, Any], original_kwargs: Dict[str, Any]) → CachedPipelineInstance[source]

Create a cached pipeline instance.

Parameters:

experiment_name – Name of the experiment
run_id – Run identifier
file_name – Path to input data
local_param_dict – Pipeline parameters
original_kwargs – Original __init__ arguments

Returns:

CachedPipelineInstance that uses cached data when available

ml_grid.util.data_cache.with_cache_validation(func: Callable) → Callable[source]: Decorator to add cache validation before method execution.

ml_grid.util.data_cache.get_cache_stats(cache: DataCache) → Dict[str, Any][source]: Helper function to get cache statistics.

ml_grid.util.data_cache.clear_experiment_cache(cache: DataCache, experiment_name: str) → int[source]: Helper function to clear cache for an experiment.