entry_points¶
Load plugins through entry points.
This module deals with handling all plugged in strategies through the entry points API and importlib metadata API.
Special functionality is put in place to handle the OTE-API-specific entry points.
Since the entry points are information complete in and of themselves, there is no need to import actual strategy classes until they are truly needed. This therefore implements lazy loading of all plugin strategies.
EntryPointNotFound (Exception)
¶
A given strategy implementation (class) cannot be found from the given entry point value.
Source code in oteapi/plugins/entry_points.py
class EntryPointNotFound(Exception):
"""A given strategy implementation (class) cannot be found from the given entry
point value."""
EntryPointStrategy
¶
A strategy realized from an entry point.
An entry point strategy is made unique by its "strategy", i.e., its
(strategy type, strategy name)-tuple, e.g., ("download", "https")
.
This tuple can be retrieved from the
strategy
property,
where the strategy type is represented by the
StrategyType
enumeration.
Note
It may be that in the future an entry points strategy is made unique by
its "full name" instead, i.e., the entry point group + the entry points name,
e.g., oteapi.download:oteapi.https
.
This value can be retrieved from the
full_name
property.
This is a condition for uniqueness that is considered to be a superset of the current condition for uniqueness. It adds an extra package-specific uniqueness trait, allowing for different packages to implement the same strategies (which is currently not allowed according to the condition of uniqueness explained above).
Currently there is no consensus on the API for handling this added strategy ambiguity.
Exceptions:
Type | Description |
---|---|
ValueError |
If the entry point name is not properly defined. |
Source code in oteapi/plugins/entry_points.py
class EntryPointStrategy:
"""A strategy realized from an entry point.
An entry point strategy is made unique by its "strategy", i.e., its
(strategy type, strategy name)-tuple, e.g., `("download", "https")`.
This tuple can be retrieved from the
[`strategy`][oteapi.plugins.entry_points.EntryPointStrategy.strategy] property,
where the strategy type is represented by the
[`StrategyType`][oteapi.plugins.entry_points.StrategyType] enumeration.
Note:
It may be that in the future an entry points strategy is made unique by
its "full name" instead, i.e., the entry point group + the entry points name,
e.g., `oteapi.download:oteapi.https`.
This value can be retrieved from the
[`full_name`][oteapi.plugins.entry_points.EntryPointStrategy.full_name]
property.
This is a condition for uniqueness that is considered to be a superset of the
current condition for uniqueness.
It adds an extra package-specific uniqueness trait, allowing for different
packages to implement the same strategies (which is currently not allowed
according to the condition of uniqueness explained above).
Currently there is no consensus on the API for handling this added strategy
ambiguity.
Raises:
ValueError: If the entry point name is not properly defined.
"""
ENTRY_POINT_NAME_REGEX = re.compile(
r"^(?P<package_name>[a-z_]+)\.(?P<strategy_name>.+)$"
)
ENTRY_POINT_NAME_SEPARATOR = ":"
def __init__(self, entry_point: "EntryPoint") -> None:
self._entry_point = entry_point
match = self.ENTRY_POINT_NAME_REGEX.match(self._entry_point.name)
if match is None:
raise ValueError(
"Could not determine package name and/or strategy name for entry "
f"point: {self.full_name}"
)
self._match = match
self._type = StrategyType(self._entry_point.group[len("oteapi.") :])
self._implementation: "Optional[Type[IStrategy]]" = None
@property
def type(self) -> StrategyType:
"""The strategy type.
One part of the (strategy type, strategy name)-tuple.
"""
return self._type
@property
def name(self) -> str:
"""The strategy name.
One part of the (strategy type, strategy name)-tuple.
"""
return self._match.group("strategy_name")
@property
def strategy(self) -> "Tuple[StrategyType, str]":
"""The unique index identifier for the strategy."""
return self.type, self.name
@property
def package(self) -> str:
"""The importable base package name for the strategy plugin."""
return self._match.group("package_name")
@property
def module(self) -> str:
"""The fully resolved importable module path."""
return self._entry_point.module # type: ignore[attr-defined]
@property
def full_name(self) -> str:
"""The full entry point name."""
return f"{self._entry_point.group}{self.ENTRY_POINT_NAME_SEPARATOR}{self._entry_point.name}"
def __str__(self) -> str:
return self.full_name
def __repr__(self) -> str:
return f"{self.__class__.__name__}(entry_point={self._entry_point!r})"
@property
def implementation_name(self) -> str:
"""The EntryPoint attr, which should be the strategy implementation class
name."""
return self._entry_point.attr # type: ignore[attr-defined]
@property
def implementation(self) -> "Type[IStrategy]":
"""The actual strategy implementation."""
if self._implementation is None:
self._implementation = self._load_implementation()
return self._implementation
def _load_implementation(self) -> "Type[IStrategy]":
"""Load the strategy implementation.
Because the actual importing of the module does not happen until this method is
called, we are lazily loading in the strategy implementation.
There is no need to check through the `globals()` built-in for whether the
module and class have already been imported, because this caching layer is
already implemented in the `importlib` API.
Raises:
EntryPointNotFound: If the strategy implementation (class) the entry point
is pointing to cannot be found in the module or if the module cannot be
imported.
Returns:
The imported strategy implementation (class).
"""
try:
module = importlib.import_module(self.module)
except ImportError as exc:
raise EntryPointNotFound(
f"{self.module} cannot be imported. Did you install the "
f"{self.package!r} package?"
) from exc
if hasattr(module, self.implementation_name):
return getattr(module, self.implementation_name)
raise EntryPointNotFound(
f"{self.implementation_name} cannot be found in {self.module}"
)
def __eq__(self, other: "Any") -> bool:
if isinstance(other, self.__class__):
return hash(self) == hash(other)
return False
def __hash__(self) -> int:
return hash(self.strategy)
def __lt__(self, other: "Any") -> bool:
"""Whether or not `self` is less than (`<`) `other`.
This is implemented to allow sorting (using `sorted()`).
The inequality is determined on the basis of the following properties:
1. [`type`][oteapi.plugins.entry_points.EntryPointStrategy.type]
2. [`package`][oteapi.plugins.entry_points.EntryPointStrategy.package]
3. [`name`][oteapi.plugins.entry_points.EntryPointStrategy.name]
Going from highest priority to lowest and in alphabetical ascending order.
"""
if isinstance(other, self.__class__):
if self.type == other.type:
if self.package == other.package:
if self.name == other.name:
# Considered equal, i.e., one can by definition not be unequal
# with the other.
return False
return sorted([self.name, other.name])[0] == self.name
return sorted([self.package, other.package])[0] == self.package
return sorted([self.type.value, other.type.value])[0] == self.type.value
raise NotImplementedError(
f"Less than comparison is not implemented for {type(other)} type objects."
)
full_name: str
property
readonly
¶
The full entry point name.
implementation: Type[IStrategy]
property
readonly
¶
The actual strategy implementation.
implementation_name: str
property
readonly
¶
The EntryPoint attr, which should be the strategy implementation class name.
module: str
property
readonly
¶
The fully resolved importable module path.
name: str
property
readonly
¶
The strategy name.
One part of the (strategy type, strategy name)-tuple.
package: str
property
readonly
¶
The importable base package name for the strategy plugin.
strategy: Tuple[StrategyType, str]
property
readonly
¶
The unique index identifier for the strategy.
type: StrategyType
property
readonly
¶
The strategy type.
One part of the (strategy type, strategy name)-tuple.
EntryPointStrategyCollection (Collection)
¶
A collection of
EntryPointStrategy
s.
Source code in oteapi/plugins/entry_points.py
class EntryPointStrategyCollection(abc.Collection):
"""A collection of
[`EntryPointStrategy`][oteapi.plugins.entry_points.EntryPointStrategy]s."""
def __init__(self, *entry_points: "EntryPointStrategy") -> None:
self._entry_points: "Set[EntryPointStrategy]" = (
set(entry_points) if entry_points else set()
)
def add(self, *entry_points: EntryPointStrategy) -> None:
"""Add entry points to the collection.
Parameters:
entry_points: Entry points to add to the collection.
"""
self._entry_points |= set(entry_points)
def remove(self, *entry_points: EntryPointStrategy) -> None:
"""Remove entry points from the collection.
Parameters:
entry_points: Entry points to remove from the collection.
"""
self._entry_points -= set(entry_points)
def exclusive_add(self, *entry_points: EntryPointStrategy) -> None:
"""Exclusively add entry points to the collection.
Parameters:
entry_points: Entry points to add to the collection.
Raises:
KeyError: If an entry point to be added already exists in the collection.
"""
for entry_point in entry_points:
if entry_point in self:
raise KeyError(
f"{entry_point.strategy} already exists in {self}. "
f"(Tried adding {entry_point}.)"
)
self.add(entry_point)
def __len__(self) -> int:
return len(self._entry_points)
def __contains__(self, item: "Any") -> bool:
"""Whether or not `item` is contained in the collection.
One can test with an `EntryPointStrategy`, a string of an entry point
strategy's full name, or a tuple of an entry point's strategy type and name.
Parameters:
item: Item to test whether it is contained in the collection.
Returns:
Whether or not `item` is contained in the collection.
If the `item` is an unrecognized type, `False` will be returned.
"""
if isinstance(item, EntryPointStrategy):
return item in self._entry_points
if isinstance(item, str):
for entry_point in self._entry_points:
if item == entry_point.full_name:
return True
if isinstance(item, tuple):
if len(item) != 2 or (
not isinstance(item[0], (StrategyType, str))
or not isinstance(item[1], str)
):
# Only tuples of type (Union[StrategyType, str], str) are allowed.
return False
try:
item_ = (StrategyType.init(item[0]), item[1])
except ValueError:
# We only want to return True or False
return False
for entry_point in self._entry_points:
if item_ == entry_point.strategy:
return True
# For any other type:
return False
def __iter__(self) -> "Iterator[EntryPointStrategy]":
yield from self._entry_points
def __getitem__(self, key: "Any") -> EntryPointStrategy:
return self.get_entry_point(key)
def get_entry_point(
self,
key: "Union[EntryPointStrategy, str, Tuple[Union[StrategyType, str], str]]",
) -> EntryPointStrategy:
"""Retrieve an entry point from the collection.
Parameters:
key: The key to check for in the collection.
Raises:
KeyError: If an entry point cannot be found in the collection.
TypeError: If the `key` is not of an expected type. See the `key` parameter
above for the expected types.
Returns:
An entry point in the collection representing the key.
"""
if isinstance(key, (EntryPointStrategy, str, tuple)):
if key not in self:
raise KeyError(f"{key} not found in {self}")
return self._get_entry_point(key)
raise TypeError(
"key should either be of type EntryPointStrategy, a string of the full "
"name or a strategy tuple."
)
def _get_entry_point(
self,
key: "Union[EntryPointStrategy, str, Tuple[Union[StrategyType, str], str]]",
) -> EntryPointStrategy:
"""Helper method for retrieving an entry point from the collection.
Important:
It is expected that the entry point representing the key exists in the
collection. For example through a `key in self` test.
Parameters:
key: The key to check for in the collection.
Raises:
RuntimeError: If an entry point cannot be found in the collection, since
this is a requirement.
Returns:
An entry point in the collection representing the key.
"""
if isinstance(key, EntryPointStrategy):
return key
if isinstance(key, str):
for entry_point in self._entry_points:
if key == entry_point.full_name:
return entry_point
if isinstance(key, tuple):
key_ = (StrategyType(key[0]), key[1])
for entry_point in self._entry_points:
if key_ == entry_point.strategy:
return entry_point
raise RuntimeError(
f"{key} not found in {self}, which is a requirement for the "
"_get_entry_point method."
)
def __eq__(self, other: "Any") -> bool:
if isinstance(other, self.__class__):
return hash(self) == hash(other)
return False
def __hash__(self) -> int:
return hash(tuple(_ for _ in sorted(self._entry_points)))
def __str__(self) -> str:
res = {} # type: ignore[var-annotated]
for _ in self._entry_points:
if _.type.value in res:
res[_.type.value] += 1
else:
res[_.type.value] = 1
res = sorted(f"{key} ({value})" for key, value in res.items())
return f"<{self.__class__.__name__}: " f"Strategies={', '.join(res)}>"
def __repr__(self) -> str:
return f"{self.__class__.__name__}(*{tuple(sorted(self._entry_points))!r})"
add(self, *entry_points)
¶
Add entry points to the collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
entry_points |
EntryPointStrategy |
Entry points to add to the collection. |
() |
Source code in oteapi/plugins/entry_points.py
def add(self, *entry_points: EntryPointStrategy) -> None:
"""Add entry points to the collection.
Parameters:
entry_points: Entry points to add to the collection.
"""
self._entry_points |= set(entry_points)
exclusive_add(self, *entry_points)
¶
Exclusively add entry points to the collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
entry_points |
EntryPointStrategy |
Entry points to add to the collection. |
() |
Exceptions:
Type | Description |
---|---|
KeyError |
If an entry point to be added already exists in the collection. |
Source code in oteapi/plugins/entry_points.py
def exclusive_add(self, *entry_points: EntryPointStrategy) -> None:
"""Exclusively add entry points to the collection.
Parameters:
entry_points: Entry points to add to the collection.
Raises:
KeyError: If an entry point to be added already exists in the collection.
"""
for entry_point in entry_points:
if entry_point in self:
raise KeyError(
f"{entry_point.strategy} already exists in {self}. "
f"(Tried adding {entry_point}.)"
)
self.add(entry_point)
get_entry_point(self, key)
¶
Retrieve an entry point from the collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
key |
Union[EntryPointStrategy, str, Tuple[Union[StrategyType, str], str]] |
The key to check for in the collection. |
required |
Exceptions:
Type | Description |
---|---|
KeyError |
If an entry point cannot be found in the collection. |
TypeError |
If the |
Returns:
Type | Description |
---|---|
EntryPointStrategy |
An entry point in the collection representing the key. |
Source code in oteapi/plugins/entry_points.py
def get_entry_point(
self,
key: "Union[EntryPointStrategy, str, Tuple[Union[StrategyType, str], str]]",
) -> EntryPointStrategy:
"""Retrieve an entry point from the collection.
Parameters:
key: The key to check for in the collection.
Raises:
KeyError: If an entry point cannot be found in the collection.
TypeError: If the `key` is not of an expected type. See the `key` parameter
above for the expected types.
Returns:
An entry point in the collection representing the key.
"""
if isinstance(key, (EntryPointStrategy, str, tuple)):
if key not in self:
raise KeyError(f"{key} not found in {self}")
return self._get_entry_point(key)
raise TypeError(
"key should either be of type EntryPointStrategy, a string of the full "
"name or a strategy tuple."
)
remove(self, *entry_points)
¶
Remove entry points from the collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
entry_points |
EntryPointStrategy |
Entry points to remove from the collection. |
() |
Source code in oteapi/plugins/entry_points.py
def remove(self, *entry_points: EntryPointStrategy) -> None:
"""Remove entry points from the collection.
Parameters:
entry_points: Entry points to remove from the collection.
"""
self._entry_points -= set(entry_points)
StrategyType (Enum)
¶
An enumeration of available strategy types.
Available strategy types:
- download
- filter
- mapping
- parse
- resource
- transformation
Source code in oteapi/plugins/entry_points.py
class StrategyType(Enum):
"""An enumeration of available strategy types.
Available strategy types:
- download
- filter
- mapping
- parse
- resource
- transformation
"""
DOWNLOAD = "download"
FILTER = "filter"
FUNCTION = "function"
MAPPING = "mapping"
PARSE = "parse"
RESOURCE = "resource"
TRANSFORMATION = "transformation"
def map_to_field(self) -> str:
"""Map enumeration value to the strategy type's field."""
return {
"download": "scheme",
"filter": "filterType",
"function": "functionType",
"mapping": "mappingType",
"parse": "mediaType",
"resource": "accessService",
"transformation": "transformationType",
}[self.value]
@classmethod
def map_from_field(cls, strategy_type_field: str) -> "StrategyType":
"""Map the strategy type's field to enumeration.
Parameters:
strategy_type_field: The strategy type's field. E.g., `scheme` for
`download`.
Raises:
KeyError: If the `strategy_type_field` is not valid.
Returns:
An enumeration instance representing the strategy type's field.
"""
return {
"scheme": cls.DOWNLOAD,
"filterType": cls.FILTER,
"functionType": cls.FUNCTION,
"mappingType": cls.MAPPING,
"mediaType": cls.PARSE,
"accessService": cls.RESOURCE,
"transformationType": cls.TRANSFORMATION,
}[strategy_type_field]
@classmethod
def init(cls, value: "Union[str, StrategyType]") -> "StrategyType":
"""Initialize a StrategyType with more than just the enumeration value.
This method allows one to also initialize a StrategyType with an actual
strategy type string, e.g., `scheme`, `mediaType`, etc.
Raises:
ValueError: As normal if the enumeration value is not valid.
"""
if isinstance(value, str):
try:
return cls.map_from_field(value)
except KeyError:
pass
return cls(value)
@classmethod
@lru_cache
def all_values(cls) -> "Tuple[str, ...]":
"""Return all values."""
return tuple(strategy_type.value for strategy_type in cls)
def __str__(self) -> str:
return str(self.value)
def __repr__(self) -> str:
return repr(str(self))
get_strategy_entry_points(strategy_type, enforce_uniqueness=True)
¶
Retrieve all entry points from a specific strategy type.
Exceptions:
Type | Description |
---|---|
ValueError |
If the strategy type is not supported. |
KeyError |
If |
Parameters:
Name | Type | Description | Default |
---|---|---|---|
strategy_type |
Union[StrategyType, str] |
A strategy type for which the entry points will be retrieved. |
required |
enforce_uniqueness |
bool |
Whether or not duplicate entry point strategies are
allowed. Defaults to |
True |
Returns:
Type | Description |
---|---|
EntryPointStrategyCollection |
A collection of entry points for the specific strategy type. |
Source code in oteapi/plugins/entry_points.py
def get_strategy_entry_points(
strategy_type: "Union[StrategyType, str]",
enforce_uniqueness: bool = True,
) -> EntryPointStrategyCollection:
"""Retrieve all entry points from a specific strategy type.
Raises:
ValueError: If the strategy type is not supported.
KeyError: If `enforce_uniqueness` is `True` and an entry point strategy is
duplicated.
Parameters:
strategy_type: A strategy type for which the entry points will be retrieved.
enforce_uniqueness: Whether or not duplicate entry point strategies are
allowed. Defaults to `True`, meaning duplicates are *not* allowed.
Returns:
A collection of entry points for the specific strategy type.
"""
try:
strategy_type = StrategyType(strategy_type)
except ValueError as exc:
raise ValueError(
"Strategy type "
f"{strategy_type if isinstance(strategy_type, str) else str(strategy_type.value)!r}"
" is not supported."
) from exc
collection = EntryPointStrategyCollection()
for group, oteapi_entry_points in get_entry_points().items():
if group.startswith("oteapi.") and group[len("oteapi.") :] == str(
strategy_type.value
):
if enforce_uniqueness:
collection.exclusive_add(
*(EntryPointStrategy(_) for _ in oteapi_entry_points)
)
else:
collection.add(*(EntryPointStrategy(_) for _ in oteapi_entry_points))
return collection