fix(visualizer,manager): pass base_dir to save_dag and use it for graph outputs

- PipelineManager: forward self._base_dir when calling PipelineVisualizer.save_dag
  and perform small formatting/whitespace cleanups.
- PipelineVisualizer: add base_dir parameter to save_dag and use it to determine
  the graphs directory (instead of project_cfg.base_dir).
- Bump package versions: pyproject.toml -> 0.32.1, uv.lock flowerpower -> 0.32.0

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.0"
+version = "0.32.1"
 keywords = ["hamilton", "workflow", "pipeline", "scheduler", "dask", "ray"]
 
 dependencies = [

src/flowerpower/pipeline/manager.py

@@ -72,7 +72,6 @@ def __init__(
         fs: AbstractFileSystem | None = None,
         cfg_dir: str | None = CONFIG_DIR,
         pipelines_dir: str | None = PIPELINES_DIR,
-        
         log_level: str | None = None,
     ) -> None:
         """Initialize the PipelineManager.
@@ -91,7 +90,7 @@ def __init__(
                 Example: "config" or "settings".
             pipelines_dir: Override default pipelines directory name ('pipelines').
                 Example: "flows" or "dags".
-            
+
             log_level: Set logging level for the manager.
                 Valid values: "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
 
@@ -111,7 +110,7 @@ def __init__(
             ...         "key": "ACCESS_KEY",
             ...         "secret": "SECRET_KEY"
             ...     },
-            
+
             ...     log_level="DEBUG"
             ... )
         """
@@ -129,7 +128,7 @@ def _setup_filesystem(
         storage_options: dict | Munch | BaseStorageOptions | None,
         fs: AbstractFileSystem | None,
         cfg_dir: str | None,
-        pipelines_dir: str | None
+        pipelines_dir: str | None,
     ) -> None:
         """Setup filesystem and configuration directories.
 
@@ -185,7 +184,7 @@ def _initialize_managers(self) -> None:
             base_dir=self._base_dir,
             fs=self._fs,
             storage_options=self._storage_options,
-            cfg_dir=self._cfg_dir
+            cfg_dir=self._cfg_dir,
         )
 
         # Load project configuration
@@ -201,25 +200,21 @@ def _initialize_managers(self) -> None:
 
         # Initialize specialized managers
         self._executor = PipelineExecutor(
-            config_manager=self._config_manager,
-            registry=self.registry
+            config_manager=self._config_manager, registry=self.registry
         )
         self._lifecycle_manager = PipelineLifecycleManager(registry=self.registry)
 
         # Initialize other components
         self._project_context = None
         self.visualizer = PipelineVisualizer(
-            project_cfg=self._config_manager.project_config,
-            fs=self._fs
+            project_cfg=self._config_manager.project_config, fs=self._fs
         )
         self.io = PipelineIOManager(registry=self.registry)
 
     def _ensure_directories_exist(self) -> None:
         """Ensure essential directories exist."""
         self._fs_helper.ensure_directories_exist(
-            self._fs,
-            self._cfg_dir,
-            self._pipelines_dir
+            self._fs, self._cfg_dir, self._pipelines_dir
         )
 
     def _add_modules_path(self) -> None:
@@ -289,7 +284,6 @@ def __exit__(
         # Add cleanup code if needed
         pass
 
-
     def load_pipeline(self, name: str, reload: bool = False) -> PipelineConfig:
         """Load or reload configuration for a specific pipeline.
 
@@ -353,12 +347,8 @@ def pipeline_cfg(self) -> PipelineConfig:
 
     # --- Core Execution Method ---
 
-
     def run(
-        self,
-        name: str,
-        run_config: RunConfig | None = None,
-        **kwargs
+        self, name: str, run_config: RunConfig | None = None, **kwargs
     ) -> dict[str, Any]:
         """Execute a pipeline synchronously and return its results.
 
@@ -433,7 +423,7 @@ def run(
         # Set project context for executor
         if hasattr(self, "_project_context") and self._project_context is not None:
             self._executor._project_context = self._project_context
-        
+
         # Delegate to executor
         return self._executor.run(name=name, run_config=run_config, **kwargs)
 
@@ -600,10 +590,12 @@ def show_pipelines(self, format: str = "table") -> None:
         try:
             if fmt == "json":
                 import json
+
                 print(json.dumps(names))
                 return None
             if fmt == "yaml":
                 import yaml  # type: ignore
+
                 print(yaml.safe_dump(names, sort_keys=False))
                 return None
         except Exception as e:
@@ -1054,8 +1046,13 @@ def save_dag(
             ...     reload=True
             ... )
         """
+
         return self.visualizer.save_dag(
-            name=name, format=format, reload=reload, output_path=output_path
+            name=name,
+            format=format,
+            reload=reload,
+            output_path=output_path,
+            base_dir=self._base_dir,
         )
 
     def show_dag(

src/flowerpower/pipeline/visualizer.py

@@ -65,6 +65,7 @@ def _get_dag_object(self, name: str, reload: bool = False):
     def save_dag(
         self,
         name: str,
+        base_dir: str,
         format: str = "png",
         reload: bool = False,
         output_path: str | None = None,
@@ -89,7 +90,7 @@ def save_dag(
 
         # Determine final output path
         if output_path is None:
-            graph_dir = posixpath.join(self.project_cfg.base_dir, "graphs")
+            graph_dir = posixpath.join(base_dir, "graphs")
             self._fs.makedirs(graph_dir, exist_ok=True)
             base = posixpath.join(graph_dir, name)
             final_path = f"{base}.{format}"