[python] Directory utilities (#51)

Reusable subclasses of pathlib.Path representing requestor and provider
directory structures.

Bump package to v0.20.0.

Author: Wiezzel

python/golem_task_api/dirutils.py

@@ -0,0 +1,68 @@
+from pathlib import Path
+
+from golem_task_api import constants
+
+# Concrete path type appropriate for the platform (WindowsPath or PosixPath)
+FlavouredPath = type(Path())
+
+
+class RequestorTaskDir(FlavouredPath):
+
+    @property
+    def task_inputs_dir(self) -> Path:
+        return Path(self / constants.TASK_INPUTS_DIR)
+
+    @property
+    def task_outputs_dir(self) -> Path:
+        return Path(self / constants.TASK_OUTPUTS_DIR)
+
+    @property
+    def subtask_inputs_dir(self) -> Path:
+        return Path(self / constants.SUBTASK_INPUTS_DIR)
+
+    @property
+    def subtask_outputs_top_dir(self) -> Path:
+        return Path(self / constants.SUBTASK_OUTPUTS_DIR)
+
+    def subtask_outputs_dir(self, subtask_id: str) -> Path:
+        return self.subtask_outputs_top_dir / subtask_id
+
+    def prepare(self) -> None:
+        self.mkdir()
+        self.task_inputs_dir.mkdir()
+        self.task_outputs_dir.mkdir()
+        self.subtask_inputs_dir.mkdir()
+        self.subtask_outputs_top_dir.mkdir()
+
+
+class RequestorDir(FlavouredPath):
+
+    def task_dir(self, task_id: str) -> RequestorTaskDir:
+        return RequestorTaskDir(self / task_id)
+
+    def subtask_outputs_dir(self, task_id: str, subtask_id: str) -> Path:
+        return self.task_dir(task_id).subtask_outputs_dir(subtask_id)
+
+
+class ProviderTaskDir(FlavouredPath):
+
+    @property
+    def subtask_inputs_dir(self) -> Path:
+        return Path(self / constants.SUBTASK_INPUTS_DIR)
+
+    def subtask_dir(self, subtask_id: str) -> Path:
+        return Path(self / subtask_id)
+
+    def prepare(self) -> None:
+        self.mkdir()
+        self.subtask_inputs_dir.mkdir()
+
+
+class ProviderDir(FlavouredPath):
+
+    def task_dir(self, task_id: str) -> ProviderTaskDir:
+        return ProviderTaskDir(self / task_id)
+
+    def subtask_dir(self, task_id: str, subtask_id: str) -> Path:
+        return self.task_dir(task_id).subtask_dir(subtask_id)
+

python/golem_task_api/handlers.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import List, Tuple, Optional
 
+from golem_task_api.dirutils import ProviderTaskDir, RequestorTaskDir
 from golem_task_api.enums import VerifyResult
 from golem_task_api.structs import Subtask, Task
 
@@ -38,7 +39,7 @@ class RequestorAppHandler:
     @abc.abstractmethod
     async def create_task(
             self,
-            task_work_dir: Path,
+            task_work_dir: RequestorTaskDir,
             max_subtasks_count: int,
             task_params: dict,
     ) -> Task:
@@ -47,23 +48,23 @@ async def create_task(
     @abc.abstractmethod
     async def next_subtask(
             self,
-            task_work_dir: Path,
+            task_work_dir: RequestorTaskDir,
             opaque_node_id: str,
      ) -> Optional[Subtask]:
         pass
 
     @abc.abstractmethod
     async def verify(
             self,
-            task_work_dir: Path,
+            task_work_dir: RequestorTaskDir,
             subtask_id: str,
     ) -> Tuple[VerifyResult, Optional[str]]:
         pass
 
     @abc.abstractmethod
     async def discard_subtasks(
             self,
-            task_work_dir: Path,
+            task_work_dir: RequestorTaskDir,
             subtask_ids: List[str],
     ) -> List[str]:
         pass
@@ -81,7 +82,7 @@ class ProviderAppHandler:
     @abc.abstractmethod
     async def compute(
             self,
-            task_work_dir: Path,
+            task_work_dir: ProviderTaskDir,
             subtask_id: str,
             subtask_params: dict,
     ) -> Path:

python/golem_task_api/server.py

@@ -8,6 +8,7 @@
 from grpclib import server
 from grpclib.health.service import Health
 
+from golem_task_api.dirutils import ProviderTaskDir, RequestorDir
 from golem_task_api.proto.golem_task_api_grpc import (
     ProviderAppBase,
     RequestorAppBase,
@@ -63,15 +64,14 @@ def __init__(
             handler: RequestorAppHandler,
             lifecycle: AppLifecycleHandler,
     ) -> None:
-        self._work_dir = work_dir
+        self._work_dir = RequestorDir(work_dir)
         self._handler = handler
         self._lifecycle = lifecycle
 
     @forward_exceptions()
     async def CreateTask(self, stream):
         request: CreateTaskRequest = await stream.recv_message()
-        task_id = request.task_id
-        task_work_dir = self._work_dir / task_id
+        task_work_dir = self._work_dir.task_dir(request.task_id)
         max_subtasks_count = request.max_subtasks_count
         task_params = json.loads(request.task_params_json)
         task = await self._handler.create_task(
@@ -87,9 +87,8 @@ async def CreateTask(self, stream):
     @forward_exceptions()
     async def NextSubtask(self, stream):
         request: NextSubtaskRequest = await stream.recv_message()
-        task_id = request.task_id
+        task_work_dir = self._work_dir.task_dir(request.task_id)
         opaque_node_id = request.opaque_node_id
-        task_work_dir = self._work_dir / task_id
         reply = NextSubtaskReply()
         subtask = await self._handler.next_subtask(
             task_work_dir, opaque_node_id)
@@ -104,9 +103,8 @@ async def NextSubtask(self, stream):
     @forward_exceptions()
     async def Verify(self, stream):
         request: VerifyRequest = await stream.recv_message()
-        task_id = request.task_id
+        task_work_dir = self._work_dir.task_dir(request.task_id)
         subtask_id = request.subtask_id
-        task_work_dir = self._work_dir / task_id
         result, reason = await self._handler.verify(task_work_dir, subtask_id)
         reply = VerifyReply()
         reply.result = result.value
@@ -117,9 +115,8 @@ async def Verify(self, stream):
     @forward_exceptions()
     async def DiscardSubtasks(self, stream):
         request: DiscardSubtasksRequest = await stream.recv_message()
-        task_id = request.task_id
+        task_work_dir = self._work_dir.task_dir(request.task_id)
         subtask_ids = request.subtask_ids
-        task_work_dir = self._work_dir / task_id
         discarded_subtask_ids = \
             await self._handler.discard_subtasks(task_work_dir, subtask_ids)
         reply = DiscardSubtasksReply()
@@ -137,7 +134,7 @@ async def RunBenchmark(self, stream):
     @forward_exceptions()
     async def HasPendingSubtasks(self, stream):
         request: HasPendingSubtasksRequest = await stream.recv_message()
-        task_work_dir = self._work_dir / request.task_id
+        task_work_dir = self._work_dir.task_dir(request.task_id)
         has_pending_subtasks = \
             await self._handler.has_pending_subtasks(task_work_dir)
         reply = HasPendingSubtasksReply()
@@ -159,7 +156,7 @@ def __init__(
             handler: ProviderAppHandler,
             lifecycle: AppLifecycleHandler,
     ) -> None:
-        self._work_dir = work_dir
+        self._work_dir = ProviderTaskDir(work_dir)
         self._handler = handler
         self._lifecycle = lifecycle