# Copyright (c) 2018-2024 by xcube team and contributors
# Permissions are hereby granted under the terms of the MIT License:
# https://opensource.org/licenses/MIT.
from abc import abstractmethod, ABC
from typing import Iterator
from xcube.util.jsonschema import JsonObjectSchema
from .assertions import assert_valid_params
from .datatype import DataTypeLike
from .descriptor import DataDescriptor
[docs]
class DataSearcher(ABC):
"""Allow searching data in a data store."""
[docs]
@classmethod
@abstractmethod
def get_search_params_schema(
cls, data_type: DataTypeLike = None
) -> JsonObjectSchema:
"""Get the schema for the parameters that can be passed
as *search_params* to :meth:`search_data`.
Parameters are named and described by the properties of the
returned JSON object schema.
Args:
data_type: If given, the search parameters will be tailored
to search for data for the given *data_type*.
Returns:
A JSON object schema whose properties describe this store's
search parameters.
"""
[docs]
@abstractmethod
def search_data(
self, data_type: DataTypeLike = None, **search_params
) -> Iterator[DataDescriptor]:
"""Search this store for data resources.
If *data_type* is given, the search is restricted
to data resources of that type.
Returns an iterator over the search results which are
returned as :class:`DataDescriptor` objects.
If a store implementation supports only a single data type,
it should verify that *data_type*
is either None or compatible with the supported data type
specifier.
Args:
data_type: An optional data type that is known to be
supported by this data store.
**search_params: The search parameters.
Returns:
An iterator of data descriptors for the found data
resources.
Raises:
DataStoreError: If an error occurs.
"""
# noinspection PyUnresolvedReferences
class DefaultSearchMixin(DataSearcher):
"""A mixin for data store implementations that implements default
search behaviour.
It is expected that such data stores have no dedicated search
parameters.
"""
# noinspection PyUnusedLocal
@classmethod
def get_search_params_schema(
cls, data_type: DataTypeLike = None
) -> JsonObjectSchema:
"""Get search parameters JSON object schema.
The default implementation returns a schema that does
not allow for any search parameters.
Args:
data_type
Returns:
a JSON object schema for the search parameters
"""
return JsonObjectSchema(additional_properties=False)
def search_data(
self, data_type: DataTypeLike = None, **search_params
) -> Iterator[DataDescriptor]:
"""Search the data store.
The default implementation returns all data resources that
may be filtered using the optional *data_type*.
Args:
data_type: Type specifier to filter returned data resources.
**search_params: Not supported (yet)
Returns:
an iterator of :class:`DataDescriptor` instances
"""
search_params_schema = self.get_search_params_schema(data_type=data_type)
assert_valid_params(
search_params, name="search_params", schema=search_params_schema
)
for data_id in self.get_data_ids(data_type=data_type):
data_descriptor = self.describe_data(data_id, data_type=data_type)
yield data_descriptor