refactoring metadata cache
MODIFY
pangeo_forge_recipes/storage.py
pangeo_forge_recipes/storage.py
import hashlib
+import json
import logging
import os
import re
@@ -7,7 +8,7 @@ import unicodedata
from abc import ABC, abstractmethod
from contextlib import contextmanager
from dataclasses import dataclass
-from typing import Any, Iterator, Optional, Union
+from typing import Any, Iterator, Optional, Sequence, Union
import fsspec
@@ -147,6 +148,22 @@ class CacheFSSpecTarget(FlatFSSpecTarget):
_copy_btw_filesystems(input_opener, target_opener)
+class MetadataTarget(FSSpecTarget):
+ """Target for storing metadata dictionaries as json."""
+
+ def __setitem__(self, key: str, value: dict) -> None:
+ mapper = self.get_mapper()
+ mapper[key] = json.dumps(value).encode("utf-8")
+
+ def __getitem__(self, key: str) -> dict:
+ return json.loads(self.get_mapper()[key])
+
+ def getitems(self, keys: Sequence[str]) -> dict:
+ mapper = self.get_mapper()
+ all_meta_raw = mapper.getitems(keys)
+ return {k: json.loads(raw_bytes) for k, raw_bytes in all_meta_raw.items()}
+
+
@contextmanager
def file_opener(
fname: str,

MODIFY
tests/conftest.py
tests/conftest.py
@@ -23,7 +23,12 @@ from pangeo_forge_recipes.patterns import (
MergeDim,
pattern_from_file_sequence,
)
-from pangeo_forge_recipes.storage import CacheFSSpecTarget, FSSpecTarget, UninitializedTarget
+from pangeo_forge_recipes.storage import (
+ CacheFSSpecTarget,
+ FSSpecTarget,
+ MetadataTarget,
+ UninitializedTarget,
+)
def pytest_addoption(parser):
@@ -120,11 +125,13 @@ def netcdf_local_paths_by_variable(daily_xarray_dataset, tmpdir_factory, request
# TODO: refactor to allow netcdf_local_paths_by_variable to be passed without
# duplicating the whole test.
-@pytest.fixture()
-def netcdf_http_server(netcdf_local_paths, request):
+@pytest.fixture(scope="session")
+def netcdf_http_paths(netcdf_local_paths, request):
paths, items_per_file = netcdf_local_paths
- def make_netcdf_http_server(username="", password=""):
+ username = ""
+ password = ""
+
first_path = paths[0]
# assume that all files are in the same directory
basedir = first_path.dirpath()
@@ -144,15 +151,15 @@ def netcdf_http_server(netcdf_local_paths, request):
command_list += [username, password]
p = subprocess.Popen(command_list, cwd=basedir)
url = f"http://127.0.0.1:{port}"
- time.sleep(1) # let the server start up
+ time.sleep(2) # let the server start up
def teardown():
p.kill()
request.addfinalizer(teardown)
- return url, fnames, items_per_file
- return make_netcdf_http_server
+ all_urls = ["/".join([url, str(fname)]) for fname in fnames]
+ return all_urls, items_per_file
@pytest.fixture()
@@ -170,6 +177,14 @@ def tmp_cache(tmpdir_factory):
return cache
+@pytest.fixture()
+def tmp_metadata_target(tmpdir_factory):
+ path = str(tmpdir_factory.mktemp("cache"))
+ fs = fsspec.get_filesystem_class("file")()
+ cache = MetadataTarget(fs, path)
+ return cache
+
+
@pytest.fixture()
def uninitialized_target():
return UninitializedTarget()

MODIFY
tests/test_storage.py
tests/test_storage.py
import pytest
+from pytest_lazyfixture import lazy_fixture
from pangeo_forge_recipes.storage import UninitializedTargetError, file_opener
@@ -40,10 +41,20 @@ def test_cache(tmp_cache):
assert not tmp_cache.exists("foo")
+def test_metadata_target(tmp_metadata_target):
+ data = {"foo": 1, "bar": "baz"}
+ tmp_metadata_target["key1"] = data
+ assert tmp_metadata_target["key1"] == data
+ assert tmp_metadata_target.getitems(["key1"]) == {"key1": data}
+
+
+@pytest.mark.parametrize(
+ "file_paths", [lazy_fixture("netcdf_local_paths"), lazy_fixture("netcdf_http_paths")]
+)
@pytest.mark.parametrize("copy_to_local", [False, True])
@pytest.mark.parametrize("use_cache, cache_first", [(False, False), (True, False), (True, True)])
-def test_file_opener(netcdf_local_paths, tmp_cache, copy_to_local, use_cache, cache_first):
- all_paths, _ = netcdf_local_paths
+def test_file_opener(file_paths, tmp_cache, copy_to_local, use_cache, cache_first):
+ all_paths, _ = file_paths
path = str(all_paths[0])
cache = tmp_cache if use_cache else None

refactoring metadata cache