chore: Update version to 0.10.7 and add CACHE_DIR setting; refactor filesystem handling in job queue and pipeline manager

Author: legout

pyproject.toml

@@ -4,7 +4,7 @@ description = "A simple workflow framework. Hamilton + APScheduler = FlowerPower
 authors = [{ name = "Volker L.", email = "[email protected]" }]
 readme = "README.md"
 requires-python = ">= 3.11"
-version = "0.10.6.4"
+version = "0.10.7"
 keywords = [
   "hamilton",
   "workflow",

src/flowerpower/fs/base.py

@@ -220,7 +220,7 @@ def __getattribute__(self, item):
             # new items
             "size",
             "glob",
-            "sync",
+            "sync_cache",
             # previous
             "load_cache",
             "_open",

src/flowerpower/job_queue/base.py

@@ -24,7 +24,7 @@
 from ..cfg import ProjectConfig
 from ..fs import AbstractFileSystem, get_filesystem
 # from ..utils.misc import update_config_from_dict
-from ..settings import BACKEND_PROPERTIES
+from ..settings import BACKEND_PROPERTIES, CACHE_DIR, CONFIG_DIR, PIPELINES_DIR
 
 
 class BackendType(str, Enum):
@@ -350,12 +350,21 @@ def __init__(
         self._storage_options = storage_options or {}
         self._backend = backend
         self._type = type
-        self._pipelines_dir = kwargs.get("pipelines_dir", "pipelines")
-        self._conf_dir = "conf"
+        self._pipelines_dir = kwargs.get("pipelines_dir", PIPELINES_DIR)
+        self._conf_dir = CONFIG_DIR
 
-        if fs is None:
-            fs = get_filesystem(self._base_dir, **(self._storage_options or {}))
+        if storage_options is not None:
+            cached = True
+            cache_storage = posixpath.join(posixpath.expanduser(CACHE_DIR), self._base_dir.split("://")[-1])
+            posixpath.makedirs(cache_storage, exist_ok=True)
+        else:
+            cached = False
+            cache_storage = None
+        if not fs:
+            fs = get_filesystem(self._base_dir, storage_options=storage_options, cached=cached, cache_storage=cache_storage)
         self._fs = fs
+        if cached:
+            self._fs.sync()
 
         self._add_modules_path()
         self._load_config()
@@ -378,11 +387,17 @@ def _add_modules_path(self):
             None
         """
         if self._fs.is_cache_fs:
-            self._fs.sync()
+            self._fs.sync_cache()
+            project_path = self._fs.mapper.directory
+            modules_path = posixpath.join(project_path, self._pipelines_dir)
+
+        else:
+            # Use the base directory directly if not using cache
+            project_path = self._fs.path
+            modules_path = posixpath.join(project_path, self._pipelines_dir)
 
-        if self._fs.path not in sys.path:
-            sys.path.insert(0, self._fs.path)
+        if project_path not in sys.path:
+            sys.path.insert(0, project_path)
 
-        modules_path = posixpath.join(self._fs.path, self._pipelines_dir)
         if modules_path not in sys.path:
-            sys.path.insert(0, modules_path)
+            sys.path.insert(0, modules_path)

src/flowerpower/pipeline/manager.py

@@ -125,9 +125,18 @@ def __init__(
 
         self._base_dir = base_dir or str(Path.cwd())
         self._storage_options = storage_options
+        if storage_options is not None:
+            cached = True
+            cache_storage = posixpath.join(posixpath.expanduser(settings.CACHE_DIR), self._base_dir.split("://")[-1])
+            posixpath.makedirs(cache_storage, exist_ok=True)
+        else:
+            cached = False
+            cache_storage = None
         if not fs:
-            fs = get_filesystem(self._base_dir, storage_options=storage_options)
+            fs = get_filesystem(self._base_dir, storage_options=storage_options, cached=cached, cache_storage=cache_storage)
         self._fs = fs
+        if cached:
+            self._fs.sync()
 
         # Store overrides for ProjectConfig loading
         self._cfg_dir = cfg_dir or settings.CONFIG_DIR
@@ -260,12 +269,18 @@ def _add_modules_path(self) -> None:
             >>> import my_pipeline  # Now importable
         """
         if self._fs.is_cache_fs:
-            self._fs.sync()
+            self._fs.sync_cache()
+            project_path = self._fs.mapper.directory
+            modules_path = posixpath.join(project_path, self._pipelines_dir)
+
+        else:
+            # Use the base directory directly if not using cache
+            project_path = self._fs.path
+            modules_path = posixpath.join(project_path, self._pipelines_dir)
 
-        if self._fs.path not in sys.path:
-            sys.path.insert(0, self._fs.path)
+        if project_path not in sys.path:
+            sys.path.insert(0, project_path)
 
-        modules_path = posixpath.join(self._fs.path, self._pipelines_dir)
         if modules_path not in sys.path:
             sys.path.insert(0, modules_path)
 

src/flowerpower/settings/general.py

@@ -3,3 +3,4 @@
 PIPELINES_DIR = os.getenv("FP_PIPELINES_DIR", "pipelines")
 CONFIG_DIR = os.getenv("FP_CONFIG_DIR", "conf")
 HOOKS_DIR = os.getenv("FP_HOOKS_DIR", "hooks")
+CACHE_DIR = os.getenv("FP_CACHE_DIR", "~/.flowerpower/cache")