Refactor pipeline configuration and update dependencies; enhance project and scheduler models, and clean up unused code

Author: legout

examples/hello-world/conf/pipelines/hello_world.yml

@@ -4,18 +4,17 @@ params:
   spend_zero_mean:
     offset: 0
 run:
+  config: {}
   executor: null
   final_vars:
-    [
-      spend,
-      signups,
-      avg_x_wk_spend,
-      spend_per_signup,
-      spend_zero_mean_unit_variance,
-    ]
+  - spend
+  - signups
+  - avg_x_wk_spend
+  - spend_per_signup
+  - spend_zero_mean_unit_variance
   inputs: {}
-  with_tracker: false
   with_opentelemetry: false
+  with_tracker: false
 schedule:
   run:
     coalesce: latest
@@ -27,41 +26,42 @@ schedule:
     misfire_grace_time: null
     paused: false
   trigger:
-    type_: cron
+    calendar:
+      days: null
+      end_date: null
+      hour: null
+      minute: null
+      months: null
+      second: null
+      start_date: null
+      timezone: null
+      weeks: null
+      years: null
     cron:
-      crontab: "* * * * *"
-      year: null
-      month: null
-      week: null
+      crontab: '* * * * *'
+      day: null
       day_of_week: null
+      end_time: null
       hour: null
       minute: null
+      month: null
       second: null
       start_time: null
-      end_time: null
       timezone: null
+      week: null
+      year: null
+    date:
+      run_time: null
     interval:
-      weeks: null
       days: null
+      end_time: null
       hours: null
+      microseconds: null
       minutes: null
       seconds: null
-      microseconds: null
       start_time: null
-      end_time: null
-    calendar:
-      years: null
-      months: null
       weeks: null
-      days: null
-      hour: null
-      minute: null
-      second: null
-      start_date: null
-      end_date: null
-      timezone: null
-    date:
-      run_time: null
+    type_: cron
 tracker:
   dag_name: null
   project_id: null

examples/hello-world/conf/project.yml

@@ -2,20 +2,18 @@ name: hello-world
 open_telemetry:
   host: localhost
   port: 6381
+tracker:
+  api_key: null
+  api_url: http://localhost:8241
+  ui_url: http://localhost:8242
+  username: null
 worker:
   cleanup_interval: 900
   data_store:
     type: sqlalchemy
     uri: sqlite+aiosqlite:///flowerpower.db
-    #uri: postgresql+asyncpg://postgres:password@localhost:5432/flowerpower
+    #uri: postgresql+asyncpg://postgres:password@localhost/flowerpower
   event_broker:
-    {}
-    #host: localhost
-    #port: 6379
-    #type: redis
+    type: redis
+    uri: redis://localhost:6379
   max_concurrent_jobs: 100
-tracker:
-  api_key: null
-  api_url: http://localhost:8241
-  ui_url: http://localhost:8242
-  username: null

examples/hello-world/pipelines/hello_world.py

@@ -13,6 +13,8 @@
     Path(__file__).parents[1], pipeline_name="hello_world"
 ).pipeline.h_params
 
+print(Path(__file__).parents[1])
+
 
 def spend() -> pd.Series:
     """Returns a series of spend data."""

pyproject.toml

