added executors
ADD
pangeo_forge/executors/__init__.py
pangeo_forge/executors/__init__.py
ADD
pangeo_forge/executors/prefect.py
+from typing import Any, Iterable, Tuple
+
+import prefect
+
+from ..recipe import DatasetRecipe
+
+class PrefectExecutor():
+
+ def prepare_plan(self, r: DatasetRecipe) -> prefect.Flow:
+
+ # wrap our functions as prefect tasks
+ @prefect.task
+ def prepare() -> None:
+ r.prepare()
+
+ @prefect.task
+ def cache_input(input: Any) -> None:
+ r.cache_input(input)
+
+ @prefect.task
+ def store_chunk(input: Any) -> None:
+ r.store_chunk(input)
+
+ @prefect.task
+ def finalize() -> None:
+ r.finalize()
+
+ with prefect.Flow("Pangeo-Forge") as flow:
+ prepare()
+ cache_input.map(list(r.iter_inputs()))
+ store_chunk.map(list(r.iter_chunks()))
+ finalize()
+
+ return flow
+
+
+ def execute_plan(self, plan: prefect.Flow, **kwargs):
+ return plan.run(**kwargs)

ADD
pangeo_forge/executors/python.py
+from functools import partial
+
+#from ..types import Pipeline, Stage, Task
+from ..recipe import DatasetRecipe
+
+from typing import Callable, Iterable
+
+Task = Callable[[], None]
+
+class PythonExecutor:
+
+ def prepare_plan(self, r: DatasetRecipe) -> Task:
+ tasks = []
+ tasks.append(r.prepare)
+ for input in r.iter_inputs():
+ tasks.append(partial(r.cache_input, input))
+ for chunk in r.iter_chunks():
+ tasks.append(partial(r.store_chunk, chunk))
+ tasks.append(r.finalize)
+
+ return partial(_execute_all, tasks)
+
+ def execute_plan(self, plan: Task, **kwargs):
+ plan()
+
+
+def _execute_all(tasks: Iterable[Task]) -> None:
+ for task in tasks:
+ task()

added executors