Repository: dcs4cop/xcube

Issue:
Allow explicit configuration of datasets served by the S3 store
**Is your feature request related to a problem? Please describe.**

Consider a situation where we want a store to serve data from Zarrs from a very large S3 bucket (millions of files, petabytes of data; e.g. Eodata on Creodias). The usual store creation parameters, `root` and `max_depth`, are insufficient, because:

 - The Zarrs we want to serve are not guaranteed to be grouped under a common prefix
 - A search for Zarrs in a bucket of this size, even starting from the longest common prefix, would take a very long time
 - There may be Zarrs that we wish to exclude from the store, for instance because of known problems with data or compatibility
 - The paths of the Zarrs in the store may have lengthy prefixes and names, and may therefore be inconvenient for users to work with

**Describe the solution you'd like**

I would like to be able to do something like the following:

```python
store = new_data_store('s3', root='some-bucket',
            serve_only={
                'nice-name-1': 'long/messy/path/long-and-messy-name.zarr',
                'nice-name-2': 'some/other/long/path/another-long-name.zarr',
                'nice-name-3': 'i/think/you/get/the-idea.zarr'
            })
```

Supplying the `serve_only` parameter disables automatic data discovery, and serves only the datasets specified in the dictionary values, presenting them to the user under the aliases given by the dictionary keys.

**Describe alternatives you've considered**

 - Creating a separate store for each dataset: inconvenient (especially for large numbers of datasets), doesn't allow aliasing, doesn't allow exclusion of unwanted datasets with the same prefix as wanted datasets
 - Include/exclude list parameters: doesn't allow aliasing; doesn't necessarily disable automatic dataset discovery
 - Implement an external S3 ‘facade’, using the original bucket as a back-end and serving selected Zarrs under selected aliases: more work and more points of failure than implementing the suggested feature directly in xcube.


Discussion:

Patch:
diff --git a/CHANGES.md b/CHANGES.md
index 2e6a6c92..5e35c3a8 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -2,6 +2,17 @@
 
 ### Other
 
