feat: Enhance configuration management and CLI for FlowerPower

- Added detailed docstrings and examples to Config, PipelineConfig, and ProjectConfig classes for better usability.
- Introduced init_config, init_pipeline_config, and init_project_config functions to streamline configuration initialization.
- Refactored the CLI to replace the scheduler commands with worker commands, consolidating functionality under a new worker.py module.
- Removed the deprecated scheduler.py file and its associated commands.
- Implemented hook management in PipelineManager, allowing users to add hooks to pipelines.
- Updated Worker class to dynamically determine the worker type from ProjectConfig.
- Enhanced logging setup across the CLI commands for better traceability.
- Adjusted filesystem caching behavior in load methods for ProjectConfig and PipelineConfig.
- Updated settings to use EXECUTOR_MAX_WORKERS for worker count configuration.
- Cleaned up unused template code in templates.py.

Author: legout

examples/hello-world/conf/project.yml

@@ -1,6 +1,7 @@
 name: null
 worker:
   type: apscheduler
+  
   backend:
     data_store:
       type: postgresql
@@ -33,7 +34,7 @@ worker:
     cleanup_interval: 300
     max_concurrent_jobs: 10
     default_job_executor: threadpool
-    num_workers: 10
+    num_workers: 100
 adapter:
   hamilton_tracker:
     username: null

src/flowerpower/pipeline.py src/flowerpower/_pipeline.pysrc/flowerpower/pipeline.py renamed to src/flowerpower/_pipeline.py

@@ -5,11 +5,38 @@
 
 from ..fs import AbstractFileSystem, get_filesystem
 from .base import BaseConfig
-from .pipeline import PipelineConfig
-from .project import ProjectConfig
+from .pipeline import PipelineConfig, init_pipeline_config
+from .project import ProjectConfig, init_project_config
 
 
 class Config(BaseConfig):
+    """Main configuration class for FlowerPower, combining project and pipeline settings.
+
+    This class serves as the central configuration manager, handling both project-wide
+    and pipeline-specific settings. It provides functionality for loading and saving
+    configurations using various filesystem abstractions.
+
+    Attributes:
+        pipeline (PipelineConfig): Configuration for the pipeline.
+        project (ProjectConfig): Configuration for the project.
+        fs (AbstractFileSystem | None): Filesystem abstraction for I/O operations.
+        base_dir (str | Path | None): Base directory for the configuration.
+        storage_options (dict | Munch): Options for filesystem operations.
+
+    Example:
+        ```python
+        # Load configuration
+        config = Config.load(
+            base_dir="my_project",
+            name="project1",
+            pipeline_name="data-pipeline"
+        )
+
+        # Save configuration
+        config.save(project=True, pipeline=True)
+        ```
+    """
+
     pipeline: PipelineConfig = msgspec.field(default_factory=PipelineConfig)
     project: ProjectConfig = msgspec.field(default_factory=ProjectConfig)
     fs: AbstractFileSystem | None = None
@@ -26,6 +53,29 @@ def load(
         fs: AbstractFileSystem | None = None,
         storage_options: dict | Munch = Munch(),
     ):
