Enhance PipelineRunner with context management and logging improvements

- Added context manager support to PipelineRunner for better resource management.
- Integrated logging level configuration from settings into the setup_logging function.
- Enhanced adapter setup to allow additional Hamilton adapters and improved logging of enabled adapters.
- Updated pipeline execution logging to include execution time using the humanize library.
- Refactored settings to include Hamilton-specific configurations for telemetry and data capture limits.
- Added humanize library as a dependency for better time formatting.

Author: legout

examples/hello-world/conf/project.yml

@@ -1,4 +1,4 @@
-name: null
+name: hello-world
 worker:
   type: apscheduler
   backend:

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
   'aiobotocore<2.18.0',
   'dill>=0.3.8',
   'fsspec>=2024.10.0',
+  "humanize>=4.12.2",
   "msgspec>=0.19.0",
   'munch>=4.0.0',
   "orjson>=3.10.15",

src/flowerpower/cfg/pipeline/adapter.py

@@ -2,12 +2,15 @@
 from munch import munchify
 
 from ..base import BaseConfig
-
+from ... import settings
 
 class HamiltonTracerConfig(BaseConfig):
     project_id: int | None = msgspec.field(default=None)
     dag_name: str | None = msgspec.field(default=None)
     tags: dict = msgspec.field(default_factory=dict)
+    capture_data_statistics: bool = msgspec.field(default=settings.HAMILTON_CAPTURE_DATA_STATISTICS)
+    max_list_length_capture: int = msgspec.field(default=settings.HAMILTON_MAX_LIST_LENGTH_CAPTURE)
+    max_dict_length_capture: int = msgspec.field(default=settings.HAMILTON_MAX_DICT_LENGTH_CAPTURE)
 
     def __post_init__(self):
         self.tags = munchify(self.tags)

src/flowerpower/cfg/pipeline/run.py

@@ -4,7 +4,7 @@
 from munch import munchify
 
 from ..base import BaseConfig
-
+from ... import settings
 
 class WithAdapterConfig(BaseConfig):
     tracker: bool = msgspec.field(default=False)
@@ -17,9 +17,9 @@ class WithAdapterConfig(BaseConfig):
 
 
 class ExecutorConfig(BaseConfig):
-    type: str | None = msgspec.field(default=None)
-    max_workers: int | None = msgspec.field(default=10)
-    num_cpus: int | None = msgspec.field(default_factory=os.cpu_count)
+    type: str | None = msgspec.field(default=settings.FP_EXECUTOR)
+    max_workers: int | None = msgspec.field(default=settings.FP_EXECUTOR_MAX_WORKERS)
+    num_cpus: int | None = msgspec.field(default=settings.FP_EXECUTOR_NUM_CPUS)
 
 
 class RunConfig(BaseConfig):
@@ -31,6 +31,8 @@ class RunConfig(BaseConfig):
         default_factory=WithAdapterConfig
     )
     executor: ExecutorConfig | dict = msgspec.field(default_factory=ExecutorConfig)
+    log_level: str | None = msgspec.field(default=None)
+
 
     def __post_init__(self):
         if isinstance(self.inputs, dict):

src/flowerpower/cfg/project/adapter.py

@@ -2,12 +2,12 @@
 from munch import munchify
 
 from ..base import BaseConfig
-
+from ... import settings
 
 class HamiltonTrackerConfig(BaseConfig):
     username: str | None = msgspec.field(default=None)
-    api_url: str = msgspec.field(default="http://localhost:8241")
-    ui_url: str = msgspec.field(default="http://localhost:8242")
+    api_url: str = msgspec.field(default=settings.HAMILTON_API_URL)
+    ui_url: str = msgspec.field(default=settings.HAMILTON_UI_URL)
     api_key: str | None = msgspec.field(default=None)
     verify: bool = msgspec.field(default=False)
 

src/flowerpower/cfg/project/worker.py

@@ -1,7 +1,7 @@
 import datetime as dt
 
 import msgspec
-
+from ... import settings
 from ..base import BaseConfig
 
 # class Worker(Enum):
@@ -30,15 +30,46 @@ class WorkerBackendConfig(BaseConfig):
 
 
 class APSDataStoreConfig(WorkerBackendConfig):
