v0.0.3

Author: guangrei

syncio/Multiprocessing.py

@@ -1,6 +1,6 @@
 # -*-coding:utf8;-*-
 from multiprocessing import Process, Manager
-from syncio.Task import NewTask, TaskOutput
+from syncio.Task import NewTask, TaskOutput, TaskReturnException
 
 
 def gather(*args: NewTask) -> TaskOutput:
@@ -10,7 +10,10 @@ def gather(*args: NewTask) -> TaskOutput:
 
     def runner(tasker: NewTask, task_id: int) -> None:
         task_id += 1
-        result = tasker.func(*tasker.args, **tasker.kwargs)
+        try:
+            result = tasker.func(*tasker.args, **tasker.kwargs)
+        except BaseException as e:
+            result = TaskReturnException(e)
         ret[f"task_{task_id}"] = result
 
     for k, f in enumerate(args):

syncio/Task.py

@@ -2,6 +2,8 @@
 from abc import ABC, abstractmethod
 from typing import Callable, Tuple, List, Any, Iterator, Dict
 from typing_extensions import Self
+import traceback
+import pprint
 
 
 class NewTask(ABC):
@@ -17,6 +19,10 @@ def __init__(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
     def __call__(self, *args: Any, **kwargs: Any) -> Self:
         pass
 
+    @abstractmethod
+    def __repr__(self) -> str:
+        pass
+
 
 class create_task(NewTask):
     def __init__(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
@@ -29,6 +35,12 @@ def __call__(self, *args: Any, **kwargs: Any) -> Self:
         self.kwargs = kwargs
         return self
 
+    def __repr__(self) -> str:
+        if hasattr(self.func, "__name__"):
+            return f"<NewTask '{self.func.__name__}'>"
+        else:
+            return f"<NewTask '{type(self.func).__name__}'>"
+
 
 class TaskOutput(dict):  # type: ignore[type-arg]
     def __iter__(self) -> Iterator[Any]:
@@ -45,3 +57,25 @@ def __next__(self) -> Any:
         value: Any = self._output_list[self._index]
         self._index += 1
         return value
+
+    def __str__(self) -> str:
+        return pprint.pformat(self, indent=2)
+
+
+class TaskReturnException:
+    def __init__(self, e: BaseException) -> None:
+        self.exception: BaseException = e
+
+    def print_exception(self) -> None:
+        traceback.print_exception(
+            type(self.exception), self.exception, self.exception.__traceback__
+        )
+
+    def __repr__(self) -> str:
+        return f"<TaskReturnException '{self.exception.__class__.__name__}'>"
+
+    def __bool__(self) -> bool:
+        return False
+
+    def __len__(self) -> int:
+        return 0

syncio/Threading.py

@@ -1,6 +1,6 @@
 # -*-coding:utf8;-*-
 from threading import Thread
-from syncio.Task import NewTask, TaskOutput
+from syncio.Task import NewTask, TaskOutput, TaskReturnException
 
 
 def thread_gather(*args: NewTask) -> TaskOutput:
@@ -9,7 +9,10 @@ def thread_gather(*args: NewTask) -> TaskOutput:
 
     def runner(tasker: NewTask, task_id: int) -> None:
         task_id += 1
-        result = tasker.func(*tasker.args, **tasker.kwargs)
+        try:
+            result = tasker.func(*tasker.args, **tasker.kwargs)
+        except BaseException as e:
+            result = TaskReturnException(e)
         ret[f"task_{task_id}"] = result
 
     for k, f in enumerate(args):

syncio/__init__.py

@@ -1,6 +1,6 @@
 # -*-coding:utf8;-*-
-from syncio.Task import create_task
+from syncio.Task import create_task, TaskReturnException
 from syncio.Multiprocessing import gather
 from syncio.Threading import thread_gather
 
-__all__ = ["create_task", "gather", "thread_gather"]
+__all__ = ["create_task", "gather", "thread_gather", "TaskReturnException"]