factories¶

Factory class for registering and creating strategy instances.

Factory wrapper methods for creating the individual strategies.

`StrategiesNotLoaded (Exception)` ¶

Entry point strategies have not been loaded, run load_strategies().

Source code in oteapi/plugins/factories.py

class StrategiesNotLoaded(Exception):
    """Entry point strategies have not been loaded, run
    [`load_strategies()`][oteapi.plugins.factories.load_strategies]."""

`StrategyFactory` ¶

Decorator-based Factory class.

Attributes:

Name	Type	Description
`strategy_create_func`	`Dict[StrategyType, EntryPointStrategyCollection]`	An in-memory cache of all registered strategies.

Source code in oteapi/plugins/factories.py

class StrategyFactory:
    """Decorator-based Factory class.

    Attributes:
        strategy_create_func (Dict[StrategyType, EntryPointStrategyCollection]): An
            in-memory cache of all registered strategies.

    """

    strategy_create_func: "Dict[StrategyType, EntryPointStrategyCollection]"

    @classmethod
    def make_strategy(
        cls, config: "StrategyConfig", strategy_type: "Union[StrategyType, str]"
    ) -> "IStrategy":
        """Instantiate a strategy in a context class.

        Parameters:
            config: A strategy configuration.
            strategy_type: The strategy type, e.g., `"scheme"`, `"mediaType"`, ... or
                `"download"`, `"parse"`, ...
                See the [`StrategyType`][oteapi.plugins.entry_points.StrategyType]
                enumeration for a definition of valid strategy types.

        Raises:
            NotImplementedError: If the strategy cannot be found.
            ValueError: If the `strategy_type` is not a valid strategy type.
                See the [`StrategyType`][oteapi.plugins.entry_points.StrategyType]
                enumeration for a definition of valid strategy types.
            StrategiesNotLoaded: If the entry point strategies have not been loaded.

        Returns:
            An instantiated strategy. The strategy is instantiated with the provided
            configuration, through the `config` parameter.

        """
        if not hasattr(cls, "strategy_create_func"):
            raise StrategiesNotLoaded(
                "Strategies have not been loaded, run `load_strategies()` or "
                "`StrategyFactory.load_strategies()`."
            )

        if isinstance(strategy_type, str):
            try:
                strategy_type = StrategyType.init(strategy_type)
            except ValueError as exc:
                raise ValueError(
                    f"Strategy type {strategy_type!r} is not supported."
                ) from exc
        elif not isinstance(strategy_type, StrategyType):
            raise TypeError(
                "strategy_type should be either of type StrategyType or a string."
            )

        strategy_name: str = cls._get_strategy_name(config, strategy_type)

        if (strategy_type, strategy_name) in cls.strategy_create_func[strategy_type]:
            return cls.strategy_create_func[strategy_type][
                (strategy_type, strategy_name)
            ].implementation(config)
        raise NotImplementedError(
            f"The {strategy_type.value} strategy {strategy_name!r} does not exist."
        )

    @classmethod
    def _get_strategy_name(
        cls,
        config: "StrategyConfig",
        strategy_type: StrategyType,
    ) -> str:
        """Return the strategy name through the config.

        This is a method to accommodate strategy type-specific quirks to retrieve the
        strategy name.

        Parameters:
            config: A strategy configuration.
            strategy_type: The strategy type as initialized in `make_strategy()`.

        Returns:
            The strategy name provided in the configuration.

        """
        if strategy_type == StrategyType.DOWNLOAD:
            return config.downloadUrl.scheme if config.downloadUrl is not None else ""
        return getattr(config, strategy_type.map_to_field(), "")

    @classmethod
    def load_strategies(cls, test_for_uniqueness: bool = True) -> None:
        """Load strategies from entry points and store in class attribute.

        Important:
            This is not a cached method.
            The importlib.metadata API will be re-requested to load the entry points
            and strategies.

        Note:
            This does *not* import the actual strategy implementations (classes).
            It only loads the strategies from the registerred OTE-API entry points.

        Raises:
            KeyError: If `test_for_uniqueness` is `True` and an entry point strategy is
                duplicated.

        Parameters:
            test_for_uniqueness: If `True`, this will raise `KeyError` should an entry
                point strategy be duplicated. Otherwise, the first loaded entry point
                strategy will silently become the implementation of choice for the
                duplicated strategy and the duplicates will be ignored.

        """
        cls.strategy_create_func = {
            strategy_type: get_strategy_entry_points(
                strategy_type, enforce_uniqueness=test_for_uniqueness
            )
            for strategy_type in StrategyType
        }

