refactored storage targets
MODIFY
pangeo_forge/recipe.py
pangeo_forge/recipe.py
@@ -13,7 +13,7 @@ import xarray as xr
import zarr
from rechunker.types import MultiStagePipeline, ParallelPipelines, Stage
-from .storage import InputCache, Target
+from .storage import AbstractTarget
from .utils import chunked_iterable, fix_scalar_attr_encoding
logger = logging.getLogger(__name__)
@@ -132,6 +132,9 @@ class NetCDFtoZarrSequentialRecipe(BaseRecipe):
:param nitems_per_input: The length of each input along the `sequence_dim` dimension.
:param target: A location in which to put the dataset. Can also be assigned at run time.
:param input_cache: A location in which to cache temporary data.
+ :param require_cache: Whether to allow opening inputs directly which have not
+ yet been cached. This could lead to very slow behavior if the inputs
+ live on a slow network.
:param consolidate_zarr: Whether to consolidate the resulting Zarr dataset.
:param xarray_open_kwargs: Extra options for opening the inputs with Xarray.
:param xarray_concat_kwargs: Extra options to pass to Xarray when concatenating
@@ -142,8 +145,9 @@ class NetCDFtoZarrSequentialRecipe(BaseRecipe):
sequence_dim: str
inputs_per_chunk: int = 1
nitems_per_input: int = 1
- target: Optional[Target] = None
- input_cache: Optional[InputCache] = None
+ target: Optional[AbstractTarget] = None
+ input_cache: Optional[AbstractTarget] = None
+ require_cache: bool = True
consolidate_zarr: bool = True
xarray_open_kwargs: dict = field(default_factory=dict)
xarray_concat_kwargs: dict = field(default_factory=dict)
@@ -217,19 +221,18 @@ class NetCDFtoZarrSequentialRecipe(BaseRecipe):
@contextmanager
def input_opener(self, fname: str):
- if self.input_cache is None:
- logger.info(f"No cache. Opening input `{fname}` directly.")
- # This will bypass the cache. May be slow.
- with input_opener(fname, mode="rb") as f:
- yield f
- elif self.input_cache.exists(fname):
- logger.info(f"Input '{fname}' found in cache")
+ try:
with self.input_cache.open(fname, mode="rb") as f:
+ logger.info(f"Opening '{fname}' from cache")
yield f
+ except IOError: # TODO figure out the excpetion to catch
+ if self.require_cache:
+ raise
else:
- raise ValueError(
- f"Input '{fname}' has not been cached yet. " "Call .cache_input() first."
- )
+ logger.info(f"No cache found. Opening input `{fname}` directly.")
+ # This will bypass the cache. May be slow.
+ with input_opener(fname, mode="rb") as f:
+ yield f
def open_input(self, fname: str):
with self.input_opener(fname) as f:

MODIFY
pangeo_forge/storage.py
pangeo_forge/storage.py
import os
+import re
+import unicodedata
+from abc import ABC, abstractmethod
from contextlib import contextmanager
from dataclasses import dataclass
from typing import BinaryIO, NoReturn
@@ -6,20 +9,25 @@ from typing import BinaryIO, NoReturn
import fsspec
-@dataclass
-class Target:
- """Representation of a storage target for Pangeo Forge.
+class AbstractTarget(ABC):
+ @abstractmethod
+ def get_mapper(self):
+ pass
- :param fs: The filesystem object we are writing to.
- :param path: The path where the target data will be saved.
- """
+ @abstractmethod
+ def exists(self, path) -> bool:
+ """Check that the file exists."""
+ pass
- fs: fsspec.AbstractFileSystem
- path: str
+ @abstractmethod
+ def rm(self, path) -> NoReturn:
+ """Remove file."""
+ pass
- def get_mapper(self) -> fsspec.mapping.FSMap:
- """Get a mutable mapping object suitable for storing Zarr data."""
- return self.fs.get_mapper(self.path)
+ @contextmanager
+ def open(self, path, **kwargs) -> BinaryIO:
+ """Open file with a context manager."""
+ pass
def _hash_path(path: str) -> str:
@@ -27,19 +35,22 @@ def _hash_path(path: str) -> str:
@dataclass
-class InputCache:
- """Representation of an intermediate storage location where remote files
- Can be cached locally.
+class FSSpecTarget(AbstractTarget):
+ """Representation of a storage target for Pangeo Forge.
- :param fs: The filesystem we are writing to.
- :param prefix: A path prepended to all paths.
+ :param fs: The filesystem object we are writing to.
+ :param root_path: The path under which the target data will be stored.
"""
fs: fsspec.AbstractFileSystem
- prefix: str = ""
+ root_path: str = ""
+
+ def get_mapper(self) -> fsspec.mapping.FSMap:
+ """Get a mutable mapping object suitable for storing Zarr data."""
+ return self.fs.get_mapper(self.root_path)
def _full_path(self, path):
- return os.path.join(self.prefix, _hash_path(path))
+ return os.path.join(self.root_path, path)
def exists(self, path) -> bool:
"""Check that the file is in the cache."""
@@ -56,5 +67,29 @@ class InputCache:
yield f
def __post_init__(self):
- if not self.fs.isdir(self.prefix):
- self.fs.mkdir(self.prefix)
+ if not self.fs.isdir(self.root_path):
+ self.fs.mkdir(self.root_path)
+
+
+class FlatFSSpecTarget(FSSpecTarget):
+ """A target that sanitizes all the path names so that everthing is stored
+ in a single directory.
+
+ Designed to be used as a cache for inputs.
+ """
+
+ def _full_path(self, path):
+ slug = _slugify(path)
+ prefix = prefix = hex(hash(path))[2:10]
+ new_path = "-".join([prefix, slug])
+ return os.path.join(self.root_path, new_path)
+
+
+def _slugify(value):
+ # Adopted from
+ # https://github.com/django/django/blob/master/django/utils/text.py
+ # https://stackoverflow.com/questions/295135/turn-a-string-into-a-valid-filename
+ value = str(value)
+ value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
+ value = re.sub(r"[^\w\s-]+", "_", value.lower())
+ return re.sub(r"[-\s]+", "-", value).strip("-_")

MODIFY
tests/conftest.py
tests/conftest.py
@@ -8,7 +8,7 @@ import pytest
import xarray as xr
from pangeo_forge import recipe
-from pangeo_forge.storage import InputCache, Target
+from pangeo_forge.storage import FlatFSSpecTarget, FSSpecTarget
# where to run the http server
_PORT = "8080"
@@ -80,14 +80,14 @@ def tmp_target(tmpdir_factory):
fs = fsspec.get_filesystem_class("file")()
path = str(tmpdir_factory.mktemp("target"))
- return Target(fs, path)
+ return FSSpecTarget(fs, path)
@pytest.fixture()
def tmp_cache(tmpdir_factory):
path = str(tmpdir_factory.mktemp("cache"))
fs = fsspec.get_filesystem_class("file")()
- cache = InputCache(fs, prefix=path)
+ cache = FlatFSSpecTarget(fs, path)
return cache

MODIFY
tests/test_fixtures.py
tests/test_fixtures.py
@@ -24,7 +24,7 @@ def test_fixture_http_files(daily_xarray_dataset, netcdf_http_server):
def test_target(tmp_target):
mapper = tmp_target.get_mapper()
mapper["foo"] = b"bar"
- with open(tmp_target.path + "/foo") as f:
+ with open(tmp_target.root_path + "/foo") as f:
res = f.read()
assert res == "bar"

refactored storage targets