add generator executor
DELETE
pangeo_forge_recipes/executors/function.py
-from typing import Callable
-
-from .base import Pipeline, PipelineExecutor
-
-
-class FunctionPipelineExecutor(PipelineExecutor[Callable]):
- @staticmethod
- def compile(pipeline: Pipeline):
- def function():
- for stage in pipeline.stages:
- if stage.mappable is not None:
- for m in stage.mappable:
- stage.function(m, config=pipeline.config)
- else:
- stage.function(config=pipeline.config)
-
- return function
-
- @staticmethod
- def execute(func: Callable) -> None:
- func()

ADD
pangeo_forge_recipes/executors/python.py
+from __future__ import annotations
+from typing import Callable, Generator, Tuple, Optional, Any
+
+from .base import Pipeline, PipelineExecutor
+
+
+GeneratorPipeline = Generator[Any, None, None]
+
+class GeneratorPipelineExecutor(PipelineExecutor[GeneratorPipeline]):
+ """An executor which returns a Generator.
+ The Generator yeilds `function, args, kwargs`, which can be called step by step
+ to iterate through the recipe.
+ """
+
+ @staticmethod
+ def compile(pipeline: Pipeline):
+ def generator_function():
+ for stage in pipeline.stages:
+ if stage.mappable is not None:
+ for m in stage.mappable:
+ yield stage.function, (m,), dict(config=pipeline.config)
+ else:
+ yield stage.function, (), dict(config=pipeline.config)
+
+ return generator_function()
+
+ @staticmethod
+ def execute(generator: GenetatorPipeline) -> None:
+ for func, args, kwargs in generator:
+ func(*args, **kwargs)
+
+
+class FunctionPipelineExecutor(PipelineExecutor[Callable]):
+ """A generator which returns a single callable python function with no
+ arguments. Calling this function will run the whole recipe"""
+
+ @staticmethod
+ def compile(pipeline: Pipeline):
+ def function():
+ for stage in pipeline.stages:
+ if stage.mappable is not None:
+ for m in stage.mappable:
+ stage.function(m, config=pipeline.config)
+ else:
+ stage.function(config=pipeline.config)
+
+ return function
+
+ @staticmethod
+ def execute(func: Callable) -> None:
+ func()

MODIFY
tests/test_pipelines.py
tests/test_pipelines.py
@@ -51,18 +51,22 @@ def pipeline_with_config(tmpdir_factory):
@pytest.fixture(
scope="session",
params=[
- ("pangeo_forge_recipes.executors.dask", "DaskPipelineExecutor"),
- ("pangeo_forge_recipes.executors.function", "FunctionPipelineExecutor"),
- ("pangeo_forge_recipes.executors.prefect", "PrefectPipelineExecutor"),
- ("pangeo_forge_recipes.executors.beam", "BeamPipelineExecutor"),
+ ("pangeo_forge_recipes.executors.dask.DaskPipelineExecutor"),
+ ("pangeo_forge_recipes.executors.python.FunctionPipelineExecutor"),
+ ("pangeo_forge_recipes.executors.python.GeneratorPipelineExecutor"),
+ ("pangeo_forge_recipes.executors.prefect.PrefectPipelineExecutor"),
+ ("pangeo_forge_recipes.executors.beam.BeamPipelineExecutor"),
],
)
def Executor(request):
try:
- module = importlib.import_module(request.param[0])
- return getattr(module, request.param[1])
+ split = request.param.split('.')
+ executor_name = split[-1]
+ module_name = '.'.join(split[:-1])
+ module = importlib.import_module(module_name)
+ return getattr(module, executor_name)
except (AttributeError, ImportError):
- pytest.skip()
+ pytest.skip(f"Couldn't import {request.param}")
@pytest.mark.parametrize(

add generator executor