+* The xcube filesystem data stores such as "file", "s3", "memory"
+  can now filter the data identifiers reported by `get_data_ids()`. (#585)
+  For this purpose, the data stores now accept two new optional keywords
+  which both can take the form of a wildcard pattern or a sequence 
+  of wildcard patterns:
+
+  1. `excludes`: if given and if any pattern matches the identifier, 
+     the identifier is not reported. 
+  2. `includes`: if not given or if any pattern matches the identifier, 
+     the identifier is reported.
+
 ### Fixes
 
 
diff --git a/xcube/core/store/fs/registry.py b/xcube/core/store/fs/registry.py
index 065f4b66..a70931b4 100644
--- a/xcube/core/store/fs/registry.py
+++ b/xcube/core/store/fs/registry.py
@@ -19,7 +19,7 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-from typing import Type, Dict, Optional, Any
+from typing import Type, Dict, Optional, Any, Sequence
 
 import fsspec
 
@@ -188,11 +188,23 @@ def new_fs_data_store(
         root: str = '',
         max_depth: Optional[int] = 1,
         read_only: bool = False,
+        includes: Optional[Sequence[str]] = None,
+        excludes: Optional[Sequence[str]] = None,
         storage_options: Dict[str, Any] = None
 ) -> FsDataStore:
     """
     Create a new instance of a filesystem-based data store.
 
+    The data store is capable of filtering the data identifiers reported
+    by ``get_data_ids()``. For this purpose the optional keywords
+    `excludes` and `includes` are used which can both take the form of
+    a wildcard pattern or a sequence of wildcard patterns:
+
+    * ``excludes``: if given and if any pattern matches the identifier,
+      the identifier is not reported.
+    * ``includes``: if not given or if any pattern matches the identifier,
+      the identifier is reported.
+
     :param protocol: The filesystem protocol,
         for example "file", "s3", "memory".
     :param root: Root or base directory.
@@ -201,6 +213,12 @@ def new_fs_data_store(
         Defaults to 1.
     :param read_only: Whether this is a read-only store.
         Defaults to False.
+    :param includes: Optional sequence of wildcards that include
+        certain filesystem paths. Affects the data identifiers (paths)
+        returned by `get_data_ids()`. By default, all paths are included.
+    :param excludes: Optional sequence of wildcards that exclude
+        certain filesystem paths. Affects the data identifiers (paths)
+        returned by `get_data_ids()`. By default, no paths are excluded.
     :param storage_options: Options specific to the underlying filesystem
         identified by *protocol*.
         Used to instantiate the filesystem.
@@ -212,6 +230,8 @@ def new_fs_data_store(
                     for k, v in dict(root=root,
                                      max_depth=max_depth,
                                      read_only=read_only,
+                                     includes=includes,
+                                     excludes=excludes,
                                      storage_options=storage_options).items()
                     if v is not None}
     assert_valid_params(store_params,
diff --git a/xcube/core/store/fs/store.py b/xcube/core/store/fs/store.py
index 9a495c07..ce5f5db6 100644
--- a/xcube/core/store/fs/store.py
+++ b/xcube/core/store/fs/store.py
@@ -19,13 +19,14 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
+import fnmatch
 import os.path
 import pathlib
 import uuid
 import warnings
 from threading import RLock
 from typing import Optional, Iterator, Any, Tuple, List, Dict, \
-    Union, Container, Callable
+    Union, Container, Callable, Sequence
 
 import fsspec
 import geopandas as gpd
@@ -37,8 +38,11 @@ from xcube.util.assertions import assert_in
 from xcube.util.assertions import assert_instance
 from xcube.util.assertions import assert_true
 from xcube.util.extension import Extension
+from xcube.util.jsonschema import JsonArraySchema
 from xcube.util.jsonschema import JsonBooleanSchema
+from xcube.util.jsonschema import JsonComplexSchema
 from xcube.util.jsonschema import JsonIntegerSchema
+from xcube.util.jsonschema import JsonNullSchema
 from xcube.util.jsonschema import JsonObjectSchema
 from xcube.util.jsonschema import JsonStringSchema
 from .accessor import FsAccessor
@@ -93,12 +97,28 @@ _FORMAT_TO_DATA_TYPE_ALIASES = {
     'shapefile': (GEO_DATA_FRAME_TYPE.alias,),
 }
 
+_DataId = str
+_DataIdTuple = Tuple[_DataId, Dict[str, Any]]
+_DataIdIter = Iterator[_DataId]
+_DataIdTupleIter = Iterator[_DataIdTuple]
+_DataIds = Union[_DataIdIter, _DataIdTupleIter]
+
 
 class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
     """
     Base class for data stores that use an underlying filesystem
     of type ``fsspec.AbstractFileSystem``.
 
+    The data store is capable of filtering the data identifiers reported
+    by ``get_data_ids()``. For this purpose the optional keywords
+    `excludes` and `includes` are used which can both take the form of
+    a wildcard pattern or a sequence of wildcard patterns:
+
+    * ``excludes``: if given and if any pattern matches the identifier,
+      the identifier is not reported.
+    * ``includes``: if not given or if any pattern matches the identifier,
+      the identifier is reported.
+
     :param fs: Optional filesystem. If not given,
         :meth:_load_fs() must return a filesystem instance.
     :param root: Root or base directory.
@@ -107,13 +127,21 @@ class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
         Defaults to 1.
     :param read_only: Whether this is a read-only store.
         Defaults to False.
+    :param includes: Optional sequence of wildcards that identify included
+        filesystem paths. Affects the data identifiers (paths)
+        returned by `get_data_ids()`. By default, all paths are included.
+    :param excludes: Optional sequence of wildcards that identify excluded
+        filesystem paths. Affects the data identifiers (paths)
+        returned by `get_data_ids()`. By default, no paths are excluded.
     """
 
     def __init__(self,
                  fs: Optional[fsspec.AbstractFileSystem] = None,
                  root: str = '',
                  max_depth: Optional[int] = 1,
-                 read_only: bool = False):
+                 read_only: bool = False,
+                 includes: Optional[Union[str, Sequence[str]]] = None,
+                 excludes: Optional[Union[str, Sequence[str]]] = None):
         if fs is not None:
             assert_instance(fs, fsspec.AbstractFileSystem, name='fs')
         self._fs = fs
@@ -121,6 +149,8 @@ class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
         self._root: Optional[str] = None
         self._max_depth = max_depth
         self._read_only = read_only
+        self._includes = self._normalize_wc(includes)
+        self._excludes = self._normalize_wc(excludes)
         self._lock = RLock()
 
     @property
@@ -167,22 +197,41 @@ class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
 
     @property
     def max_depth(self) -> Optional[int]:
+        """Maximum recursion depth. None means limitless."""
         return self._max_depth
 
     @property
     def read_only(self) -> bool:
+        """Whether this is a read-only store."""
         return self._read_only
 
+    @property
+    def includes(self) -> Tuple[str]:
+        """Wildcard patterns that include paths."""
+        return self._includes
+
+    @property
+    def excludes(self) -> Tuple[str]:
+        """Wildcard patterns that exclude paths."""
+        return self._excludes
+
     #########################################################################
     # MutableDataStore impl.
 
     @classmethod
     def get_data_store_params_schema(cls) -> JsonObjectSchema:
+        wc_schema = JsonComplexSchema(one_of=[
+            JsonNullSchema(),
+            JsonStringSchema(),
+            JsonArraySchema(items=JsonStringSchema())
+        ])
         return JsonObjectSchema(
             properties=dict(
                 root=JsonStringSchema(default=''),
                 max_depth=JsonIntegerSchema(nullable=True, default=1),
                 read_only=JsonBooleanSchema(default=False),
+                includes=wc_schema,
+                excludes=wc_schema,
             ),
             additional_properties=False
         )
@@ -211,12 +260,14 @@ class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
 
     def get_data_ids(self,
                      data_type: DataTypeLike = None,
-                     include_attrs: Container[str] = None) -> \
-            Union[Iterator[str], Iterator[Tuple[str, Dict[str, Any]]]]:
+                     include_attrs: Container[str] = None) -> _DataIds:
         data_type = DataType.normalize(data_type)
         # TODO: do not ignore names in include_attrs
         return_tuples = include_attrs is not None
-        yield from self._generate_data_ids('', data_type, return_tuples, 1)
+        data_ids = self._generate_data_ids('', data_type, return_tuples, 1)
+        if self._includes or self._excludes:
+            yield from self._filter_data_ids(data_ids, return_tuples)
+        yield from data_ids
 
     def has_data(self, data_id: str, data_type: DataTypeLike = None) -> bool:
         assert_given(data_id, 'data_id')
@@ -609,7 +660,7 @@ class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
                            dir_path: str,
                            data_type: DataType,
                            return_tuples: bool,
-                           current_depth: int):
+                           current_depth: int) -> _DataIds:
         root = self.root + ('/' + dir_path if dir_path else '')
         if not self.fs.exists(root):
             return
@@ -632,6 +683,31 @@ class BaseFsDataStore(DefaultSearchMixin, MutableDataStore):
                                                    return_tuples,
                                                    current_depth + 1)
 
+    def _filter_data_ids(self,
+                         data_ids: _DataIds,
+                         return_tuples: bool) -> _DataIds:
+        for data_id in data_ids:
+            path = data_id[0] if return_tuples else data_id
+            excluded = False
+            for pattern in self._excludes:
+                excluded = fnmatch.fnmatch(path, pattern)
+                if excluded:
+                    break
+            if not excluded:
+                included = True
+                for pattern in self._includes:
+                    included = fnmatch.fnmatch(path, pattern)
+                    if included:
+                        break
+                if included:
+                    yield data_id
+
+    @staticmethod
+    def _normalize_wc(wc: Optional[Union[str, Sequence[str]]]) -> Tuple[str]:
+        return tuple() if not wc \
+            else (wc,) if isinstance(wc, str) \
+            else tuple(wc)
+
 
 class FsDataStore(BaseFsDataStore, FsAccessor):
     """
@@ -639,25 +715,45 @@ class FsDataStore(BaseFsDataStore, FsAccessor):
     also implements a :class:FsAccessor which serves
     the filesystem.
 
-    :param storage_options: Parameters specific to the underlying filesystem.
-        Used to instantiate the filesystem.
+    The data store is capable of filtering the data identifiers reported
+    by ``get_data_ids()``. For this purpose the optional keywords
+    `excludes` and `includes` are used which can both take the form of
+    a wildcard pattern or a sequence of wildcard patterns:
+
+    * ``excludes``: if given and if any pattern matches the identifier,
+      the identifier is not reported.
+    * ``includes``: if not given or if any pattern matches the identifier,
+      the identifier is reported.
+
     :param root: Root or base directory.
         Defaults to "".
     :param max_depth: Maximum recursion depth. None means limitless.
         Defaults to 1.
     :param read_only: Whether this is a read-only store.
         Defaults to False.
+    :param includes: Optional sequence of wildcards that include
+        certain filesystem paths. Affects the data identifiers (paths)
+        returned by `get_data_ids()`. By default, all paths are included.
+    :param excludes: Optional sequence of wildcards that exclude
+        certain filesystem paths. Affects the data identifiers (paths)
+        returned by `get_data_ids()`. By default, no paths are excluded.
+    :param storage_options: Parameters specific to the underlying filesystem.
+        Used to instantiate the filesystem.
     """
 
     def __init__(self,
                  root: str = '',
                  max_depth: Optional[int] = 1,
                  read_only: bool = False,
+                 includes: Optional[Sequence[str]] = None,
+                 excludes: Optional[Sequence[str]] = None,
                  storage_options: Dict[str, Any] = None):
         self._storage_options = storage_options or {}
         super().__init__(root=root,
                          max_depth=max_depth,
-                         read_only=read_only)
+                         read_only=read_only,
+                         includes=includes,
+                         excludes=excludes)
 
     @property
     def protocol(self) -> str:
diff --git a/xcube/webapi/auth/config.py b/xcube/webapi/auth/config.py
index d5c8e279..5c0b501a 100644
--- a/xcube/webapi/auth/config.py
+++ b/xcube/webapi/auth/config.py
@@ -32,7 +32,7 @@ AUTHENTICATION_SCHEMA = JsonObjectSchema(
     properties=dict(
         Authority=JsonStringSchema(format='uri'),
         Domain=JsonStringSchema(description='Deprecated, use Authority'),
-        Audience=JsonStringSchema(format='uri'),
+        Audience=JsonStringSchema(),
         Algorithms=JsonArraySchema(
             items=JsonStringSchema(default=["RS256"]),
         ),


Test patch:
diff --git a/test/core/store/fs/test_subset.py b/test/core/store/fs/test_subset.py
new file mode 100644
index 00000000..303fe8f3
--- /dev/null
+++ b/test/core/store/fs/test_subset.py
@@ -0,0 +1,192 @@
+import unittest
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+import fsspec
+
+from test.s3test import MOTO_SERVER_ENDPOINT_URL
+from test.s3test import S3Test
+from xcube.core.new import new_cube
+from xcube.core.store.fs.registry import new_fs_data_store
+from xcube.core.store.fs.store import FsDataStore
+from xcube.util.temp import new_temp_dir
+
+
+class FsStoreSubset:
+    """This class hides abstract CommonTest, so pytest will not find it."""
+
+    class CommonTest(unittest.TestCase, ABC):
+        fs: fsspec.AbstractFileSystem
+        root: str
+
+        @classmethod
+        @abstractmethod
+        def get_fs_root(cls) -> Tuple[fsspec.AbstractFileSystem, str]:
+            pass
+
+        @classmethod
+        @abstractmethod
+        def new_store(cls, **params) -> FsDataStore:
+            pass
+
+        @classmethod
+        def setUpClass(cls) -> None:
+            super().setUpClass()
+
+            cube = new_cube(width=18, height=9, x_res=20, y_res=20)
+
+            cls.fs, cls.root = cls.get_fs_root()
+
+            dir_path = f'{cls.root}/l1b'
+            cls.fs.mkdir(dir_path)
+            for i in range(3):
+                zarr_path = f'{dir_path}/olci-l1b-2022050{i + 1}.zarr'
+                cube.to_zarr(cls.fs.get_mapper(root=zarr_path), mode='w')
+
+            dir_path = f'{cls.root}/l2'
+            cls.fs.mkdir(dir_path)
+            for i in range(3):
+                zarr_path = f'{dir_path}/olci-l2-2022050{i + 1}.zarr'
+                cube.to_zarr(cls.fs.get_mapper(root=zarr_path), mode='w')
+
+            dir_path = f'{cls.root}/l3'
+            cls.fs.mkdir(dir_path)
+            for i in ['2020', '2021']:
+                levels_path = f'{dir_path}/olci-l3-{i}.levels'
+                cls.fs.mkdir(levels_path)
+                for j in range(4):
+                    zarr_path = f'{levels_path}/{j}.zarr'
+                    cube.to_zarr(cls.fs.get_mapper(root=zarr_path), mode='w')
+
+        @classmethod
+        def tearDownClass(cls) -> None:
+            cls.fs.delete(cls.root, recursive=True)
+            super().tearDownClass()
+
+        def test_no_subset(self):
+            store = self.new_store()
+            self.assertEqual({'l1b/olci-l1b-20220501.zarr',
+                              'l1b/olci-l1b-20220502.zarr',
+                              'l1b/olci-l1b-20220503.zarr',
+                              'l2/olci-l2-20220501.zarr',
+                              'l2/olci-l2-20220502.zarr',
+                              'l2/olci-l2-20220503.zarr',
+                              'l3/olci-l3-2020.levels',
+                              'l3/olci-l3-2021.levels'},
+                             set(store.get_data_ids()))
+
+        def test_include(self):
+            store = self.new_store(includes='*/*20220502*')
+            self.assertEqual({'l1b/olci-l1b-20220502.zarr',
+                              'l2/olci-l2-20220502.zarr'},
+                             set(store.get_data_ids()))
+
+            store = self.new_store(includes=['*/*20220502*', '*.levels'])
+            self.assertEqual({'l1b/olci-l1b-20220502.zarr',
+                              'l2/olci-l2-20220502.zarr',
+                              'l3/olci-l3-2020.levels',
+                              'l3/olci-l3-2021.levels'},
+                             set(store.get_data_ids()))
+
+            store = self.new_store(includes=['*.tiff'])
+            self.assertEqual(set(),
+                             set(store.get_data_ids()))
+
+        def test_exclude(self):
+            store = self.new_store(excludes='*/*20220502*')
+            self.assertEqual({'l1b/olci-l1b-20220501.zarr',
+                              'l1b/olci-l1b-20220503.zarr',
+                              'l2/olci-l2-20220501.zarr',
+                              'l2/olci-l2-20220503.zarr',
+                              'l3/olci-l3-2020.levels',
+                              'l3/olci-l3-2021.levels'},
+                             set(store.get_data_ids()))
+
+            store = self.new_store(excludes=['*/*20220502*', '*.levels'])
+            self.assertEqual({'l1b/olci-l1b-20220501.zarr',
+                              'l1b/olci-l1b-20220503.zarr',
+                              'l2/olci-l2-20220501.zarr',
+                              'l2/olci-l2-20220503.zarr'},
+                             set(store.get_data_ids()))
+
+            store = self.new_store(excludes=['*.zarr', '*.levels'])
+            self.assertEqual(set(),
+                             set(store.get_data_ids()))
+
+        def test_include_exclude(self):
+            store = self.new_store(includes='*.levels',
+                                   excludes='*2021*')
+            self.assertEqual({'l3/olci-l3-2020.levels'},
+                             set(store.get_data_ids()))
+
+            store = self.new_store(includes='*2022*',
+                                   excludes=['*.levels', 'l1b/*'])
+            self.assertEqual({'l2/olci-l2-20220501.zarr',
+                              'l2/olci-l2-20220502.zarr',
+                              'l2/olci-l2-20220503.zarr'},
+                             set(store.get_data_ids()))
+
+
+class MemoryFsStoreSubsetTest(FsStoreSubset.CommonTest):
+    @classmethod
+    def get_fs_root(cls) -> Tuple[fsspec.AbstractFileSystem, str]:
+        fs = fsspec.get_filesystem_class("memory")()
+        root = 'xcube'
+        fs.mkdirs(root, exist_ok=True)
+        return fs, root
+
+    @classmethod
+    def new_store(cls, **params) -> FsDataStore:
+        return new_fs_data_store('memory',
+                                 root=cls.root,
+                                 max_depth=3,
+                                 read_only=True,
+                                 **params)
+
+
+class FileFsStoreSubsetTest(FsStoreSubset.CommonTest):
+
+    @classmethod
+    def get_fs_root(cls) -> Tuple[fsspec.AbstractFileSystem, str]:
+        return fsspec.get_filesystem_class("file")(), new_temp_dir()
+
+    @classmethod
+    def new_store(cls, **params) -> FsDataStore:
+        return new_fs_data_store('file',
+                                 root=cls.root,
+                                 max_depth=3,
+                                 read_only=True,
+                                 **params)
+
+
+# TODO (forman): check, why we get OSError for moto test
+@unittest.skip("OSError: [Errno 5] Internal Server Error")
+class S3FsStoreSubsetTest(FsStoreSubset.CommonTest, S3Test):
+    @classmethod
+    def get_fs_root(cls) -> Tuple[fsspec.AbstractFileSystem, str]:
+        fs = fsspec.get_filesystem_class("s3")(**cls.get_storage_options())
+        root = 'xcube'
+        fs.mkdirs(root, exist_ok=True)
+        return fs, root
+
+    @classmethod
+    def new_store(cls, **params) -> FsDataStore:
+        return new_fs_data_store(
+            's3',
+            root=cls.root,
+            max_depth=3,
+            read_only=True,
+            storage_options=cls.get_storage_options(),
+            **params
+        )
+
+    @classmethod
+    def get_storage_options(cls):
+        return dict(
+            anon=True,
+            key="",
+            secret="",
+            client_kwargs=dict(
+                endpoint_url=MOTO_SERVER_ENDPOINT_URL,
+            )
+        )
diff --git a/test/s3test.py b/test/s3test.py
index e818dc95..79750168 100644
--- a/test/s3test.py
+++ b/test/s3test.py
@@ -20,6 +20,7 @@ class S3Test(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls) -> None:
+        super().setUpClass()
 
         """Mocked AWS Credentials for moto."""
         os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
@@ -48,3 +49,5 @@ class S3Test(unittest.TestCase):
     @classmethod
     def tearDownClass(cls) -> None:
         cls._moto_server.kill()
+        super().tearDownClass()
+