@@ -52,11 +52,11 @@ opentelemetry = [
 ray = ["ray>=2.34.0"]
 redis = ["redis>=5.0.4"]
 scheduler = [
-  "aiosqlite>=0.20.0",
   "greenlet>=3.0.3",
   "asyncpg>=0.29.0",
   "sqlalchemy>=2.0.30",
   "apscheduler>=4.0.0a5", #"apscheduler @ git+https://github.com/agronholm/apscheduler",
+  "aiosqlite>=0.21.0",
 ]
 tui = ["textual>=0.85.2"]
 ui = ["sf-hamilton-ui>=0.0.11"]
@@ -109,5 +109,8 @@ dev-dependencies = [
   "obstore>=0.3.0",
   "pytest>=8.3.4",
   "mocker>=1.1.1",
+  "marimo>=0.10.19",
+  "panel>=1.6.0",
+  "ipywidgets>=8.1.5",
 ]
 package = true

src/flowerpower/cli/pipeline.py

@@ -254,6 +254,7 @@ def schedule(
     calendarinterval_params: str | None = None,
     date_params: str | None = None,
     storage_options: str | None = None,
+    overwrite: bool = False,
 ):
     """
     Schedule a pipeline with various configuration options.
@@ -280,6 +281,7 @@ def schedule(
         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
+        overwrite: Overwrite existing schedule
 
     Examples:
     # JSON inputs
@@ -364,12 +366,98 @@ def schedule(
             max_jitter=max_jitter,
             max_running_jobs=max_running_jobs,
             conflict_policy=conflict_policy,
+            overwrite=overwrite,
             **kwargs,
         )
 
     logger.info(f"Job {id_} scheduled.")
 
 
+@app.command()
+def schedule_all(
+    executor: str = "local",
+    base_dir: str | None = None,
+    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,
+    max_jitter: float | None = None,
+    max_running_jobs: int | None = None,
+    conflict_policy: str = "do_nothing",
+    crontab: str | None = None,
+    cron_params: str | None = None,
+    interval_params: str | None = None,
+    calendarinterval_params: str | None = None,
+    date_params: str | None = None,
+    storage_options: str | None = None,
+    overwrite: bool = False,
+):
+    """
+    Schedule all pipelines using the pipeline specific configurations (`conf/pipelines/<name>.yml`).
+
+    Args:
+        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
+        overwrite: Overwrite existing schedule
+
+    Examples:
+    pipeline schedule-all
+    """
+    if get_schedule_manager is None:
+        raise ValueError("APScheduler not installed. Please install it first.")
+
+    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.schedule_all(
+            executor=executor,
+            type=type,
+            inputs=inputs,
+            final_vars=final_vars,
+            config=config,
+            with_tracker=with_tracker,
+            with_opentelemetry=with_opentelemetry,
+            paused=paused,
+            coalesce=coalesce,
+            misfire_grace_time=misfire_grace_time,
+            max_jitter=max_jitter,
+            max_running_jobs=max_running_jobs,
+            conflict_policy=conflict_policy,
+            overwrite=overwrite,
+            crontab=crontab,
+            cron_params=cron_params,
+            interval_params=interval_params,
+            calendarinterval_params=calendarinterval_params,
+            date_params=date_params,
+        )
+
+
 @app.command()
 def new(
     name: str,

src/flowerpower/filesystem/base.py

@@ -285,10 +285,6 @@ def get_filesystem(
 
     pp = infer_storage_options(str(path) if isinstance(path, Path) else path)
     protocol = pp.get("protocol")
-    host = pp.get("host", "")
-    path = pp.get("path", "")
-    if host and host not in path:
-        path = posixpath.join(host, path)
 
     if protocol == "file" or protocol == "local":
         fs = filesystem(protocol)
@@ -298,6 +294,11 @@ def get_filesystem(
             fs.is_cache_fs = False
         return fs
 
+    host = pp.get("host", "")
+    path = pp.get("path", "").lstrip("/")
+    if len(host) and host not in path:
+        path = posixpath.join(host, path)
+
     if isinstance(storage_options, dict):
         storage_options = storage_options_from_dict(protocol, storage_options)
 

src/flowerpower/http/api/cfg.py

@@ -5,7 +5,7 @@
 from sanic.response import json
 from sanic_ext import openapi, validate
 
-from ...cfg import PipelineConfig
+from ...cfg import PipelineConfig, ProjectConfig
 
 bp = Blueprint("api_flowerpower_cfg", url_prefix="api/cfg")
 
@@ -17,6 +17,18 @@ async def get_project(request) -> json:
     return json({"cfg": cfg})
 
 
+@bp.post("/project")
+@validate(json=ProjectConfig)
+async def update_project(request, body: ProjectConfig) -> json:
+    cfg = request.app.ctx.pipeline_manager.cfg.copy()
+    cfg.project.update(body.model_dump())
+    try:
+        cfg.save()
+    except NotImplementedError as e:
+        raise SanicException(f"Update failed. {e}", status_code=404)
+    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:
@@ -26,23 +38,15 @@ async def get_pipeline(request, pipeline_name) -> json:
 
 
 @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)
+    cfg = request.app.ctx.pipeline_manager.cfg.copy()
+    cfg.pipeline.update(body.model_dump())
     try:
-        cfg.to_yaml(
-            posixpath.join(
-                "pipelines",
-                pipeline_name + ".yml",
-            ),
-            fs=request.app.ctx.pipeline_manager.cfg.fs,
-        )
+        cfg.save()
     except NotImplementedError as e:
         raise SanicException(f"Update failed. {e}", status_code=404)
-    cfg
+
     return json({"cfg": cfg})