Skip to content

factories

Factory class for registering and creating strategy instances.

Factory wrapper methods for creating the individual strategies.

StrategiesNotLoaded

Bases: Exception

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

Source code in oteapi/plugins/factories.py
20
21
22
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
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 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
161
162
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
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: Union[StrategyConfig, dict[str, Any]],
        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."
            )

        # 'config': Must be a dict when instantiating the strategy's implementation.
        # 'config_model': Is used to retrieve the correct strategy requested.
        # Furthermore, creating 'config_model' ensures that the config is valid with
        # respect to the strategy type, further reducing the risk of incorrect logical
        # conclusions.
        if isinstance(config, dict):
            config_model = strategy_type.config_cls(**config)
        elif isinstance(config, get_args(StrategyConfig)):
            config_model = config
            config = config.model_dump(mode="json", exclude_unset=True)
        else:
            raise TypeError("config should be either of type StrategyConfig or a dict.")

        strategy_name: str = cls._get_strategy_name(config_model, 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  # type: ignore[arg-type]
            )
        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  # type: ignore[union-attr]
                if config.downloadUrl is not None  # type: ignore[union-attr]
                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
        }

    @classmethod
    def list_loaded_strategies(cls) -> dict[StrategyType, list[str]]:
        """Lists all loaded strategy plugins (endpoints).

        Returns:
            A dictionary where keys are strategy types and values are lists of
            loaded strategy names for each type.

        Raises:
            StrategiesNotLoaded: If the strategies are not loaded or
                `strategy_create_func` is not properly initialized.
        """
        if not hasattr(cls, "strategy_create_func") or not cls.strategy_create_func:
            raise StrategiesNotLoaded(
                "Strategies are not loaded or `strategy_create_func` is not properly "
                "initialized."
            )

        loaded_strategies = {}
        for strategy_type, strategy_collection in cls.strategy_create_func.items():
            # Assuming each item in the collection has a 'name' attribute or similar
            loaded_strategies[strategy_type] = [
                strategy.name for strategy in strategy_collection
            ]

        return loaded_strategies

list_loaded_strategies() classmethod

Lists all loaded strategy plugins (endpoints).

Returns:

Type Description
dict[StrategyType, list[str]]

A dictionary where keys are strategy types and values are lists of

dict[StrategyType, list[str]]

loaded strategy names for each type.

Raises:

Type Description
StrategiesNotLoaded

If the strategies are not loaded or strategy_create_func is not properly initialized.

Source code in oteapi/plugins/factories.py
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
@classmethod
def list_loaded_strategies(cls) -> dict[StrategyType, list[str]]:
    """Lists all loaded strategy plugins (endpoints).

    Returns:
        A dictionary where keys are strategy types and values are lists of
        loaded strategy names for each type.

    Raises:
        StrategiesNotLoaded: If the strategies are not loaded or
            `strategy_create_func` is not properly initialized.
    """
    if not hasattr(cls, "strategy_create_func") or not cls.strategy_create_func:
        raise StrategiesNotLoaded(
            "Strategies are not loaded or `strategy_create_func` is not properly "
            "initialized."
        )

    loaded_strategies = {}
    for strategy_type, strategy_collection in cls.strategy_create_func.items():
        # Assuming each item in the collection has a 'name' attribute or similar
        loaded_strategies[strategy_type] = [
            strategy.name for strategy in strategy_collection
        ]

    return loaded_strategies

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.

Raises:

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
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
161
162
@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 Union[StrategyConfig, dict[str, Any]]

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

Raises:

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

IStrategy

configuration, through the config parameter.

Source code in oteapi/plugins/factories.py
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 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
@classmethod
def make_strategy(
    cls,
    config: Union[StrategyConfig, dict[str, Any]],
    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."
        )

    # 'config': Must be a dict when instantiating the strategy's implementation.
    # 'config_model': Is used to retrieve the correct strategy requested.
    # Furthermore, creating 'config_model' ensures that the config is valid with
    # respect to the strategy type, further reducing the risk of incorrect logical
    # conclusions.
    if isinstance(config, dict):
        config_model = strategy_type.config_cls(**config)
    elif isinstance(config, get_args(StrategyConfig)):
        config_model = config
        config = config.model_dump(mode="json", exclude_unset=True)
    else:
        raise TypeError("config should be either of type StrategyConfig or a dict.")

    strategy_name: str = cls._get_strategy_name(config_model, 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  # type: ignore[arg-type]
        )
    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 Union[StrategyConfig, dict[str, Any]]

A strategy configuration.

required

Returns:

Type Description
IStrategy

The created strategy.

Source code in oteapi/plugins/factories.py
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def create_strategy(
    strategy_type: Union[StrategyType, str],
    config: Union[StrategyConfig, dict[str, Any]],
) -> 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)

list_strategies()

Proxy function for StrategyFactory.list_loaded_strategies().

Returns:

Type Description
dict[StrategyType, list[str]]

A dictionary where keys are strategy types and values are lists of

dict[StrategyType, list[str]]

loaded strategy names for each type.

Source code in oteapi/plugins/factories.py
206
207
208
209
210
211
212
213
214
def list_strategies() -> dict[StrategyType, list[str]]:
    """Proxy function for
    [`StrategyFactory.list_loaded_strategies()`][oteapi.plugins.factories.StrategyFactory.list_loaded_strategies].

    Returns:
        A dictionary where keys are strategy types and values are lists of
        loaded strategy names for each type.
    """
    return StrategyFactory.list_loaded_strategies()

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
192
193
194
195
196
197
198
199
200
201
202
203
def load_strategies(test_for_uniqueness: bool = True) -> None:
    """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)