Skip to content

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

Bases: Exception

A given strategy implementation (class) cannot be found from the given entry point value.

Source code in oteapi/plugins/entry_points.py
55
56
57
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.

Raises:

Type Description
ValueError

If the entry point name is not properly defined.

Source code in oteapi/plugins/entry_points.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
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-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9._-]*[A-Za-z0-9])\.(?P<strategy_name>.+)$"
    )
    """Regex for entry point names.

    The package_name group is a valid non-normalized package name regex adapted from
    PEP 508.
    For more information, see: https://peps.python.org/pep-0508/#names.
    """

    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[IStrategyType] = 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}"
            f"{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) -> IStrategyType:
        """The actual strategy implementation."""
        if self._implementation is None:
            self._implementation = self._load_implementation()
        return self._implementation

    def _load_implementation(self) -> IStrategyType:
        """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: object) -> 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."
        )

ENTRY_POINT_NAME_REGEX = re.compile('^(?P<package_name>[A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9._-]*[A-Za-z0-9])\\.(?P<strategy_name>.+)$') class-attribute instance-attribute

Regex for entry point names.

The package_name group is a valid non-normalized package name regex adapted from PEP 508. For more information, see: https://peps.python.org/pep-0508/#names.

full_name: str property

The full entry point name.

implementation: IStrategyType property

The actual strategy implementation.

implementation_name: str property

The EntryPoint attr, which should be the strategy implementation class name.

module: str property

The fully resolved importable module path.

name: str property

The strategy name.

One part of the (strategy type, strategy name)-tuple.

package: str property

The importable base package name for the strategy plugin.

strategy: tuple[StrategyType, str] property

The unique index identifier for the strategy.

type: StrategyType property

The strategy type.

One part of the (strategy type, strategy name)-tuple.

EntryPointStrategyCollection

Bases: Collection

A collection of EntryPointStrategys.

Source code in oteapi/plugins/entry_points.py
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
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 (Iterable[EntryPointStrategy]): 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 (Iterable[EntryPointStrategy]): 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 (Iterable[EntryPointStrategy]): 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: object) -> 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:
        number_of_strategies: dict[str, int] = {}
        for entry_point in self._entry_points:
            if entry_point.type.value in number_of_strategies:
                number_of_strategies[entry_point.type.value] += 1
            else:
                number_of_strategies[entry_point.type.value] = 1
        sorted_list = sorted(
            f"{key} ({value})" for key, value in number_of_strategies.items()
        )
        return f"<{self.__class__.__name__}: " f"Strategies={', '.join(sorted_list)}>"

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(*{tuple(sorted(self._entry_points))!r})"

add(*entry_points)

Add entry points to the collection.

Parameters:

Name Type Description Default
*entry_points Iterable[EntryPointStrategy]

Entry points to add to the collection.

()
Source code in oteapi/plugins/entry_points.py
357
358
359
360
361
362
363
364
365
def add(self, *entry_points: EntryPointStrategy) -> None:
    """Add entry points to the collection.

    Parameters:
        *entry_points (Iterable[EntryPointStrategy]): Entry points to add to the
            collection.

    """
    self._entry_points |= set(entry_points)

exclusive_add(*entry_points)

Exclusively add entry points to the collection.

Parameters:

Name Type Description Default
*entry_points Iterable[EntryPointStrategy]

Entry points to add to the collection.

()

Raises:

Type Description
KeyError

If an entry point to be added already exists in the collection.

Source code in oteapi/plugins/entry_points.py
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def exclusive_add(self, *entry_points: EntryPointStrategy) -> None:
    """Exclusively add entry points to the collection.

    Parameters:
        *entry_points (Iterable[EntryPointStrategy]): 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(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

Raises:

Type Description
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:

Type Description
EntryPointStrategy

An entry point in the collection representing the key.

Source code in oteapi/plugins/entry_points.py
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
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(*entry_points)

Remove entry points from the collection.

Parameters:

Name Type Description Default
*entry_points Iterable[EntryPointStrategy]

Entry points to remove from the collection.

()
Source code in oteapi/plugins/entry_points.py
367
368
369
370
371
372
373
374
375
def remove(self, *entry_points: EntryPointStrategy) -> None:
    """Remove entry points from the collection.

    Parameters:
        *entry_points (Iterable[EntryPointStrategy]): Entry points to remove from
            the collection.

    """
    self._entry_points -= set(entry_points)

StrategyType

Bases: Enum

An enumeration of available strategy types.

Available strategy types:

  • download
  • filter
  • function
  • mapping
  • parse
  • resource
  • transformation
Source code in oteapi/plugins/entry_points.py
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class StrategyType(Enum):
    """An enumeration of available strategy types.

    Available strategy types:

    - download
    - filter
    - function
    - 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": "parserType",
            "resource": "resourceType",
            "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,
            "parserType": cls.PARSE,
            "resourceType": 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
    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))

    @property
    def config_cls(self) -> StrategyConfigType:
        """Return the strategy-specific `*Config` class."""
        return {  # type: ignore[return-value]
            "download": ResourceConfig,
            "filter": FilterConfig,
            "function": FunctionConfig,
            "mapping": MappingConfig,
            "parse": ParserConfig,
            "resource": ResourceConfig,
            "transformation": TransformationConfig,
        }[self.value]

config_cls: StrategyConfigType property

Return the strategy-specific *Config class.

all_values() classmethod

Return all values.

Source code in oteapi/plugins/entry_points.py
138
139
140
141
@classmethod
def all_values(cls) -> tuple[str, ...]:
    """Return all values."""
    return tuple(strategy_type.value for strategy_type in cls)

init(value) classmethod

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:

Type Description
ValueError

As normal if the enumeration value is not valid.

Source code in oteapi/plugins/entry_points.py
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@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)

map_from_field(strategy_type_field) classmethod

Map the strategy type's field to enumeration.

Parameters:

Name Type Description Default
strategy_type_field str

The strategy type's field. E.g., scheme for download.

required

Raises:

Type Description
KeyError

If the strategy_type_field is not valid.

Returns:

Type Description
StrategyType

An enumeration instance representing the strategy type's field.

Source code in oteapi/plugins/entry_points.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@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,
        "parserType": cls.PARSE,
        "resourceType": cls.RESOURCE,
        "transformationType": cls.TRANSFORMATION,
    }[strategy_type_field]

map_to_field()

Map enumeration value to the strategy type's field.

Source code in oteapi/plugins/entry_points.py
83
84
85
86
87
88
89
90
91
92
93
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": "parserType",
        "resource": "resourceType",
        "transformation": "transformationType",
    }[self.value]

get_strategy_entry_points(strategy_type, enforce_uniqueness=True)

Retrieve all entry points from a specific strategy type.

Raises:

Type Description
ValueError

If the strategy type is not supported.

KeyError

If enforce_uniqueness is True and an entry point strategy is duplicated.

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, meaning duplicates are not allowed.

True

Returns:

Type Description
EntryPointStrategyCollection

A collection of entry points for the specific strategy type.

Source code in oteapi/plugins/entry_points.py
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
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:
        str_strategy_type = (
            strategy_type
            if isinstance(strategy_type, str)
            else str(strategy_type.value)
        )
        raise ValueError(
            f"Strategy type {str_strategy_type!r} is not supported."
        ) from exc

    collection = EntryPointStrategyCollection()
    oteapi_entry_points = sorted(
        set(get_entry_points(group=f"oteapi.{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