Enhance pipeline functionality by serializing final variables and updating optional dependencies in pyproject.toml

Author: legout

pyproject.toml

@@ -34,10 +34,18 @@ flowerpower = "flowerpower.cli:app"
 
 [project.optional-dependencies]
 # dask = ["dask[complete]>=2024.7.1"]
-filesystem-ext = [
-  "pyarrow>=18.1.0",
-  "polars>=1.15.0", #"duckdb>=1.1.3",  #"datafusion>=42.0.0",
+#filesystem-ext = [
+#  "pyarrow>=18.1.0",
+#  "polars>=1.15.0", #"duckdb>=1.1.3",  #"datafusion>=42.0.0",
+#  "orjson>=3.10.12",
+#]
+io = [
+  "datafusion>=43.1.0",
+  "duckdb>=1.1.3",
   "orjson>=3.10.12",
+  "polars>=1.15.0",
+  "pyarrow>=18.1.0",
+  "pydala2>=0.9.4.5",
 ]
 mongodb = ["pymongo>=4.7.2"]
 mqtt = ["paho-mqtt>=2.1.0", "orjson>=3.10.11"]

src/flowerpower/http/api/pipeline.py

@@ -45,8 +45,8 @@ async def run(
         ) as pipeline:
             final_vars = pipeline.run(**body.model_dump())
 
-        # final_vars = {k: dill.dumps(v) for k, v in final_vars.items()}
-        # return bytes(final_vars)
+        final_vars = {k: dill.dumps(v) for k, v in final_vars.items()}
+        return bytes(final_vars)
         return json({"status": "success", "message": "Pipeline ran successfully"})
     except Exception as e:
         raise SanicException(str(e))

src/flowerpower/http/api/scheduler.py

@@ -1,7 +1,7 @@
 import uuid
 
 import dill
-from sanic import Blueprint 
+from sanic import Blueprint
 from sanic.exceptions import SanicException
 from sanic.response import json, raw
 from sanic_ext import openapi
@@ -11,7 +11,7 @@
     SchedulerModify,
     SchedulerDelete,
     SchedulerList,
-    SchedulerInfo
+    SchedulerInfo,
 )
 from ..utils import deserialize_and_validate
 
@@ -62,16 +62,15 @@ async def job_result(request, job_id: str) -> raw:
 
 
 @bp.get("/schedules")
-@openapi.summary("List all schedules") 
+@openapi.summary("List all schedules")
 @openapi.description("Get a list of all pipeline schedules")
 @openapi.body({"application/json": SchedulerList}, required=False)
 @openapi.response(200, {"application/json": dict})
 async def schedules(request) -> json:
     try:
         body = await deserialize_and_validate(SchedulerList, query=request.args)
         schedules = request.app.ctx.scheduler.get_schedules(
-            pattern=body.pattern, 
-            as_dict=True
+            pattern=body.pattern, as_dict=True
         )
         return json({"schedules": schedules})
     except Exception as e:
@@ -87,7 +86,7 @@ async def schedules(request) -> json:
 @openapi.response(404, {"application/json": dict})
 async def schedule(request, schedule_id: str) -> json:
     try:
-        body = await deserialize_and_validate(SchedulerInfo, body=request.json)
+        # body = await deserialize_and_validate(SchedulerInfo, body=request.json)
         if schedule_id not in [s.id for s in request.app.ctx.scheduler.get_schedules()]:
             raise SanicException("Schedule not found", status_code=404)
         schedule = request.app.ctx.scheduler.get_schedule(schedule_id, as_dict=True)
@@ -113,15 +112,15 @@ async def add_schedule(request) -> json:
 @bp.patch("/schedule/<schedule_id>")
 @openapi.summary("Modify schedule")
 @openapi.description("Modify an existing pipeline schedule")
-@openapi.parameter("schedule_id", str, required=True) 
+@openapi.parameter("schedule_id", str, required=True)
 @openapi.body({"application/json": SchedulerModify}, required=True)
 @openapi.response(200, {"application/json": dict})
 @openapi.response(404, {"application/json": dict})
 async def modify_schedule(request, schedule_id: str) -> json:
     try:
         if schedule_id not in [s.id for s in request.app.ctx.scheduler.get_schedules()]:
             raise SanicException("Schedule not found", status_code=404)
-        
+
         body = await deserialize_and_validate(SchedulerModify, body=request.json)
         request.app.ctx.scheduler.modify_schedule(schedule_id, **body.model_dump())
         return json({"status": "success"})
@@ -138,7 +137,7 @@ async def modify_schedule(request, schedule_id: str) -> json:
 @openapi.response(404, {"application/json": dict})
 async def remove_schedule(request, schedule_id: str) -> json:
     try:
-        body = await deserialize_and_validate(SchedulerDelete, body=request.json)
+        # body = await deserialize_and_validate(SchedulerDelete, body=request.json)
         if schedule_id not in [s.id for s in request.app.ctx.scheduler.get_schedules()]:
             raise SanicException("Schedule not found", status_code=404)
         request.app.ctx.scheduler.remove_schedule(schedule_id)

src/flowerpower/io/base.py

@@ -121,6 +121,7 @@ def _update_fs(self):
             dirfs=False,
         )
 
+    @property
     def _glob_path(self):
         if isinstance(self.path, list):
             return
@@ -140,7 +141,7 @@ def list_files(self):
         if isinstance(self.path, list):
             return self.path
 
-        glob_path = self._glob_path()
+        glob_path = self._glob_path
         return self.fs.glob(glob_path)
 
 
@@ -180,18 +181,16 @@ class BaseFileLoader(BaseFileIO):
 
     include_file_path: bool = False
     concat: bool = True
+    batch_size: int | None = None
     conn: duckdb.DuckDBPyConnection | None = None
     ctx: datafusion.SessionContext | None = None
+    jsonlines: bool | None = None
 
     def _load(self, **kwargs):
-        if "include_file_path" not in kwargs:
-            self.include_file_path = kwargs.get(
-                "include_file_path", self.include_file_path
-            )
-        if "concat" not in kwargs:
-            self.concat = kwargs.get("concat", self.concat)
+        self.include_file_path = kwargs.get("include_file_path", self.include_file_path)
+        self.concat = kwargs.get("concat", self.concat)
 
-        if not hasattr(self, "_data"):
+        if not hasattr(self, "_data") or self._data is None:
             self._data = self.fs.read_files(
                 self._glob_path,
                 self.format,
@@ -208,8 +207,11 @@ def _load(self, **kwargs):
                 reload = True
             if isinstance(self._data, list) and self.concat:
                 self._data = pl.concat(self._data, how="diagonal_relaxed")
+
             if reload:
-                self._load(
+                self._data = self.fs.read_files(
+                    self._glob_path,
+                    self.format,
                     include_file_path=self.include_file_path,
                     concat=self.concat,
                     jsonlines=self.jsonlines or None,