`load_strategies(test_for_uniqueness=True)` `classmethod` ¶

Load strategies from entry points and store in class attribute.

Important

This is not a cached method. The importlib.metadata API will be re-requested to load the entry points and strategies.

Note

This does not import the actual strategy implementations (classes). It only loads the strategies from the registerred OTE-API entry points.

Exceptions:

Type	Description
`KeyError`	If `test_for_uniqueness` is `True` and an entry point strategy is duplicated.

Parameters:

Name	Type	Description	Default
`test_for_uniqueness`	`bool`	If `True`, this will raise `KeyError` should an entry point strategy be duplicated. Otherwise, the first loaded entry point strategy will silently become the implementation of choice for the duplicated strategy and the duplicates will be ignored.	`True`

Source code in oteapi/plugins/factories.py

@classmethod
def load_strategies(cls, test_for_uniqueness: bool = True) -> None:
    """Load strategies from entry points and store in class attribute.

    Important:
        This is not a cached method.
        The importlib.metadata API will be re-requested to load the entry points
        and strategies.

    Note:
        This does *not* import the actual strategy implementations (classes).
        It only loads the strategies from the registerred OTE-API entry points.

    Raises:
        KeyError: If `test_for_uniqueness` is `True` and an entry point strategy is
            duplicated.

    Parameters:
        test_for_uniqueness: If `True`, this will raise `KeyError` should an entry
            point strategy be duplicated. Otherwise, the first loaded entry point
            strategy will silently become the implementation of choice for the
            duplicated strategy and the duplicates will be ignored.

    """
    cls.strategy_create_func = {
        strategy_type: get_strategy_entry_points(
            strategy_type, enforce_uniqueness=test_for_uniqueness
        )
        for strategy_type in StrategyType
    }

`make_strategy(config, strategy_type)` `classmethod` ¶

Instantiate a strategy in a context class.

Parameters:

Name	Type	Description	Default
`config`	`StrategyConfig`	A strategy configuration.	required
`strategy_type`	`Union[StrategyType, str]`	The strategy type, e.g., `"scheme"`, `"mediaType"`, ... or `"download"`, `"parse"`, ... See the `StrategyType` enumeration for a definition of valid strategy types.	required

Exceptions:

Type	Description
`NotImplementedError`	If the strategy cannot be found.
`ValueError`	If the `strategy_type` is not a valid strategy type. See the `StrategyType` enumeration for a definition of valid strategy types.
`StrategiesNotLoaded`	If the entry point strategies have not been loaded.

Returns:

Type	Description
`IStrategy`	An instantiated strategy. The strategy is instantiated with the provided configuration, through the `config` parameter.

Source code in oteapi/plugins/factories.py

