fix(executor): treat "local" as synchronous and add debug logs

- Normalize executor type "local" -> "synchronous" so local configs use the
  synchronous executor.
- Add debug logging for executor selection (synchronous/local, threadpool,
  processpool, ray, dask).
- Minor formatting and whitespace cleanup across pipeline and executor modules.
- chore(pyproject): bump version to 0.32.3

Author: legout

pyproject.toml

@@ -4,7 +4,7 @@ description = "A simple workflow framework for building and managing data proces
 authors = [{ name = "Volker L.", email = "[email protected]" }]
 readme = "README.md"
 requires-python = ">= 3.11"
-version = "0.32.2"
+version = "0.32.3"
 keywords = ["hamilton", "workflow", "pipeline", "scheduler", "dask", "ray"]
 
 dependencies = [

src/flowerpower/pipeline/executor.py

@@ -13,22 +13,22 @@
 
 class PipelineExecutor:
     """Handles pipeline execution with comprehensive parameter handling.
-    
+
     This class is responsible for:
     - Executing pipelines with various configurations
     - Merging runtime parameters with pipeline defaults
     - Setting up execution environment (logging, etc.)
     - Delegating to Pipeline objects for actual execution
     """
-    
+
     def __init__(
         self,
         config_manager: "PipelineConfigManager",
         registry: "PipelineRegistry",
-        project_context: Optional[Any] = None
+        project_context: Optional[Any] = None,
     ):
         """Initialize the pipeline executor.
-        
+
         Args:
             config_manager: Configuration manager for accessing pipeline configs
             registry: Pipeline registry for accessing pipeline objects
@@ -37,89 +37,83 @@ def __init__(
         self._config_manager = config_manager
         self._registry = registry
         self._project_context = project_context
-    
+
     def run(
-        self,
-        name: str,
-        run_config: Optional[RunConfig] = None,
-        **kwargs
+        self, name: str, run_config: Optional[RunConfig] = None, **kwargs
     ) -> dict[str, Any]:
         """Execute a pipeline synchronously and return its results.
-        
+
         This is the main method for running pipelines directly. It handles configuration
         loading, adapter setup, and execution via Pipeline objects.
-        
+
         Args:
             name: Name of the pipeline to run. Must be a valid identifier.
             run_config: Run configuration object containing all execution parameters.
                 If None, the default configuration from the pipeline will be used.
             **kwargs: Additional parameters to override the run_config.
-                
+
         Returns:
             dict[str, Any]: Results of pipeline execution
-            
+
         Raises:
             ValueError: If pipeline configuration cannot be loaded
             Exception: If pipeline execution fails
         """
         # Load pipeline configuration
         pipeline_config = self._config_manager.load_pipeline_config(name=name)
-        
+
         # Initialize run_config with pipeline defaults if not provided
         run_config = run_config or pipeline_config.run
-        
+
         # Merge kwargs into run_config
         if kwargs:
             run_config = merge_run_config_with_kwargs(run_config, kwargs)
-        
+
         # Set up logging for this specific run if log_level is provided
         if run_config.log_level is not None:
             setup_logging(level=run_config.log_level)
-        
+
         # Get the pipeline object from registry
         pipeline = self._registry.get_pipeline(
             name=name,
             project_context=self._project_context,
         )
-        
+
         # Execute the pipeline
         return pipeline.run(run_config=run_config)
-    
+
     async def run_async(
-        self,
-        name: str,
-        run_config: Optional[RunConfig] = None,
-        **kwargs
+        self, name: str, run_config: Optional[RunConfig] = None, **kwargs
     ) -> dict[str, Any]:
         """Execute a pipeline asynchronously and return its results.
-        
+
         Args:
             name: Name of the pipeline to run
             run_config: Run configuration object
             **kwargs: Additional parameters to override the run_config
-            
+
         Returns:
             dict[str, Any]: Results of pipeline execution
         """
         # Load pipeline configuration
         pipeline_config = self._config_manager.load_pipeline_config(name=name)
-        
+
         # Initialize run_config with pipeline defaults if not provided
         run_config = run_config or pipeline_config.run
-        
+
         # Merge kwargs into run_config
         if kwargs:
             run_config = merge_run_config_with_kwargs(run_config, kwargs)
-        
+
         # Set up logging for this specific run if log_level is provided
         if run_config.log_level is not None:
             setup_logging(level=run_config.log_level)
-        
+
         # Get the pipeline object from registry
         pipeline = self._registry.get_pipeline(
             name=name,
             project_context=self._project_context,
         )
-        
+
         # Execute the pipeline asynchronously
-        return await pipeline.run_async(run_config=run_config)
+        return await pipeline.run_async(run_config=run_config)

src/flowerpower/pipeline/pipeline.py

@@ -125,7 +125,9 @@ def run(self, run_config: RunConfig | None = None, **kwargs) -> dict[str, Any]:
             retry_cfg.max_retries,
             retry_cfg.retry_delay,
             retry_cfg.jitter_factor,
-            tuple(retry_cfg.retry_exceptions) if isinstance(retry_cfg.retry_exceptions, (list, tuple)) else retry_cfg.retry_exceptions,
+            tuple(retry_cfg.retry_exceptions)
+            if isinstance(retry_cfg.retry_exceptions, (list, tuple))
+            else retry_cfg.retry_exceptions,
         )
         max_retries = retry_config["max_retries"]
         retry_delay = retry_config["retry_delay"]
@@ -151,9 +153,19 @@ def _setup_retry_config(
     ) -> dict:
         """Set up retry configuration with defaults and validation."""
         cfg = self.config.run.retry
-        max_retries = max_retries if max_retries is not None else (cfg.max_retries if cfg else 0)
-        retry_delay = retry_delay if retry_delay is not None else (cfg.retry_delay if cfg else 1.0)
-        jitter_factor = jitter_factor if jitter_factor is not None else (cfg.jitter_factor if cfg else 0.1)
+        max_retries = (
+            max_retries if max_retries is not None else (cfg.max_retries if cfg else 0)
+        )
+        retry_delay = (
+            retry_delay
+            if retry_delay is not None
+            else (cfg.retry_delay if cfg else 1.0)
+        )
+        jitter_factor = (
+            jitter_factor
+            if jitter_factor is not None
+            else (cfg.jitter_factor if cfg else 0.1)
+        )
 
         # Convert string exceptions to actual exception classes
         if retry_exceptions and isinstance(retry_exceptions, (list, tuple)):

src/flowerpower/utils/executor.py

@@ -73,16 +73,23 @@ def _normalize_config(
     def _create_executor_by_type(self, executor_cfg: Any) -> Any:
         """Create executor based on type."""
         executor_type = executor_cfg.type or "synchronous"
+        if executor_type == "local":
+            executor_type = "synchronous"
 
         if executor_type in ("synchronous", None):
+            logger.debug("Using synchronous/local executor.")
             return self._create_synchronous_executor()
         elif executor_type == "threadpool":
+            logger.debug("Using thread pool executor.")
             return self._create_threadpool_executor(executor_cfg)
         elif executor_type == "processpool":
+            logger.debug("Using process pool executor.")
             return self._create_processpool_executor(executor_cfg)
         elif executor_type == "ray":
+            logger.debug("Using Ray executor.")
             return self._create_ray_executor(executor_cfg)
         elif executor_type == "dask":
+            logger.debug("Using Dask executor.")
             return self._create_dask_executor(executor_cfg)
         else:
             logger.warning(