Source code for xcube.util.extension

# The MIT License (MIT)
# Copyright (c) 2019 by the xcube development team and contributors
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of
# this software and associated documentation files (the "Software"), to deal in
# the Software without restriction, including without limitation the rights to
# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
# of the Software, and to permit persons to whom the Software is furnished to do
# so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.


import importlib
from typing import List, Any, Dict, Callable, Mapping, Sequence, Optional

__author__ = "Norman Fomferra (Brockmann Consult GmbH)"

from xcube.util.ipython import register_json_formatter

Component = Any
ComponentLoader = Callable[['Extension'], Component]
ComponentTransform = Callable[[Component, 'Extension'], Component]
ExtensionPredicate = Callable[['Extension'], bool]


[docs]class Extension: """ An extension that provides a component of any type. Extensions are registered in a :class:`ExtensionRegistry`. Extension objects are not meant to be instantiated directly. Instead, :meth:`ExtensionRegistry.add_extension` is used to register extensions. :param point: extension point identifier :param name: extension name :param component: extension component :param loader: extension component loader function :param metadata: extension metadata """ # noinspection PyShadowingBuiltins def __init__(self, point: str, name: str, component: Component = None, loader: ComponentLoader = None, **metadata): if point is None: raise ValueError(f'point must be given') if name is None: raise ValueError(f'name must be given') if (loader is not None and component is not None) or (loader is None and component is None): raise ValueError(f'either component or loader must be given') if loader is not None and not callable(loader): raise ValueError(f'loader must be callable') self._component = component self._loader = loader self._point = point self._name = name self._metadata = metadata self._deleted = False @property def is_lazy(self) -> bool: """Whether this is a lazy extension that uses a loader.""" return self._loader is not None @property def component(self) -> Component: """Extension component.""" if self._component is None and self._loader is not None: self._component = self._loader(self) return self._component @property def point(self) -> str: """Extension point identifier.""" return self._point @property def name(self) -> str: """Extension name.""" return self._name @property def metadata(self) -> Dict[str, Any]: """Extension metadata.""" return dict(self._metadata)
[docs] def to_dict(self) -> Dict[str, Any]: """ Get a JSON-serializable dictionary representation of this extension. """ # Note: we avoid loading the component! if self._component is not None: if hasattr(self._component, 'to_dict') and callable(getattr(self._component, 'to_dict')): component = self._component.to_dict() else: component = repr(self._component) else: component = '<not loaded yet>' d = dict( name=self.name, **self.metadata, point=self.point, component=component, ) return d
register_json_formatter(Extension) # noinspection PyShadowingBuiltins
[docs]class ExtensionRegistry: """ A registry of extensions. Typically used by plugins to register extensions. """ def __init__(self): self._extension_points = {}
[docs] def has_extension(self, point: str, name: str) -> bool: """ Test if an extension with given *point* and *name* is registered. :param point: extension point identifier :param name: extension name :return: True, if extension exists """ return point in self._extension_points and name in self._extension_points[point]
[docs] def get_extension(self, point: str, name: str) -> Optional[Extension]: """ Get registered extension for given *point* and *name*. :param point: extension point identifier :param name: extension name :return: the extension or None, if no such exists """ if point not in self._extension_points: return None return self._extension_points[point].get(name)
[docs] def get_component(self, point: str, name: str) -> Any: """ Get extension component for given *point* and *name*. Raises a ValueError if no such extension exists. :param point: extension point identifier :param name: extension name :return: extension component """ extension = self.get_extension(point, name) if extension is None: raise ValueError(f'extension {name!r} not found for extension point {point!r}') return extension.component
[docs] def find_extensions(self, point: str, predicate: ExtensionPredicate = None) -> List[Extension]: """ Find extensions for *point* and optional filter function *predicate*. The filter function is called with an extension and should return a truth value to indicate a match or mismatch. :param point: extension point identifier :param predicate: optional filter function :return: list of matching extensions """ if point not in self._extension_points: return [] point_extensions = self._extension_points[point] if predicate is None: return list(point_extensions.values()) return [extension for extension in point_extensions.values() if predicate(extension)]
[docs] def find_components(self, point: str, predicate: ExtensionPredicate = None) -> List[Component]: """ Find extension components for *point* and optional filter function *predicate*. The filter function is called with an extension and should return a truth value to indicate a match or mismatch. :param point: extension point identifier :param predicate: optional filter function :return: list of matching extension components """ return [extension.component for extension in self.find_extensions(point, predicate=predicate)]
[docs] def add_extension(self, point: str, name: str, component: Component = None, loader: ComponentLoader = None, **metadata) -> Extension: """ Register an extension *component* or an extension component *loader* for the given extension *point*, *name*, and additional *metadata*. Either *component* or *loader* must be specified, but not both. A given *loader* must be a callable with one positional argument *extension* of type :class:`Extension` and is expected to return the actual extension component, which may be of any type. The *loader* will only be called once and only when the actual extension component is requested for the first time. Consider using the function :func:`import_component` to create a loader that lazily imports a component from a module and optionally executes it. :param point: extension point identifier :param name: extension name :param component: extension component :param loader: extension component loader function :param metadata: extension metadata :return: a registered extension """ extension = Extension(point, name, component=component, loader=loader, **metadata) if point in self._extension_points: self._extension_points[point][name] = extension else: self._extension_points[point] = {name: extension} return extension
[docs] def remove_extension(self, point: str, name: str): """ Remove registered extension *name* from given *point*. :param point: extension point identifier :param name: extension name """ point_extensions = self._extension_points[point] del point_extensions[name]
[docs] def to_dict(self): """ Get a JSON-serializable dictionary representation of this extension registry. """ return {k: {ek: ev.to_dict() for ek, ev in v.items()} for k, v in self._extension_points.items()}
register_json_formatter(ExtensionRegistry) _EXTENSION_REGISTRY_SINGLETON = ExtensionRegistry() def get_extension_registry() -> ExtensionRegistry: """Return the extension registry singleton.""" return _EXTENSION_REGISTRY_SINGLETON
[docs]def import_component(spec: str, transform: ComponentTransform = None, call: bool = False, call_args: Sequence[Any] = None, call_kwargs: Mapping[str, Any] = None) -> ComponentLoader: """ Return a component loader that imports a module or module component from *spec*. To import a module, *spec* should be the fully qualified module name. To import a component, *spec* must also append the component name to the fully qualified module name separated by a color (":") character. An optional *transform* callable my be used to transform the imported component. If given, a new component is computed:: component = transform(component, extension) If the *call* flag is set, the component is expected to be a callable which will be called using the given *call_args* and *call_kwargs* to produce a new component:: component = component(*call_kwargs, **call_kwargs) Finally, the component is returned. :param spec: String of the form "module_path" or "module_path:component_name" :param transform: callable that takes two positional arguments, the imported component and the extension of type :class:`Extension` :param call: Whether to finally call the component with given *call_args* and *call_kwargs* :param call_args: arguments passed to a callable component if *call* flag is set :param call_kwargs: keyword arguments passed to callable component if *call* flag is set :return: a component loader """ # noinspection PyUnusedLocal def _load(extension: Extension): nonlocal spec component = _import_component(spec, force_component=call) if transform is not None: component = transform(component, extension) if call or call_args or call_kwargs: component = component(*(call_args or []), **(call_kwargs or {})) return component return _load
def _import_component(component_spec: str, force_component: bool = False): """ Import a module or module component from *spec*. :param component_spec: String of the form "module_name" or "module_name:component_name" where module_name must an absolute, fully qualified path to a module. :param force_component: If True, *spec* must specify a component name :return: the imported module or module component """ if ':' in component_spec: module_name, component_name = component_spec.split(':', maxsplit=1) else: module_name, component_name = component_spec, None if force_component and component_name is None: raise ValueError('illegal spec, must specify a component') if module_name == '': raise ValueError('illegal spec, missing module path') if component_name == '': raise ValueError('illegal spec, missing component name') module = importlib.import_module(module_name) return getattr(module, component_name) if component_name else module