Persistent storage

Author: codeSamuraii

docs/FEATURES.md

@@ -210,6 +210,28 @@ offwork worker --backend redis://localhost:6379 --sandbox   # run with isolation
 
 See [Sandbox](SANDBOX.md) for configuration and management.
 
+## Persistent storage
+
+When a worker provides a persistent mount (for example the hosted broker gives each user a private volume), tasks reach it through `offwork.storage_path()`:
+
+```python
+import offwork
+
+@offwork.task
+def cache_model(url: str) -> str:
+    dest = offwork.storage_path("models", "weights.bin")
+    dest.parent.mkdir(parents=True, exist_ok=True)   # subdirectories are yours to create
+    if not dest.exists():
+        download(url, dest)
+    return str(dest)
+```
+
+`storage_path()` returns the storage root, creating it if needed. Passing path
+parts joins them onto the root without creating them. The location comes from
+the `OFFWORK_STORAGE` environment variable, falling back to `./offwork-storage`
+when unset, so the same task code works locally and on a worker with a mounted
+volume.
+
 ## Signing
 
 Pre-shared token or PIN-based pairing + HMAC-SHA256 — workers reject untrusted or tampered tasks:

offwork/__init__.py

@@ -62,6 +62,7 @@
 from offwork.core.version import _VERSION
 from offwork.core.progress import ProgressInfo
 from offwork.core.progress import progress as progress
+from offwork.core.storage import storage_path as storage_path
 from offwork.core._timeout import TimeoutIn, WaitForever, ReturnImmediately
 from offwork.worker.remote import serve, connect, disconnect, _ConnectionContext
 from offwork.worker.result import Result, ResultEnvelope
@@ -130,6 +131,7 @@ def pack(func: Callable[..., object], *args: Any, **kwargs: Any) -> Task:
     "install_package_as",
     "worker_only_import",
     "progress",
+    "storage_path",
     # Timeout types
     "TimeoutIn",
     "WaitForever",

offwork/core/storage.py

@@ -0,0 +1,37 @@
+"""Persistent per-worker storage path.
+
+A worker may expose a directory that survives between task executions
+(e.g. a per-user volume mounted into the worker container). Tasks read
+its location through :func:`storage_path` rather than hard-coding a path,
+so the same code runs locally and on a hosted broker.
+"""
+
+import os
+from pathlib import Path
+
+#: Environment variable a worker sets to advertise its persistent storage
+#: directory. Hosted brokers point this at a per-user volume.
+STORAGE_ENV = "OFFWORK_STORAGE"
+
+
+def storage_path(*parts: str) -> Path:
+    """Return the persistent storage directory, creating it if needed.
+
+    Files written under the returned directory survive between task
+    executions on workers that provide persistent storage. When no
+    storage is configured (``OFFWORK_STORAGE`` unset) this falls back to
+    ``./offwork-storage`` in the worker's working directory.
+
+    Optional *parts* are joined onto the base path::
+
+        offwork.storage_path()                 # the storage root
+        offwork.storage_path("models")         # a subdirectory
+        offwork.storage_path("data", "in.csv") # a file path
+
+    The storage root is created on access. The returned path is absolute.
+    Joined *parts* are not created — call ``.mkdir(parents=True)`` on the
+    result yourself if you need a subdirectory.
+    """
+    base = Path(os.environ.get(STORAGE_ENV, "offwork-storage")).expanduser()
+    base.mkdir(parents=True, exist_ok=True)
+    return base.joinpath(*parts).resolve()

tests/test_storage.py

@@ -0,0 +1,35 @@
+"""Tests for offwork.storage_path()."""
+
+from pathlib import Path
+
+import offwork
+from offwork.core.storage import STORAGE_ENV
+
+
+def test_storage_path_uses_env(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv(STORAGE_ENV, str(tmp_path))
+    p = offwork.storage_path()
+    assert p == tmp_path.resolve()
+    assert p.is_dir()
+
+
+def test_storage_path_joins_parts(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv(STORAGE_ENV, str(tmp_path))
+    p = offwork.storage_path("models", "weights.bin")
+    assert p == (tmp_path / "models" / "weights.bin").resolve()
+    # Only the storage root is created, not joined subdirectories.
+    assert tmp_path.is_dir()
+    assert not p.parent.exists()
+
+
+def test_storage_path_default_when_unset(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.delenv(STORAGE_ENV, raising=False)
+    monkeypatch.chdir(tmp_path)
+    p = offwork.storage_path()
+    assert p == (tmp_path / "offwork-storage").resolve()
+    assert p.is_dir()
+
+
+def test_storage_path_in_public_api() -> None:
+    assert "storage_path" in offwork.__all__
+    assert callable(offwork.storage_path)