Add a function to save state across tasks

Author: ctrueden

src/appose/python_worker.py

@@ -51,7 +51,10 @@
 
 
 class Task:
-    def __init__(self, uuid: str, script: str, inputs: Args | None = None) -> None:
+    def __init__(
+        self, worker: "Worker", uuid: str, script: str, inputs: Args | None = None
+    ) -> None:
+        self._worker = worker
         self._uuid = uuid
         self._script = script
         self._inputs = inputs
@@ -62,6 +65,9 @@ def __init__(self, uuid: str, script: str, inputs: Args | None = None) -> None:
         self.outputs = {}
         self.cancel_requested = False
 
+    def export(self, **kwargs):
+        self._worker.exports.update(kwargs)
+
     def update(
         self,
         message: str | None = None,
@@ -96,6 +102,7 @@ def _run(self) -> None:
         try:
             # Populate script bindings.
             binding = {"task": self}
+            binding.update(self._worker.exports)
             if self._inputs is not None:
                 binding.update(self._inputs)
 
@@ -184,6 +191,7 @@ class Worker:
     def __init__(self):
         self.tasks = {}
         self.queue: list[Task] = []
+        self.exports = {}
         self.running = True
 
         # Flag this process as a worker, not a service.
@@ -223,7 +231,7 @@ def _process_input(self) -> None:
                     script = request.get("script")
                     inputs = request.get("inputs")
                     queue = request.get("queue")
-                    task = Task(uuid, script, inputs)
+                    task = Task(self, uuid, script, inputs)
                     self.tasks[uuid] = task
                     if queue == "main":
                         # Add the task to the main thread queue.

tests/test_appose.py

@@ -57,10 +57,11 @@
 task.outputs["result"] = time
 """
 
-sqrt_import = """
+calc_sqrt_python = """
 from math import sqrt
 def sqrt_age(age):
     return sqrt(age)
+task.export(sqrt_age=sqrt_age)
 task.outputs["result"] = sqrt_age(age)
 """
 
@@ -103,16 +104,22 @@ def test_service_startup_failure():
         ) == str(e)
 
 
-def test_scope():
+def test_scope_python():
     env = appose.system()
     with env.python() as service:
         maybe_debug(service)
-        task = service.task(sqrt_import, {"age": 100})
+        task = service.task(calc_sqrt_python, {"age": 100})
         task.wait_for()
         assert TaskStatus.COMPLETE == task.status
         result = round(task.outputs.get("result"))
         assert result == 10
 
+        task = service.task("task.outputs['result'] = sqrt_age(age)", {"age": 81})
+        task.wait_for()
+        assert TaskStatus.COMPLETE == task.status
+        result = round(task.outputs.get("result"))
+        assert result == 9
+
 
 def test_main_thread_queue_groovy():
     env = appose.system()