warning for to_pipelines
MODIFY
pangeo_forge_recipes/recipes/base.py
pangeo_forge_recipes/recipes/base.py
+import warnings
from abc import ABC, abstractmethod
from functools import partial
from typing import Callable, Hashable, Iterable
+from rechunker.types import MultiStagePipeline, ParallelPipelines, Stage
+
# How to manually execute a recipe: ###
#
# t = PangeoForgeTarget()
@@ -73,6 +76,24 @@ class BaseRecipe(ABC):
"""
pass
+ def to_pipelines(self) -> ParallelPipelines:
+ """Translate recipe to pipeline for execution.
+ """
+ warnings.warn(
+ "'to_pipelines' is deprecated. Use one of 'to_function', 'to_dask', or "
+ "'to_prefect' directly instead.",
+ FutureWarning,
+ )
+ pipeline = [] # type: MultiStagePipeline
+ if getattr(self, "cache_inputs", False): # TODO: formalize this contract
+ pipeline.append(Stage(self.cache_input, list(self.iter_inputs())))
+ pipeline.append(Stage(self.prepare_target))
+ pipeline.append(Stage(self.store_chunk, list(self.iter_chunks())))
+ pipeline.append(Stage(self.finalize_target))
+ pipelines = [] # type: ParallelPipelines
+ pipelines.append(pipeline)
+ return pipelines
+
def to_function(self) -> Callable[[], None]:
"""
Translate the recipe to a Python function for execution.

MODIFY
tests/test_recipes.py
tests/test_recipes.py
@@ -13,6 +13,13 @@ all_recipes = [
]
+def test_to_pipelines_warns(netCDFtoZarr_sequential_recipe):
+ RecipeClass, file_pattern, kwargs, ds_expected, target = netCDFtoZarr_sequential_recipe
+ rec = RecipeClass(file_pattern, **kwargs)
+ with pytest.warns(FutureWarning):
+ rec.to_pipelines()
+
+
@pytest.mark.parametrize("recipe_fixture", all_recipes)
def test_recipe(recipe_fixture, execute_recipe):
"""The basic recipe test. Use this as a template for other tests."""

warning for to_pipelines