Merge pull request #28 from savannahghi/next

release v1.1.0-rc.3

docs/conf.py

@@ -71,11 +71,15 @@
     ("py:class", "TracebackType"),  # Used as type annotation. Only available when type checking
     ("py:class", "concurrent.futures._base.Executor"),  # sphinx can't find it
     ("py:class", "concurrent.futures._base.Future"),  # sphinx can't find it
+    ("py:class", "sghi.exceptions.SGHIError"),  # sphinx can't find it
     ("py:class", "sghi.etl.commons.processors._RDT"),  # private type annotations
     ("py:class", "sghi.etl.commons.processors._PDT"),  # private type annotations
     ("py:class", "sghi.etl.commons.sinks._PDT"),  # private type annotations
     ("py:class", "sghi.etl.commons.sources._RDT"),  # private type annotations
     ("py:class", "sghi.etl.commons.utils.result_gatherers._T"),  # private type annotations
+    ("py:class", "sghi.etl.commons.utils.result_gatherers._T1"),  # private type annotations
+    ("py:class", "sghi.etl.commons.workflow_builder._RDT"),  # private type annotations
+    ("py:class", "sghi.etl.commons.workflow_builder._PDT"),  # private type annotations
     ("py:class", "sghi.etl.commons.workflow_definitions._RDT"),  # private type annotations
     ("py:class", "sghi.etl.commons.workflow_definitions._PDT"),  # private type annotations
     ("py:class", "sghi.etl.core._RDT"),  # private type annotations
@@ -93,6 +97,8 @@
     ("py:obj", "sghi.etl.commons.processors._RDT"),  # private type annotations
     ("py:obj", "sghi.etl.commons.sinks._PDT"),  # private type annotations
     ("py:obj", "sghi.etl.commons.sources._RDT"),  # private type annotations
+    ("py:obj", "sghi.etl.commons.workflow_builder._RDT"),  # private type annotations
+    ("py:obj", "sghi.etl.commons.workflow_builder._PDT"),  # private type annotations
     ("py:obj", "sghi.etl.commons.workflow_definitions._RDT"),  # private type annotations
     ("py:obj", "sghi.etl.commons.workflow_definitions._PDT"),  # private type annotations
 ]

docs/index.rst

@@ -40,6 +40,7 @@ API Reference
      sghi.etl.commons.sinks
      sghi.etl.commons.sources
      sghi.etl.commons.utils
+     sghi.etl.commons.workflow_builder
      sghi.etl.commons.workflow_definitions
 
 

src/sghi/etl/commons/__init__.py

@@ -11,18 +11,26 @@
 from .sinks import NullSink, ScatterSink, SplitSink, sink
 from .sources import GatherSource, source
 from .utils import fail_fast, fail_fast_factory, ignored_failed, run_workflow
+from .workflow_builder import (
+    NoSourceProvidedError,
+    SoleValueAlreadyRetrievedError,
+    WorkflowBuilder,
+)
 from .workflow_definitions import SimpleWorkflowDefinition
 
 __all__ = [
     "GatherSource",
     "NOOPProcessor",
+    "NoSourceProvidedError",
     "NullSink",
     "ProcessorPipe",
     "SimpleWorkflowDefinition",
+    "SoleValueAlreadyRetrievedError",
     "ScatterGatherProcessor",
     "ScatterSink",
     "SplitGatherProcessor",
     "SplitSink",
+    "WorkflowBuilder",
     "fail_fast",
     "fail_fast_factory",
     "ignored_failed",

src/sghi/etl/commons/processors.py

@@ -198,7 +198,7 @@ class ProcessorPipe(Processor[_RDT, _PDT], Generic[_RDT, _PDT]):
     processors.
 
     .. admonition:: Regarding retry safety
-        :class: tip
+        :class: caution
 
         Instances of this ``Processor`` are **NOT SAFE** to retry.
     """
@@ -331,9 +331,9 @@ def dispose(self) -> None:
     def _processor_to_task(self, p: Processor[_RDT, _PDT]) -> Task[_RDT, _PDT]:
         @task
         def do_apply(raw_data: _RDT) -> _PDT:
-            with p as _p:
-                apply = self._retry_policy_factory().retry(_p.apply)
-                return apply(raw_data)
+            _p = p.__enter__()
+            apply = self._retry_policy_factory().retry(_p.apply)
+            return apply(raw_data)
 
         return do_apply
 
@@ -368,7 +368,7 @@ class ScatterGatherProcessor(
     processors.
 
     .. admonition:: Regarding retry safety
-        :class: tip
+        :class: caution
 
         Instances of this ``Processor`` are **NOT SAFE** to retry.
     """
@@ -518,8 +518,8 @@ def apply(self, raw_data: _RDT) -> Sequence[_PDT]:
             "Forking processing of the received data to all embedded "
             "processors."
         )
-        with self._executor as executor:
-            futures = executor.execute(raw_data)
+        executor = self._executor.__enter__()
+        futures = executor.execute(raw_data)
 
         return tuple(self._result_gatherer(futures))
 
@@ -551,9 +551,9 @@ def dispose(self) -> None:
     def _processor_to_task(self, p: Processor[_RDT, _PDT]) -> Task[_RDT, _PDT]:
         @task
         def do_apply(raw_data: _RDT) -> _PDT:
-            with p as _p:
-                apply = self._retry_policy_factory().retry(_p.apply)
-                return apply(raw_data)
+            _p = p.__enter__()
+            apply = self._retry_policy_factory().retry(_p.apply)
+            return apply(raw_data)
 
         return do_apply
 
