do some linting
MODIFY
pangeo_forge_recipes/executors/base.py
pangeo_forge_recipes/executors/base.py
import enum
-from abc import ABC, abstractmethod
from dataclasses import dataclass
-from typing import Any, Dict, Generic, Hashable, Iterable, Optional, Protocol, TypeVar, Union
+from typing import Any, Callable, Dict, Generic, Hashable, Iterable, Optional, Protocol, TypeVar
Config = Any
+
# https://stackoverflow.com/questions/57837609/python-typing-signature-typing-callable-for-function-with-kwargs
class NoArgumentStageFunction(Protocol):
def __call__(*, config: Optional[Config] = None) -> None:
@@ -16,7 +16,15 @@ class SingleArgumentStageFunction(Protocol):
...
-StageFunction = Union[NoArgumentStageFunction, SingleArgumentStageFunction]
+# For some reason, mypy does not like this
+# StageFunction = Union[NoArgumentStageFunction, SingleArgumentStageFunction]
+# pangeo_forge_recipes/recipes/xarray_zarr.py:525: error:
+# Argument "function" to "Stage" has incompatible type
+# "Callable[[Index, NamedArg(XarrayZarrRecipe, 'config')], None]";
+# expected "NoArgumentStageFunction" [arg-type]
+
+# TODO: fix this to be a stricter type as above
+StageFunction = Callable
class StageAnnotationType(enum.Enum):

MODIFY
pangeo_forge_recipes/recipes/base.py
pangeo_forge_recipes/recipes/base.py
from __future__ import annotations
-from abc import ABC, abstractmethod
+from abc import ABC
from dataclasses import dataclass, replace
-from typing import Callable, ClassVar, Hashable, Iterable
+from typing import Callable, ClassVar
from ..executors.base import Pipeline
from ..patterns import FilePattern, prune_pattern

MODIFY
pangeo_forge_recipes/recipes/xarray_zarr.py
pangeo_forge_recipes/recipes/xarray_zarr.py
@@ -11,7 +11,7 @@ from contextlib import ExitStack, contextmanager
from dataclasses import dataclass, field, replace
from itertools import chain, product
from math import ceil
-from typing import Callable, Dict, Hashable, Iterator, List, Optional, Sequence, Set, Tuple
+from typing import Callable, Dict, Iterator, List, Optional, Sequence, Set, Tuple
import dask
import numpy as np
@@ -21,7 +21,7 @@ import zarr
from ..chunk_grid import ChunkGrid
from ..executors.base import Pipeline, Stage
from ..patterns import CombineOp, DimIndex, FilePattern, Index
-from ..storage import AbstractTarget, CacheFSSpecTarget, MetadataTarget, file_opener
+from ..storage import CacheFSSpecTarget, FSSpecTarget, MetadataTarget, file_opener
from ..utils import calc_subsets, fix_scalar_attr_encoding, lock_for_conflicts
from .base import BaseRecipe, FilePatternMixin
@@ -84,7 +84,7 @@ def inputs_for_chunk(
return input_keys
-def expand_target_dim(target: CacheFSSpecTarget, concat_dim: Optional[str], dimsize: int) -> None:
+def expand_target_dim(target: FSSpecTarget, concat_dim: Optional[str], dimsize: int) -> None:
target_mapper = target.get_mapper()
zgroup = zarr.open_group(target_mapper)
ds = open_target(target)
@@ -105,7 +105,7 @@ def expand_target_dim(target: CacheFSSpecTarget, concat_dim: Optional[str], dims
zgroup[concat_dim][:] = 0
-def open_target(target: CacheFSSpecTarget) -> xr.Dataset:
+def open_target(target: FSSpecTarget) -> xr.Dataset:
return xr.open_zarr(target.get_mapper())
@@ -343,6 +343,8 @@ def calculate_sequence_lens(
def prepare_target(*, config: XarrayZarrRecipe) -> None:
+ if config.target is None:
+ raise ValueError("Cannot proceed without a target")
try:
ds = open_target(config.target)
logger.info("Found an existing dataset in target")
@@ -582,7 +584,7 @@ class XarrayZarrRecipe(BaseRecipe, FilePatternMixin):
inputs_per_chunk: int = 1
target_chunks: Dict[str, int] = field(default_factory=dict)
- target: Optional[AbstractTarget] = None
+ target: Optional[FSSpecTarget] = None
input_cache: Optional[CacheFSSpecTarget] = None
metadata_cache: Optional[MetadataTarget] = None
cache_inputs: Optional[bool] = None

MODIFY
tests/test_pipelines.py
tests/test_pipelines.py
@@ -55,7 +55,7 @@ def pipeline_with_config(tmpdir_factory):
)
@pytest.mark.parametrize(
"pipeline_config_tmpdir",
- [lazy_fixture("pipeline_no_config"), lazy_fixture("pipeline_with_config"),],
+ [lazy_fixture("pipeline_no_config"), lazy_fixture("pipeline_with_config")],
)
def test_pipeline(pipeline_config_tmpdir, Executor):
pipeline, config, tmpdir = pipeline_config_tmpdir

do some linting