add executors to __init__.py
MODIFY
pangeo_forge_recipes/executors/__init__.py
pangeo_forge_recipes/executors/__init__.py
+from .python import FunctionPipelineExecutor, GeneratorPipelineExecutor
+
+__all__ = ["GeneratorPipelineExecutor", "FunctionPipelineExecutor"]
+
+try:
+ from .dask import DaskPipelineExecutor
+
+ __all__ += ["DaskPipelineExecutor"]
+except ImportError:
+ pass
+
+try:
+ from .prefect import PrefectPipelineExecutor
+
+ __all__ += ["PrefectPipelineExecutor"]
+except ImportError:
+ pass
+
+try:
+ from .beam import BeamPipelineExecutor
+
+ __all__ += ["BeamPipelineExecutor"]
+except ImportError:
+ pass

MODIFY
pangeo_forge_recipes/executors/python.py
pangeo_forge_recipes/executors/python.py
from __future__ import annotations
-from typing import Callable, Generator, Tuple, Optional, Any
-from .base import Pipeline, PipelineExecutor
+from typing import Any, Callable, Generator
+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
@@ -25,7 +26,7 @@ class GeneratorPipelineExecutor(PipelineExecutor[GeneratorPipeline]):
return generator_function()
@staticmethod
- def execute(generator: GenetatorPipeline) -> None:
+ def execute(generator: GeneratorPipeline) -> None:
for func, args, kwargs in generator:
func(*args, **kwargs)

MODIFY
tests/test_pipelines.py
tests/test_pipelines.py
"""
Test Pipline Executors
"""
-import importlib
-
import pytest
from pytest_lazyfixture import lazy_fixture
@@ -51,21 +49,19 @@ def pipeline_with_config(tmpdir_factory):
@pytest.fixture(
scope="session",
params=[
- ("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"),
+ "DaskPipelineExecutor",
+ "FunctionPipelineExecutor",
+ "GeneratorPipelineExecutor",
+ "PrefectPipelineExecutor",
+ "BeamPipelineExecutor",
],
)
def Executor(request):
try:
- 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):
+ import pangeo_forge_recipes.executors as exec_module
+
+ return getattr(exec_module, request.param)
+ except AttributeError:
pytest.skip(f"Couldn't import {request.param}")

add executors to __init__.py