Refactor logging setup and configuration

- Introduced a centralized logging setup in `src/flowerpower/utils/logging.py` to standardize log formatting and levels across the application.
- Updated `src/flowerpower/pipeline/scheduler.py`, `src/flowerpower/worker/__init__.py`, and other worker modules to utilize the new logging setup.
- Created a new settings module `src/flowerpower/settings.py` to manage configuration variables, including logging levels and executor settings.
- Removed unused imports and commented-out code in various files for improved code clarity.
- Enhanced timezone handling in `src/flowerpower/utils/sql.py` to ensure proper error handling and timezone awareness.
- Updated dependency markers in `uv.lock` for better compatibility across platforms.
- Added support for `openlineage-python` package in the dependency management.

Author: legout

examples/hello-world/pipelines/hello_world.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 
 import pandas as pd
-from hamilton.function_modifiers import parameterize, config
+from hamilton.function_modifiers import config, parameterize
 from loguru import logger
 
 from flowerpower.cfg import Config
@@ -15,24 +15,28 @@
     Path(__file__).parents[1], pipeline_name="hello_world"
 ).pipeline.h_params
 
+
 @config.when(range=10_000)
 def spend__10000() -> pd.Series:
     """Returns a series of spend data."""
     # time.sleep(2)
     return pd.Series(range(10_000)) * 10
 
+
 @config.when(range=10_000)
 def signups__10000() -> pd.Series:
     """Returns a series of signups data."""
     time.sleep(1)
     return pd.Series(range(10_000))
 
+
 @config.when(range=1_000)
 def spend__1000() -> pd.Series:
     """Returns a series of spend data."""
     # time.sleep(2)
     return pd.Series(range(10_000)) * 10
 
+
 @config.when(range=1_000)
 def signups__1000() -> pd.Series:
     """Returns a series of signups data."""

pyproject.toml

@@ -123,3 +123,6 @@ package = true
 mlflow = [
     "mlflow>=2.21.3",
 ]
+openlineage = [
+    "openlineage-python>=1.31.0",
+]

src/flowerpower/__init__.py

@@ -1,5 +1,10 @@
 import importlib.metadata
 
-from .flowerpower import init as init_flowerpower
+from .flowerpower import init as init_flowerpower # noqa: E402
 
 __version__ = importlib.metadata.version("FlowerPower")
+
+__all__ = [
+    "__version__",
+    "init_flowerpower",
+]

src/flowerpower/cfg/__init__.py

@@ -3,146 +3,11 @@
 import msgspec
 from munch import Munch
 
-from ..fs import get_filesystem, AbstractFileSystem
-#from .base import BaseConfig
+from ..fs import AbstractFileSystem, get_filesystem
+from .base import BaseConfig
 from .pipeline import PipelineConfig
 from .project import ProjectConfig
 
-import copy
-
-from typing import Any, Self
-
-from fsspec import filesystem
-
-
-class BaseConfig(msgspec.Struct, kw_only=True):
-    def to_dict(self) -> dict[str, Any]:
-        return msgspec.to_builtins(self)
-
-    def to_yaml(self, path: str, fs: AbstractFileSystem | None = None) -> None:
-        """
-        Converts the instance to a YAML file.
-        
-        Args:
-            path: The path to the YAML file.
-            fs: An optional filesystem instance to use for file operations.
-            
-        Raises:
-            NotImplementedError: If the filesystem does not support writing files.  
-        """
-        if fs is None:
-            fs = filesystem("file")
-        try:
-            with fs.open(path, "wb") as f:
-                f.write(msgspec.yaml.encode(self, order="deterministic"))
-                # yaml.dump(self.to_dict(), f, default_flow_style=False)
-        except NotImplementedError:
-            raise NotImplementedError("The filesystem does not support writing files.")
-
-    @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> "BaseConfig":
-        """
-        Converts a dictionary to an instance of the class.
-        Args:
-            data: The dictionary to convert.
-        
-        Returns:
-            An instance of the class with the values from the dictionary.
-        """
-        return msgspec.convert(data, cls)
-
-    @classmethod
-    def from_yaml(cls, path: str, fs: AbstractFileSystem | None = None) -> "BaseConfig":
-        """
-        Loads a YAML file and converts it to an instance of the class.
-        
-        Args:
-            path: The path to the YAML file.
-            fs: An optional filesystem instance to use for file operations.
-
-        Returns:
-            An instance of the class with the values from the YAML file.
-        
-        """
-        if fs is None:
-            fs = filesystem("file")
-        with fs.open(path) as f:
-            # data = yaml.full_load(f)
-            # return cls.from_dict(data)
-            return msgspec.yaml.decode(f.read(), type=cls, strict=False)
-
-    def update(self, d: dict[str, Any]) -> None:
-        for k, v in d.items():
-            if hasattr(self, k):
-                current_value = getattr(self, k)
-                if isinstance(current_value, dict) and isinstance(v, dict):
-                    current_value.update(v)
-                else:
-                    setattr(self, k, v)
-            else:
-                setattr(self, k, v)
-
-    def merge_dict(self, d: dict[str, Any]) -> Self:
-        """
-        Creates a copy of this instance and updates the copy with values
-        from the provided dictionary, only if the dictionary field's value is not
-        its default value. The original instance (self) is not modified.
-        
-        Args:
-            d: The dictionary to get values from.
-
-        Returns:
-            A new instance of the struct with updated values.
-        """
-        self_copy = copy.copy(self)
-        for k, v in d.items():
-            if hasattr(self_copy, k):
-                current_value = getattr(self_copy, k)
-                if isinstance(current_value, dict) and isinstance(v, dict):
-                    current_value.update(v)
-                else:
-                    setattr(self_copy, k, v)
-            else:
-                setattr(self_copy, k, v)
-        return self_copy
-    
-    def merge(self, source: Self) -> Self:
-        """
-        Creates a copy of this instance and updates the copy with values
-        from the source struct, only if the source field's value is not
-        its default value. The original instance (self) is not modified.
-
-        Args:
-            source: The msgspec.Struct instance of the same type to get values from.
-
-        Returns:
-            A new instance of the struct with updated values.
-
-        Raises:
-            TypeError: If source is not of the same type as self.
-        """
-        if type(self) is not type(source):
-            raise TypeError(f"Source must be an instance of {type(self).__name__}, not {type(source).__name__}")
-
-        updated_instance = copy.copy(self)
-
-        # Get default values if they exist
-        defaults = getattr(source, '__struct_defaults__', {})
-
-        for field in source.__struct_fields__:
-            source_value = getattr(source, field)
-            has_explicit_default = field in defaults
-            is_default_value = False
-
-            if has_explicit_default:
-                is_default_value = (source_value == defaults[field])
-            else:
-                is_default_value = (source_value is None)
-
-            if not is_default_value:
-               setattr(updated_instance, field, source_value)
-
-        return updated_instance
 
 class Config(BaseConfig):
     pipeline: PipelineConfig = msgspec.field(default_factory=PipelineConfig)