-    type: str = msgspec.field(default_factory=lambda: "postgresql")
-    host: str = msgspec.field(default_factory=lambda: "localhost")
-    port: int = msgspec.field(default_factory=lambda: 5432)
+    type: str = msgspec.field(default=settings.FP_APS_WORKER_BACKEND_DS)
+    host: str = msgspec.field(
+        default=settings.BACKEND_PROPERTIES[settings.FP_APS_WORKER_BACKEND_DS][
+            "default_host"
+        ]
+    )
+    port: int = msgspec.field(
+        default=settings.BACKEND_PROPERTIES[settings.FP_APS_WORKER_BACKEND_DS][
+            "default_port"
+        ]
+    )
     schema: str | None = msgspec.field(default="flowerpower")
-    username: str = msgspec.field(default_factory=lambda: "postgres")
+    username: str = msgspec.field(
+        default=settings.BACKEND_PROPERTIES[settings.FP_APS_WORKER_BACKEND_DS][
+            "default_username"
+        ]
+    )
 
 
 class APSEventBrokerConfig(WorkerBackendConfig):
-    from_ds_sqla: bool = msgspec.field(default=True)
+    type: str = msgspec.field(default=settings.FP_APS_WORKER_BACKEND_EB)
+    host: str = msgspec.field(
+        default=settings.BACKEND_PROPERTIES[settings.FP_APS_WORKER_BACKEND_EB][
+            "default_host"
+        ]
+    )
+    port: int = msgspec.field(
+        default=settings.BACKEND_PROPERTIES[settings.FP_APS_WORKER_BACKEND_EB][
+            "default_port"
+        ]
+    )
+    username: str = msgspec.field(
+        default=settings.BACKEND_PROPERTIES[settings.FP_APS_WORKER_BACKEND_EB][
+            "default_username"
+        ]
+    )
+    from_ds_sqla: bool = msgspec.field(
+        default_factory=lambda: settings.FP_APS_WORKER_BACKEND_EB == "postgresql"
+        and settings.FP_APS_WORKER_BACKEND_DS == "postgresql"
+    )
 
 
 class APSBackendConfig(BaseConfig):
@@ -47,18 +78,33 @@ class APSBackendConfig(BaseConfig):
         default_factory=APSEventBrokerConfig
     )
     cleanup_interval: int | float | dt.timedelta = msgspec.field(
-        default=300
+        default=settings.FP_APS_WORKER_CLEANUP_INTERVAL
     )  # int in secods
-    max_concurrent_jobs: int = msgspec.field(default=10)
-    default_job_executor: str | None = msgspec.field(default="threadpool")
-    num_workers: int | None = msgspec.field(default=None)
+    max_concurrent_jobs: int = msgspec.field(
+        default=settings.FP_APS_WORKER_MAX_CONCURRENT_JOBS
+    )
+    default_job_executor: str | None = msgspec.field(
+        default=settings.FP_EXECUTOR
+    )
+    num_workers: int | None = msgspec.field(
+        default=settings.FP_EXECUTOR_NUM_CPUS
+    )
 
 
 class RQBackendConfig(WorkerBackendConfig):
-    type: str = msgspec.field(default_factory=lambda: "redis")
-    host: str = msgspec.field(default_factory=lambda: "localhost")
-    port: int = msgspec.field(default_factory=lambda: 6379)
-    queues: str | list[str] = msgspec.field(default_factory=lambda: ["default"])
+    type: str = msgspec.field(default="redis")
+    host: str = msgspec.field(
+        default=settings.BACKEND_PROPERTIES["redis"]["default_host"]
+    )
+    port: int = msgspec.field(
+        default=settings.BACKEND_PROPERTIES["redis"]["default_port"]
+    )
+    database: int = msgspec.field(
+        default=settings.BACKEND_PROPERTIES["redis"]["default_database"]
+    )
+    queues: str | list[str] = msgspec.field(
+        default_factory=lambda: ["low-prio", "default", "high-prio"]
+    )
 
 
 class HueyBackendConfig(WorkerBackendConfig):

src/flowerpower/pipeline/base.py

@@ -1,7 +1,7 @@
 import importlib
 import posixpath
 import sys
-from traceback import TracebackType
+from types import TracebackType
 
 from loguru import logger
 from munch import Munch