+        """Load both project and pipeline configurations.
+
+        Args:
+            base_dir (str, optional): Base directory for configurations. Defaults to ".".
+            name (str | None, optional): Project name. Defaults to None.
+            pipeline_name (str | None, optional): Pipeline name. Defaults to None.
+            worker_type (str | None, optional): Type of worker to use. Defaults to None.
+            fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+            storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+        Returns:
+            Config: Combined configuration instance.
+
+        Example:
+            ```python
+            config = Config.load(
+                base_dir="my_project",
+                name="test_project",
+                pipeline_name="etl",
+                worker_type="local"
+            )
+            ```
+        """
         if fs is None:
             fs = get_filesystem(base_dir, cached=True, dirfs=True, **storage_options)
         project = ProjectConfig.load(
@@ -57,6 +107,19 @@ def save(
         fs: AbstractFileSystem | None = None,
         storage_options: dict | Munch = Munch(),
     ):
+        """Save project and/or pipeline configurations.
+
+        Args:
+            project (bool, optional): Whether to save project config. Defaults to False.
+            pipeline (bool, optional): Whether to save pipeline config. Defaults to True.
+            fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+            storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+        Example:
+            ```python
+            config.save(project=True, pipeline=True)
+            ```
+        """
         if fs is None and self.fs is None:
             self.fs = get_filesystem(
                 self.base_dir, cached=True, dirfs=True, **storage_options
@@ -84,6 +147,25 @@ def load(
     storage_options: dict | Munch = Munch(),
     fs: AbstractFileSystem | None = None,
 ):
+    """Helper function to load configuration.
+
+    This is a convenience wrapper around Config.load().
+
+    Args:
+        base_dir (str): Base directory for configurations.
+        name (str | None, optional): Project name. Defaults to None.
+        pipeline_name (str | None, optional): Pipeline name. Defaults to None.
+        storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+        fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+
+    Returns:
+        Config: Combined configuration instance.
+
+    Example:
+        ```python
+        config = load(base_dir="my_project", name="test", pipeline_name="etl")
+        ```
+    """
     return Config.load(
         name=name,
         pipeline_name=pipeline_name,
@@ -100,6 +182,73 @@ def save(
     fs: AbstractFileSystem | None = None,
     storage_options: dict | Munch = Munch(),
 ):
+    """Helper function to save configuration.
+
+    This is a convenience wrapper around Config.save().
+
+    Args:
+        config (Config): Configuration instance to save.
+        project (bool, optional): Whether to save project config. Defaults to False.
+        pipeline (bool, optional): Whether to save pipeline config. Defaults to True.
+        fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+        storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+    Example:
+        ```python
+        config = load(base_dir="my_project")
+        save(config, project=True, pipeline=True)
+        ```
+    """
     config.save(
         project=project, pipeline=pipeline, fs=fs, storage_options=storage_options
     )
+
+
+def init_config(
+    base_dir: str = ".",
+    name: str | None = None,
+    pipeline_name: str | None = None,
+    worker_type: str | None = None,
+    fs: AbstractFileSystem | None = None,
+    storage_options: dict | Munch = Munch(),
+):
+    """Initialize a new configuration with both project and pipeline settings.
+
+    This function creates and initializes both project and pipeline configurations,
+    combining them into a single Config instance.
+
+    Args:
+        base_dir (str, optional): Base directory for configurations. Defaults to ".".
+        name (str | None, optional): Project name. Defaults to None.
+        pipeline_name (str | None, optional): Pipeline name. Defaults to None.
+        worker_type (str | None, optional): Type of worker to use. Defaults to None.
+        fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+        storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+    Returns:
+        Config: The initialized configuration instance.
+
+    Example:
+        ```python
+        config = init_config(
+            base_dir="my_project",
+            name="test_project",
+            pipeline_name="data-pipeline",
+            worker_type="local"
+        )
+        ```
+    """
+    pipeline_cfg = init_pipeline_config(
+        base_dir=base_dir,
+        name=pipeline_name,
+        fs=fs,
+        storage_options=storage_options,
+    )
+    project_cfg = init_project_config(
+        base_dir=base_dir,
+        name=name,
+        worker_type=worker_type,
+        fs=fs,
+        storage_options=storage_options,
+    )
+    return Config(pipeline=pipeline_cfg, project=project_cfg, fs=fs, base_dir=base_dir)

src/flowerpower/cfg/__init__.py

@@ -11,6 +11,36 @@
 
 
 class PipelineConfig(BaseConfig):
+    """Configuration class for managing pipeline settings in FlowerPower.
+
+    This class handles pipeline-specific configuration including run settings, scheduling,
+    parameters, and adapter settings. It supports Hamilton-style parameter configuration
+    and YAML serialization.
+
+    Attributes:
+        name (str | None): The name of the pipeline.
+        run (RunConfig): Configuration for pipeline execution.
+        schedule (ScheduleConfig): Configuration for pipeline scheduling.
+        params (dict): Pipeline parameters.
+        adapter (AdapterConfig): Configuration for the pipeline adapter.
+        h_params (dict): Hamilton-formatted parameters.
+
+    Example:
+        ```python
+        # Create a new pipeline config
+        pipeline = PipelineConfig(name="data-transform")
+
+        # Set parameters
+        pipeline.params = {
+            "input_path": "data/input",
+            "batch_size": 100
+        }
+
+        # Save configuration
+        pipeline.save(name="data-transform")
+        ```
+    """
+
     name: str | None = msgspec.field(default=None)
     run: RunConfig = msgspec.field(default_factory=RunConfig)
     schedule: ScheduleConfig = msgspec.field(default_factory=ScheduleConfig)
@@ -62,7 +92,26 @@ def update(self, d: dict | Munch):
 
     @staticmethod
     def to_h_params(d: dict) -> dict:
-        """Converts a dictionary of function arguments to Hamilton function parameters"""
+        """Convert a dictionary of parameters to Hamilton-compatible format.
+
+        This method transforms regular parameter dictionaries into Hamilton's function parameter
+        format, supporting nested parameters and source/value decorators.
+
+        Args:
+            d (dict): The input parameter dictionary.
+
+        Returns:
+            dict: Hamilton-formatted parameter dictionary.
+
+        Example:
+            ```python
+            params = {
+                "batch_size": 100,
+                "paths": {"input": "data/in", "output": "data/out"}
+            }
+            h_params = PipelineConfig.to_h_params(params)
+            ```
+        """
 
         def transform_recursive(val, original_dict, depth=1):
             if isinstance(val, dict):
@@ -97,8 +146,27 @@ def load(
         fs: AbstractFileSystem | None = None,
         storage_options: dict | Munch = Munch(),
     ):
+        """Load pipeline configuration from a YAML file.
+
+        Args:
+            base_dir (str, optional): Base directory for the pipeline. Defaults to ".".
+            name (str | None, optional): Pipeline name. Defaults to None.
+            fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+            storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+        Returns:
+            PipelineConfig: Loaded pipeline configuration.
+
+        Example:
+            ```python
+            pipeline = PipelineConfig.load(
+                base_dir="my_project",
+                name="data-pipeline"
+            )
+            ```
+        """
         if fs is None:
-            fs = get_filesystem(base_dir, cached=True, dirfs=True, **storage_options)
+            fs = get_filesystem(base_dir, cached=False, dirfs=True, **storage_options)
         if fs.exists("conf/pipelines"):
             if name is not None:
                 pipeline = PipelineConfig.from_yaml(
@@ -120,6 +188,22 @@ def save(
         fs: AbstractFileSystem | None = None,
         storage_options: dict | Munch = Munch(),
     ):
+        """Save pipeline configuration to a YAML file.
+
+        Args:
+            name (str | None, optional): Pipeline name. Defaults to None.
+            base_dir (str, optional): Base directory for the pipeline. Defaults to ".".
+            fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+            storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+        Raises:
+            ValueError: If pipeline name is not set.
+
+        Example:
+            ```python
+            pipeline_config.save(name="data-pipeline", base_dir="my_project")
+            ```
+        """
         if fs is None:
             fs = get_filesystem(base_dir, cached=True, dirfs=True, **storage_options)
 
@@ -134,3 +218,37 @@ def save(
         self.to_yaml(path=f"conf/pipelines/{self.name}.yml", fs=fs)
 
         setattr(self, "h_params", h_params)
+
+
+def init_pipeline_config(
+    base_dir: str = ".",
+    name: str | None = None,
+    fs: AbstractFileSystem | None = None,
+    storage_options: dict | Munch = Munch(),
+):
+    """Initialize a new pipeline configuration.
+
+    This function creates a new pipeline configuration and saves it to disk.
+
+    Args:
+        base_dir (str, optional): Base directory for the pipeline. Defaults to ".".
+        name (str | None, optional): Pipeline name. Defaults to None.
+        fs (AbstractFileSystem | None, optional): Filesystem to use. Defaults to None.
+        storage_options (dict | Munch, optional): Options for filesystem. Defaults to empty Munch.
+
+    Returns:
+        PipelineConfig: The initialized pipeline configuration.
+
+    Example:
+        ```python
+        pipeline = init_pipeline_config(
+            base_dir="my_project",
+            name="etl-pipeline"
+        )
+        ```
+    """
+    pipeline = PipelineConfig.load(
+        base_dir=base_dir, name=name, fs=fs, storage_options=storage_options
+    )
+    pipeline.save(name=name, base_dir=base_dir, fs=fs, storage_options=storage_options)
+    return pipeline