@classmethod
def make_strategy(
    cls, config: "StrategyConfig", strategy_type: "Union[StrategyType, str]"
) -> "IStrategy":
    """Instantiate a strategy in a context class.

    Parameters:
        config: A strategy configuration.
        strategy_type: The strategy type, e.g., `"scheme"`, `"mediaType"`, ... or
            `"download"`, `"parse"`, ...
            See the [`StrategyType`][oteapi.plugins.entry_points.StrategyType]
            enumeration for a definition of valid strategy types.

    Raises:
        NotImplementedError: If the strategy cannot be found.
        ValueError: If the `strategy_type` is not a valid strategy type.
            See the [`StrategyType`][oteapi.plugins.entry_points.StrategyType]
            enumeration for a definition of valid strategy types.
        StrategiesNotLoaded: If the entry point strategies have not been loaded.

    Returns:
        An instantiated strategy. The strategy is instantiated with the provided
        configuration, through the `config` parameter.

    """
    if not hasattr(cls, "strategy_create_func"):
        raise StrategiesNotLoaded(
            "Strategies have not been loaded, run `load_strategies()` or "
            "`StrategyFactory.load_strategies()`."
        )

    if isinstance(strategy_type, str):
        try:
            strategy_type = StrategyType.init(strategy_type)
        except ValueError as exc:
            raise ValueError(
                f"Strategy type {strategy_type!r} is not supported."
            ) from exc
    elif not isinstance(strategy_type, StrategyType):
        raise TypeError(
            "strategy_type should be either of type StrategyType or a string."
        )

    strategy_name: str = cls._get_strategy_name(config, strategy_type)

    if (strategy_type, strategy_name) in cls.strategy_create_func[strategy_type]:
        return cls.strategy_create_func[strategy_type][
            (strategy_type, strategy_name)
        ].implementation(config)
    raise NotImplementedError(
        f"The {strategy_type.value} strategy {strategy_name!r} does not exist."
    )

`create_strategy(strategy_type, config)` ¶

Proxy function for StrategyFactory.make_strategy().

Parameters:

Name	Type	Description	Default
`strategy_type`	`Union[StrategyType, str]`	A valid strategy type. See the `StrategyType` enumeration for a definition of valid strategy types.	required
`config`	`StrategyConfig`	A strategy configuration.	required

Returns:

Type	Description
`IStrategy`	The created strategy.

Source code in oteapi/plugins/factories.py

def create_strategy(
    strategy_type: "Union[StrategyType, str]", config: "StrategyConfig"
) -> "IStrategy":
    """Proxy function for
    [`StrategyFactory.make_strategy()`][oteapi.plugins.factories.StrategyFactory.make_strategy].

    Parameters:
        strategy_type: A valid strategy type.
            See the [`StrategyType`][oteapi.plugins.entry_points.StrategyType] enumeration
            for a definition of valid strategy types.
        config: A strategy configuration.

    Returns:
        The created strategy.

    """
    return StrategyFactory.make_strategy(config, strategy_type)

`load_strategies(test_for_uniqueness=True)` ¶

Proxy function for StrategyFactory.load_strategies().

Parameters:

Name	Type	Description	Default
`test_for_uniqueness`	`bool`	If `True`, this will raise `KeyError` should an entry point strategy be duplicated. Otherwise, the first loaded entry point strategy will silently become the implementation of choice for the duplicated strategy and the duplicates will be ignored.	`True`

Source code in oteapi/plugins/factories.py

def load_strategies(test_for_uniqueness: bool = True) -> None:
    # pylint: disable=line-too-long
    """Proxy function for
    [`StrategyFactory.load_strategies()`][oteapi.plugins.factories.StrategyFactory.load_strategies].

    Parameters:
        test_for_uniqueness: If `True`, this will raise `KeyError` should an entry
            point strategy be duplicated. Otherwise, the first loaded entry point
            strategy will silently become the implementation of choice for the
            duplicated strategy and the duplicates will be ignored.

    """
    StrategyFactory.load_strategies(test_for_uniqueness)

factories¶

StrategiesNotLoaded (Exception) ¶

StrategyFactory ¶

load_strategies(test_for_uniqueness=True) classmethod ¶

make_strategy(config, strategy_type) classmethod ¶

create_strategy(strategy_type, config) ¶

load_strategies(test_for_uniqueness=True) ¶

`StrategiesNotLoaded (Exception)` ¶

`StrategyFactory` ¶

`load_strategies(test_for_uniqueness=True)` `classmethod` ¶

`make_strategy(config, strategy_type)` `classmethod` ¶

`create_strategy(strategy_type, config)` ¶

`load_strategies(test_for_uniqueness=True)` ¶