feat(workflow-builder): add a WorkflowBuilder (#27)

A `WorkflowBuilder` is a class that aids in the structured construction of `WorkflowDefinition` instances. The builder class offers a convenient way to construct SGHI ETL Workflows by providing methods to register sources, processors and sinks.
savannahghi · Nov 17, 2024 · 53aee40 · 53aee40
1 parent 5df425d
commit 53aee40
Show file tree

Hide file tree

Showing 7 changed files with 2,285 additions and 0 deletions.
diff --git a/docs/conf.py b/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
 ]

diff --git a/docs/index.rst b/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
 
 

diff --git a/src/sghi/etl/commons/__init__.py b/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",