@@ -588,7 +588,7 @@ class SplitGatherProcessor(
     processors.
 
     .. admonition:: Regarding retry safety
-        :class: tip
+        :class: caution
 
         Instances of this ``Processor`` are **NOT SAFE** to retry.
     """  # noqa: D205
@@ -757,8 +757,8 @@ def apply(self, raw_data: Sequence[_RDT]) -> Sequence[_PDT]:
             "to each data part."
         )
 
-        with self._executor as executor:
-            futures = executor.execute(raw_data)
+        executor = self._executor.__enter__()
+        futures = executor.execute(raw_data)
 
         return tuple(self._result_gatherer(futures))
 
@@ -794,9 +794,9 @@ def _processor_to_task(
     ) -> Task[Sequence[_RDT], _PDT]:
         @task
         def do_apply(raw_data: Sequence[_RDT]) -> _PDT:
-            with p as _p:
-                apply = self._retry_policy_factory().retry(_p.apply)
-                return apply(raw_data[i])
+            _p = p.__enter__()
+            apply = self._retry_policy_factory().retry(_p.apply)
+            return apply(raw_data[i])
 
         return do_apply
 

src/sghi/etl/commons/sinks.py

@@ -99,7 +99,7 @@ def sink(f: Callable[[_PDT], None]) -> Sink[_PDT]:
 class NullSink(Sink[_PDT], Generic[_PDT]):
     """A :class:`Sink` that discards all the data it receives.
 
-    Like to ``dev/null`` on Unix, instances of this ``Sink`` discard all data
+    Like ``dev/null`` on Unix, instances of this ``Sink`` discard all data
     drained to them but report the drain operation as successful. This is
     mostly useful as a placeholder or where further consumption of processed
     data is not required.
@@ -190,7 +190,7 @@ class ScatterSink(Sink[_PDT], Generic[_PDT]):
     Disposing instances of this class also disposes of their embedded sinks.
 
     .. admonition:: Regarding retry safety
-        :class: tip
+        :class: caution
 
         Instances of this ``Sink`` are **NOT SAFE** to retry.
     """
@@ -324,8 +324,8 @@ def drain(self, processed_data: _PDT) -> None:
         """
         self._logger.info("Draining processed data to all available sinks.")
 
-        with self._executor as executor:
-            futures = executor.execute(processed_data)
+        executor = self._executor.__enter__()
+        futures = executor.execute(processed_data)
 
         self._result_gatherer(futures)
 
@@ -357,9 +357,9 @@ def dispose(self) -> None:
     def _sink_to_task(self, s: Sink[_PDT]) -> Task[_PDT, None]:
         @task
         def do_drain(processed_data: _PDT) -> None:
-            with s as _s:
-                drain = self._retry_policy_factory().retry(_s.drain)
-                return drain(processed_data)
+            _s = s.__enter__()
+            drain = self._retry_policy_factory().retry(_s.drain)
+            return drain(processed_data)
 
         return do_drain
 
@@ -390,7 +390,7 @@ class SplitSink(Sink[Sequence[_PDT]], Generic[_PDT]):
     Disposing instances of this class also disposes of their embedded sinks.
 
     .. admonition:: Regarding retry safety
-        :class: tip
+        :class: caution
 
         Instances of this ``Sink`` are **NOT SAFE** to retry.
     """  # noqa: D205
@@ -550,8 +550,8 @@ def drain(self, processed_data: Sequence[_PDT]) -> None:
             "to all available sinks."
         )
 
-        with self._executor as executor:
-            futures = executor.execute(processed_data)
+        executor = self._executor.__enter__()
+        futures = executor.execute(processed_data)
 
         self._result_gatherer(futures)
 
@@ -587,9 +587,9 @@ def _sink_to_task(
     ) -> Task[Sequence[_PDT], None]:
         @task
         def do_drain(processed_data: Sequence[_PDT]) -> None:
-            with s as _s:
-                drain = self._retry_policy_factory().retry(_s.drain)
-                return drain(processed_data[i])
+            _s = s.__enter__()
+            drain = self._retry_policy_factory().retry(_s.drain)
+            return drain(processed_data[i])
 
         return do_drain
 

src/sghi/etl/commons/sources.py

@@ -118,7 +118,7 @@ class GatherSource(Source[Sequence[_RDT]], Generic[_RDT]):
     Disposing instances of this class also disposes of their embedded sources.
 
     .. admonition:: Regarding retry safety
-        :class: tip
+        :class: caution
 
         Instances of this ``Source`` are **NOT SAFE** to retry.
     """
@@ -256,8 +256,8 @@ def draw(self) -> Sequence[_RDT]:
         """
         self._logger.info("Aggregating data from all available sources.")
 
-        with self._executor as executor:
-            futures = executor.execute(None)
+        executor = self._executor.__enter__()
+        futures = executor.execute(None)
 
         return tuple(self._result_gatherer(futures))
 
@@ -289,9 +289,9 @@ def dispose(self) -> None:
     def _source_to_task(self, s: Source[_RDT]) -> Supplier[_RDT]:
         @supplier
         def do_draw() -> _RDT:
-            with s as _s:
-                draw = self._retry_policy_factory().retry(_s.draw)
-                return draw()
+            _s = s.__enter__()
+            draw = self._retry_policy_factory().retry(_s.draw)
+            return draw()
 
         # noinspection PyTypeChecker
         return do_draw