Add Direct Executor, Future callbacks function

docs/package/concurrent/.pages

@@ -0,0 +1,4 @@
+nav:
+    - ...
+
+title: concurrent

docs/package/concurrent/executor/.pages

@@ -0,0 +1,2 @@
+nav:
+    - pycommons.base.concurrent.executor: executor.md

docs/package/concurrent/executor/executor.md

@@ -0,0 +1 @@
+::: pycommons.base.concurrent.executor

docs/package/concurrent/future/.pages

@@ -0,0 +1,4 @@
+nav:
+    - pycommons.base.concurrent.future: future.md
+
+title: future

docs/package/concurrent/future/future.md

@@ -0,0 +1 @@
+::: pycommons.base.concurrent.future

docs/package/container/.pages

@@ -1,2 +1,4 @@
 nav:
     - pycommons.base.container: container.md
+
+title: container

docs/package/function/.pages

@@ -1,2 +1,4 @@
 nav:
     - pycommons.base.function: function.md
+
+title: function

pycommons/base/concurrent/executor/__init__.py

@@ -0,0 +1,3 @@
+from .direct import DirectExecutor
+
+__all__ = ["DirectExecutor"]

pycommons/base/concurrent/executor/direct.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from concurrent.futures import Executor, Future
+from typing import Callable, TypeVar, Any, ClassVar
+
+_P = TypeVar("_P")
+_T = TypeVar("_T")
+
+
+class DirectExecutor(Executor):
+    """
+    An Executor that runs the intended job in the same thread as that of the caller.
+    This is usually helpful when writing tests for background processes.
+    """
+
+    __instance__: ClassVar[DirectExecutor]
+    """
+    A class instance of the direct executor which can be used everywhere
+    for executing a task in the same thread as that of the caller.
+    """
+
+    @classmethod
+    def get_instance(cls) -> DirectExecutor:
+        """
+        Gets the singleton instance of `DirectExecutor`
+
+        Returns:
+            The singleton instance of `DirectExecutor`
+        """
+        return cls.__instance__
+
+    def submit(  # pylint: disable=W0221
+        self, fn: Callable[[Any], _T], *args: Any, **kwargs: Any
+    ) -> Future[_T]:  # pylint: disable=W0221
+        """
+        Submits a callable to run in the same thread as the caller.
+
+        Args:
+            fn: The callable
+            *args: Arguments of the callable
+            **kwargs: Keyword args of the callable
+
+        Returns:
+            Future object
+        """
+        _future: Future[_T] = Future()
+
+        try:
+            result: _T = fn(*args, **kwargs)
+        except BaseException as exc:  # pylint: disable=W0718
+            _future.set_exception(exc)
+        else:
+            _future.set_result(result)
+
+        return _future
+
+
+DirectExecutor.__instance__ = DirectExecutor()

pycommons/base/concurrent/future/__init__.py

@@ -0,0 +1,3 @@
+from .callback import FutureOnDoneCallback
+
+__all__ = ["FutureOnDoneCallback"]

pycommons/base/concurrent/future/callback.py

@@ -0,0 +1,14 @@
+from abc import ABC
+from asyncio import Future
+from typing import TypeVar, Generic
+
+from ...function.function import Function
+
+_T = TypeVar("_T")
+
+
+class FutureOnDoneCallback(Function[Future, _T], ABC, Generic[_T]):  # type: ignore
+    """
+    A Functional Interface that can be used to register a callback using the
+    Future's `add_done_callback` method
+    """

tests/pycommons/base/concurrent/executors/test_direct.py

@@ -0,0 +1,26 @@
+import threading
+from unittest import TestCase
+
+from pycommons.base.concurrent.executor import DirectExecutor
+
+
+class TestDirectExecutor(TestCase):
+    def test_direct_executor_executes_runnable_on_the_same_thread(self):
+        def runnable():
+            return threading.current_thread()
+
+        executor = DirectExecutor.get_instance()
+
+        future = executor.submit(runnable)
+
+        self.assertEqual(threading.current_thread(), future.result())
+
+    def test_direct_executor_executes_runnable_and_throws_exception_on_the_same_thread(self):
+        def runnable():
+            raise Exception(threading.current_thread())
+
+        executor = DirectExecutor.get_instance()
+
+        future = executor.submit(runnable)
+
+        self.assertEqual(threading.current_thread(), future.exception().args[0])

tests/pycommons/base/concurrent/future/test_callback.py

@@ -0,0 +1,14 @@
+from concurrent.futures import Future
+from unittest import TestCase
+
+from pycommons.base.concurrent.executor import DirectExecutor
+from pycommons.base.concurrent.future import FutureOnDoneCallback
+
+
+class TestFutureOnDoneCallback(TestCase):
+    class CallbackFixture(FutureOnDoneCallback[None]):
+        def apply(self, t: Future) -> None:
+            assert True is t.result()
+
+    def test_callback(self):
+        DirectExecutor.get_instance().submit(lambda: True).add_done_callback(self.CallbackFixture())