Bump version to 0.8.4 in pyproject.toml; add config attribute to PipelineRun and PipelineRunConfig classes; enhance MQTTClient methods to accept config parameter for improved pipeline configuration management.

Author: legout

pyproject.toml

@@ -27,7 +27,7 @@ keywords = [
 name = "FlowerPower"
 readme = "README.md"
 requires-python = ">= 3.11"
-version = "0.8.3.3"
+version = "0.8.4"
 
 [project.scripts]
 flowerpower = "flowerpower.cli:app"

src/flowerpower/cfg/pipeline/run.py

@@ -8,6 +8,7 @@ class PipelineRunConfig(BaseConfig):
     final_vars: list[str] = Field(default_factory=list)
     inputs: dict | Munch = Field(default_factory=dict)
     executor: str | None = None
+    config: dict | Munch = Field(default_factory=dict)
     with_tracker: bool = False
     with_opentelemetry: bool = False
 

src/flowerpower/cli/cfg.py

@@ -1 +1,44 @@
+import typer
+
+from ..cli.utils import parse_dict_or_list_param
 from ..cfg import Config
+
+app = typer.Typer(help="Config management commands")
+
+
+# @app.command()
+# def get_project(request) -> json:
+#     cfg = request.app.ctx.pipeline_manager.cfg.project.to_dict()
+#     # cfg.pop("fs")
+#     return json({"cfg": cfg})
+
+
+# @bp.get("/pipeline/<pipeline_name>")
+# async def get_pipeline(request, pipeline_name) -> json:
+#     if pipeline_name != request.app.ctx.pipeline_manager.cfg.pipeline.name:
+#         request.app.ctx.pipeline_manager.load_config(pipeline_name)
+#     cfg = request.app.ctx.pipeline_manager.cfg.pipeline.to_dict()
+#     return json({"cfg": cfg})
+
+
+# @bp.post("/pipeline/<pipeline_name>")
+# @openapi.body({"application/json": PipelineConfig}, required=True)
+# @validate(json=PipelineConfig)
+# async def update_pipeline(request, pipeline_name, body: PipelineConfig) -> json:
+#     data = request.json
+#     if pipeline_name != request.app.ctx.pipeline_manager.cfg.pipeline.name:
+#         request.app.ctx.pipeline_manager.load_config(pipeline_name)
+#     cfg = request.app.ctx.pipeline_manager.cfg.pipeline.copy()
+#     cfg.update(data)
+#     try:
+#         cfg.to_yaml(
+#             os.path.join(
+#                 "pipelines",
+#                 pipeline_name + ".yml",
+#             ),
+#             fs=request.app.ctx.pipeline_manager.cfg.fs,
+#         )
+#     except NotImplementedError as e:
+#         raise SanicException(f"Update failed. {e}", status_code=404)
+#     cfg
+#     return json({"cfg": cfg})

src/flowerpower/cli/mqtt.py

@@ -1,6 +1,18 @@
 import typer
 
-from flowerpower.cli.utils import parse_dict_or_list_param
-from flowerpower.mqtt import MQTTClient
+from ..cli.utils import parse_dict_or_list_param
+from ..mqtt import MQTTClient
 
 app = typer.Typer(help="MQTT management commands")
+
+
+@app.command()
+def start_broker(
+    host: str = "localhost",
+    port: int = 1883,
+    username: str = None,
+    password: str = None,
+    log_level: str = "INFO",
+):
+    """Start an MQTT broker."""
+    MQTTClient.start_broker(host, port, username, password, log_level)

src/flowerpower/cli/pipeline.py

@@ -41,7 +41,9 @@ def run(
     base_dir: str | None = None,
     inputs: str | None = None,
     final_vars: str | None = None,
+    config: str | None = None,
     with_tracker: bool = False,
+    with_opentelemetry: bool = False,
     reload: bool = False,
     storage_options: str | None = None,
 ):
@@ -54,7 +56,9 @@ def run(
         base_dir: Base directory for the pipeline
         inputs: Input parameters as JSON, dict string, or key=value pairs
         final_vars: Final variables as JSON or list
-        with_tracker: Enable tracking
+        config: Config for the hamilton pipeline executor
+        with_tracker: Enable tracking with hamilton ui
+        with_opentelemetry: Enable OpenTelemetry tracing
         reload: Reload pipeline before running
         storage_options: Storage options as JSON, dict string, or key=value pairs
 
@@ -75,6 +79,7 @@ def run(
     pipeline run my_pipeline --storage-options 'endpoint=http://localhost,use_ssl=true'
     """
     parsed_inputs = parse_dict_or_list_param(inputs, "dict")
+    parsed_config = parse_dict_or_list_param(config, "dict")
     parsed_final_vars = parse_dict_or_list_param(final_vars, "list")
     parsed_storage_options = parse_dict_or_list_param(storage_options, "dict")
 
@@ -87,7 +92,9 @@ def run(
             executor=executor,
             inputs=parsed_inputs,
             final_vars=parsed_final_vars,
+            config=parsed_config,
             with_tracker=with_tracker,
+            with_opentelemetry=with_opentelemetry,
             reload=reload,
         )
 
@@ -99,7 +106,9 @@ def run_job(
     base_dir: str | None = None,
     inputs: str | None = None,
     final_vars: str | None = None,
+    config: str | None = None,
     with_tracker: bool = False,
+    with_opentelemetry: bool = False,
     reload: bool = False,
     storage_options: str | None = None,
 ):
@@ -112,7 +121,9 @@ def run_job(
         base_dir: Base directory for the pipeline
         inputs: Input parameters as JSON, dict string, or key=value pairs
         final_vars: Final variables as JSON or list
-        with_tracker: Enable tracking
+        config: Config for the hamilton pipeline executor
+        with_tracker: Enable tracking with hamilton ui
+        with_opentelemetry: Enable OpenTelemetry tracing
         reload: Reload pipeline before running
         storage_options: Storage options as JSON, dict string, or key=value pairs
 
@@ -133,6 +144,7 @@ def run_job(
     pipeline run-job 123 --storage-options 'endpoint=http://localhost,use_ssl=true'
     """
     parsed_inputs = parse_dict_or_list_param(inputs, "dict")
+    parsed_config = parse_dict_or_list_param(config, "dict")
     parsed_final_vars = parse_dict_or_list_param(final_vars, "list")
     parsed_storage_options = parse_dict_or_list_param(storage_options, "dict")
 
@@ -145,7 +157,9 @@ def run_job(
             executor=executor,
             inputs=parsed_inputs,
             final_vars=parsed_final_vars,
+            config=parsed_config,
             with_tracker=with_tracker,
+            with_opentelemetry=with_opentelemetry,
             reload=reload,
         )
 
@@ -157,7 +171,10 @@ def add_job(
     base_dir: str | None = None,
     inputs: str | None = None,
     final_vars: str | None = None,
+    config: str | None = None,
     with_tracker: bool = False,
+    with_opentelemetry: bool = False,
+    reload: bool = False,
     storage_options: str | None = None,
 ):
     """
@@ -169,7 +186,10 @@ def add_job(
         base_dir: Base directory for the pipeline
         inputs: Input parameters as JSON, dict string, or key=value pairs
         final_vars: Final variables as JSON or list
-        with_tracker: Enable tracking
+        config: Config for the hamilton pipeline executor
+        with_tracker: Enable tracking with hamilton ui
+        with_opentelemetry: Enable OpenTelemetry tracing
+        reload: Reload pipeline before running
         storage_options: Storage options as JSON, dict string, or key=value pairs
 
     Examples:
@@ -189,6 +209,7 @@ def add_job(
     pipeline add-job my_pipeline --storage-options 'endpoint=http://localhost,use_ssl=true'
     """
     parsed_inputs = parse_dict_or_list_param(inputs, "dict")
+    parsed_config = parse_dict_or_list_param(config, "dict")
     parsed_final_vars = parse_dict_or_list_param(final_vars, "list")
     parsed_storage_options = parse_dict_or_list_param(storage_options, "dict")
 
@@ -203,7 +224,10 @@ def add_job(
             executor=executor,
             inputs=parsed_inputs,
             final_vars=parsed_final_vars,
+            config=parsed_config,
             with_tracker=with_tracker,
+            with_opentelemetry=with_opentelemetry,
+            reload=reload,
         )
 
 
@@ -215,7 +239,9 @@ def schedule(
     type: str = "cron",
     inputs: str | None = None,
     final_vars: str | None = None,
+    config: str | None = None,
     with_tracker: bool = False,
+    with_opentelemetry: bool = False,
     paused: bool = False,
     coalesce: str = "latest",
     misfire_grace_time: float | None = None,
@@ -232,13 +258,64 @@ def schedule(
     """
     Schedule a pipeline with various configuration options.
 
-    Supports flexible input parsing for inputs, final_vars, and storage_options
+    Args:
+        name: Name of the pipeline to schedule
+        executor: Executor to use
+        base_dir: Base directory for the pipeline
+        type: Type of schedule
+        inputs: Input parameters as JSON, dict string, or key=value pairs
+        final_vars: Final variables as JSON or list
+        config: Config for the hamilton pipeline executor
+        with_tracker: Enable tracking with hamilton ui
+        with_opentelemetry: Enable OpenTelemetry tracing
+        paused: Start the job in paused state
+        coalesce: Coalesce policy
+        misfire_grace_time: Misfire grace time
+        max_jitter: Maximum jitter
+        max_running_jobs: Maximum running jobs
+        conflict_policy: Conflict policy
+        crontab: Crontab expression
+        cron_params: Cron parameters as JSON or key=value pairs
+        interval_params: Interval parameters as JSON or key=value pairs
+        calendarinterval_params: Calendar interval parameters as JSON or key=value pairs
+        date_params: Date parameters as JSON or key=value pairs
+        storage_options: Storage options as JSON, dict string, or key=value pairs
+
+    Examples:
+    # JSON inputs
+    pipeline schedule my_pipeline --inputs '{"key": "value"}'
+
+    # Dict string inputs
+    pipeline schedule my_pipeline --inputs "{'key': 'value'}"
+
+    # Key-value pair inputs
+    pipeline schedule my_pipeline --inputs 'key1=value1,key2=value2'
+
+    # List final vars
+    pipeline schedule my_pipeline --final-vars '["var1", "var2"]'
+
+    # Storage options
+    pipeline schedule my_pipeline --storage-options 'endpoint=http://localhost,use_ssl=true'
+
+    # Cron schedule
+    pipeline schedule my_pipeline --type cron --crontab '0 0 * * *'
+
+    # Interval schedule
+    pipeline schedule my_pipeline --type interval --interval_params minutes=1
+
+    # Calendar interval schedule
+    pipeline schedule my_pipeline --type calendarinterval --calendarinterval_params month=5
+
+    # Date schedule
+    pipeline schedule my_pipeline --type date --date_params run_date='2021-01-01 12:00:01'
+
     """
     if get_schedule_manager is None:
         raise ValueError("APScheduler not installed. Please install it first.")
 
     # Parse inputs
     parsed_inputs = parse_dict_or_list_param(inputs, "dict")
+    parsed_config = parse_dict_or_list_param(config, "dict")
     parsed_final_vars = parse_dict_or_list_param(final_vars, "list")
     parsed_storage_options = parse_dict_or_list_param(storage_options, "dict")
 
@@ -278,7 +355,9 @@ def schedule(
             type=type,
             inputs=parsed_inputs,
             final_vars=parsed_final_vars,
+            config=parsed_config,
             with_tracker=with_tracker,
+            with_opentelemetry=with_opentelemetry,
             paused=paused,
             coalesce=coalesce,
             misfire_grace_time=misfire_grace_time,
@@ -347,7 +426,7 @@ def delete(
         base_dir=base_dir,
         storage_options=parsed_storage_options or {},
     ) as pipeline:
-        pipeline.delete(cfg=cfg,module=module)
+        pipeline.delete(cfg=cfg, module=module)
 
 
 @app.command()
@@ -448,30 +527,3 @@ def show_summary(
         storage_options=parsed_storage_options or {},
     ) as manager:
         manager.show_summary(name=name, cfg=cfg, module=module)
-
-
-# @app.command()
-# def get_summary(
-#     name: str,
-#     config: bool = True,
-#     module: bool = True,
-#     base_dir: str | None = None,
-#     storage_options: str | None = None,
-# ):
-#     """
-#     Get the summary of the specified pipeline.
-
-#     Args:
-#         name: Name of the pipeline to get
-#         base_dir: Base directory for the pipeline
-
-#     Examples:
-#     pipeline get-summary my_pipeline
-#     """
-#     parsed_storage_options = parse_dict_or_list_param(storage_options, "dict")
-
-#     with PipelineManager(
-#         base_dir=base_dir,
-#         storage_options=parsed_storage_options or {},
-#     ) as manager:
-#         manager.get_summary(name=name)

src/flowerpower/http/models/pipeline.py

@@ -6,6 +6,7 @@ class PipelineRun(BaseModel):
     inputs: dict | None = None
     final_vars: list | None = None
     executor: str | None = None
+    config: dict | None = None
     with_tracker: bool = None
     with_opentelemetry: bool = None
 

src/flowerpower/mqtt.py

@@ -256,6 +256,7 @@ def start_pipeline_listener(
         topic: str | None = None,
         inputs: dict | None = None,
         final_vars: list | None = None,
+        config: dict | None = None,
         executor: str | None = None,
         with_tracker: bool | None = None,
         with_opentelemetry: bool | None = None,
@@ -276,6 +277,7 @@ def start_pipeline_listener(
             topic: MQTT topic to listen to
             inputs: Inputs for the pipeline
             final_vars: Final variables for the pipeline
+            config: Configuration for the pipeline driver
             executor: Executor to use for the pipeline
             with_tracker: Use tracker for the pipeline
             with_opentelemetry: Use OpenTelemetry for the pipeline
@@ -306,6 +308,7 @@ def on_message(client, userdata, msg):
                             inputs=inputs,
                             final_vars=final_vars,
                             executor=executor,
+                            config=config,
                             with_tracker=with_tracker,
                             with_opentelemetry=with_opentelemetry,
                             reload=reload,
@@ -317,6 +320,7 @@ def on_message(client, userdata, msg):
                             inputs=inputs,
                             final_vars=final_vars,
                             executor=executor,
+                            config=config,
                             with_tracker=with_tracker,
                             with_opentelemetry=with_opentelemetry,
                             reload=reload,
@@ -372,6 +376,7 @@ def start_pipeline_listener(
     topic: str | None = None,
     inputs: dict | None = None,
     final_vars: list | None = None,
+    config: dict | None = None,
     executor: str | None = None,
     with_tracker: bool | None = None,
     with_opentelemetry: bool | None = None,
@@ -392,6 +397,7 @@ def start_pipeline_listener(
         topic: MQTT topic to listen to
         inputs: Inputs for the pipeline
         final_vars: Final variables for the pipeline
+        config: Configuration for the pipeline driver
         executor: Executor to use for the pipeline
         with_tracker: Use tracker for the pipeline
         with_opentelemetry: Use OpenTelemetry for the pipeline
@@ -410,6 +416,7 @@ def start_pipeline_listener(
         topic=topic,
         inputs=inputs,
         final_vars=final_vars,
+        config=config,
         executor=executor,
         with_tracker=with_tracker,
         with_opentelemetry=with_opentelemetry,