src/flowerpower/cfg/_base.py

@@ -0,0 +1,137 @@
+import copy
+from typing import Any, Self
+
+import msgspec
+from fsspec import AbstractFileSystem, filesystem
+
+
+class BaseConfig(msgspec.Struct, kw_only=True):
+    def to_dict(self) -> dict[str, Any]:
+        return msgspec.to_builtins(self)
+
+    def to_yaml(self, path: str, fs: AbstractFileSystem | None = None) -> None:
+        """
+        Converts the instance to a YAML file.
+
+        Args:
+            path: The path to the YAML file.
+            fs: An optional filesystem instance to use for file operations.
+
+        Raises:
+            NotImplementedError: If the filesystem does not support writing files.
+        """
+        if fs is None:
+            fs = filesystem("file")
+        try:
+            with fs.open(path, "wb") as f:
+                f.write(msgspec.yaml.encode(self, order="deterministic"))
+                # yaml.dump(self.to_dict(), f, default_flow_style=False)
+        except NotImplementedError:
+            raise NotImplementedError("The filesystem does not support writing files.")
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "BaseConfig":
+        """
+        Converts a dictionary to an instance of the class.
+        Args:
+            data: The dictionary to convert.
+
+        Returns:
+            An instance of the class with the values from the dictionary.
+        """
+        return msgspec.convert(data, cls)
+
+    @classmethod
+    def from_yaml(cls, path: str, fs: AbstractFileSystem | None = None) -> "BaseConfig":
+        """
+        Loads a YAML file and converts it to an instance of the class.
+
+        Args:
+            path: The path to the YAML file.
+            fs: An optional filesystem instance to use for file operations.
+
+        Returns:
+            An instance of the class with the values from the YAML file.
+
+        """
+        if fs is None:
+            fs = filesystem("file")
+        with fs.open(path) as f:
+            # data = yaml.full_load(f)
+            # return cls.from_dict(data)
+            return msgspec.yaml.decode(f.read(), type=cls, strict=False)
+
+    def update(self, d: dict[str, Any]) -> None:
+        for k, v in d.items():
+            if hasattr(self, k):
+                current_value = getattr(self, k)
+                if isinstance(current_value, dict) and isinstance(v, dict):
+                    current_value.update(v)
+                else:
+                    setattr(self, k, v)
+            else:
+                setattr(self, k, v)
+
+    def merge_dict(self, d: dict[str, Any]) -> Self:
+        """
+        Creates a copy of this instance and updates the copy with values
+        from the provided dictionary, only if the dictionary field's value is not
+        its default value. The original instance (self) is not modified.
+
+        Args:
+            d: The dictionary to get values from.
+
+        Returns:
+            A new instance of the struct with updated values.
+        """
+        self_copy = copy.copy(self)
+        for k, v in d.items():
+            if hasattr(self_copy, k):
+                current_value = getattr(self_copy, k)
+                if isinstance(current_value, dict) and isinstance(v, dict):
+                    current_value.update(v)
+                else:
+                    setattr(self_copy, k, v)
+            else:
+                setattr(self_copy, k, v)
+        return self_copy
+
+    def merge(self, source: Self) -> Self:
+        """
+        Creates a copy of this instance and updates the copy with values
+        from the source struct, only if the source field's value is not
+        its default value. The original instance (self) is not modified.
+
+        Args:
+            source: The msgspec.Struct instance of the same type to get values from.
+
+        Returns:
+            A new instance of the struct with updated values.
+
+        Raises:
+            TypeError: If source is not of the same type as self.
+        """
+        if type(self) is not type(source):
+            raise TypeError(
+                f"Source must be an instance of {type(self).__name__}, not {type(source).__name__}"
+            )
+
+        updated_instance = copy.copy(self)
+
+        # Get default values if they exist
+        defaults = getattr(source, "__struct_defaults__", {})
+
+        for field in source.__struct_fields__:
+            source_value = getattr(source, field)
+            has_explicit_default = field in defaults
+            is_default_value = False
+
+            if has_explicit_default:
+                is_default_value = source_value == defaults[field]
+            else:
+                is_default_value = source_value is None
+
+            if not is_default_value:
+                setattr(updated_instance, field, source_value)
+
+        return updated_instance