Source code for message_ix_models.util.importlib

"""Load model and project code from :mod:`message_data`."""

import functools
import re
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from importlib import import_module, util
from importlib.abc import MetaPathFinder
from importlib.machinery import ModuleSpec, SourceFileLoader
from importlib.metadata import version
from itertools import chain
from logging import INFO, getLogger
from platform import python_version
from typing import TYPE_CHECKING, Optional

from packaging.version import parse

from ._logging import once

if TYPE_CHECKING:
    from typing import Protocol

    class Condition(Protocol):
        def __call__(self) -> str: ...

    class MinimumVersionDecorated(Protocol):
        def __call__(self, *args, **kwargs): ...
        def minimum_version(self, to_wrap): ...


[docs] class MessageDataFinder(MetaPathFinder): """Load model and project code from :mod:`message_data`. This class allows for future-proof import statements. For example, if there is a module :py:`message_data.project.foo.bar`, code can be written like: .. code-block:: python from message_ix_models.project.foo import bar The :meth:`find_spec` method locates the corresponding submodule in :mod:`message_data` and imports it as if it were in :mod:`message_ix_models`. Later, if the module is migrated to :py:`message_ix_models.project.foo.bar`, the import statement will work directly, without changes or use of MessageDataFinder. Where the *same* module names exist within both packages, the submodule of :mod:`message_ix_models` will be found directly and MessageDataFinder will not be invoked. This behaviour also allows to mix model and project code in the two packages, although this **should** be avoided where possible. """ #: Expression for supported module names. expr = re.compile(r"message_ix_models\.(?P<name>(model|project)\..*)") @classmethod def find_spec(cls, name: str, path, target=None): from .common import HAS_MESSAGE_DATA if not HAS_MESSAGE_DATA: return None if match := cls.expr.match(name): # Construct the name for the actual module to load new_name = f"message_data.{match.group('name')}" else: return None try: # Get an import spec for the message_data submodule spec = util.find_spec(new_name) except ImportError: # `new_name` does not exist as a submodule of message_data return None else: # pragma: no cover # NB Coverage ignored because message_data is not installed on GHA assert spec is not None and spec.origin is not None once(getLogger(__name__), INFO, f"Import {new_name!r} as {name!r}") # - Create a new spec that loads message_data.model.foo as if it were # message_ix_models.model.foo # - Create a new loader that loads from the actual file with the desired # name new_spec = ModuleSpec( name=name, loader=SourceFileLoader(fullname=name, path=spec.origin), origin=spec.origin, ) # These can't be passed through the constructor new_spec.submodule_search_locations = spec.submodule_search_locations return new_spec
@dataclass(frozen=True) class _PackageVersion: """Condition that the version of package `name` is at least `v_min`.""" name: str v_min: str @functools.cache def __call__(self) -> str: v = python_version() if self.name == "python" else version(self.name) return f"{self.name} {v} < {self.v_min}" if parse(v) < parse(self.v_min) else "" @dataclass(frozen=True) class _Recurse: """Recurse to the minimum version decorator of `fully_qualified_name`.""" fully_qualified_name: str @functools.cache def __call__(self) -> str: # Split the fully qualified name into a module name and item name module_name, _, name = self.fully_qualified_name.strip().rpartition(".") # Import the module → retrieve the item → retrieve its MVD return getattr(import_module(module_name), name)._mvd.check_versions()
[docs] class MinimumVersionDecorator: """Mark callable objects as requiring minimum version(s) of upstream packages. If the decorated object is called and any of condition(s) in `expr` is not met, :class:`.NotImplementedError` is raised with an informative message. The decorated object gains an attribute :py:`.minimum_version`, which can be used like :py:`pytest.mark.xfail()` to decorate test functions. This marks the test as XFAIL, raising :class:`.NotImplementedError` directly; indirectly :class:`.RuntimeError` or :class:`.AssertionError` (for instance, via :mod:`.click` test utilities or :mod:`genno`), or any of the classes given by the `raises` argument. See :func:`.prepare_reporter` / :func:`.test_prepare_reporter` for a usage example. Parameters ---------- expr : Zero or more conditions like: 1. "pkgA 1.2.3.post0; pkgB 2025.2", specifying minimum version of 1 or more packages. 2. "python 3.10", specifying a minimum version of python. 3. "message_ix_models.foo.bar.baz", recursively referring to the minimum version required by :py:`baz` in the module :py:`message_ix_models.foo.bar`. This object **must** also have been decorated with :class:`MinimumVersionDecorator`. """ name: str conditions: list["Condition"] raises: list[type[Exception]] def __init__( self, *expr: str, raises: Optional[Iterable[type[Exception]]] = None ) -> None: self.raises = [NotImplementedError, AssertionError, RuntimeError] self.raises.extend(raises or ()) # Assemble a list of Condition instances to be checked self.conditions = [] for spec in chain(*[e.split(";") for e in expr]): try: # Split a string like "pkgA 1.2.3.post0" package, v_min = spec.strip().split(" ") except ValueError: # Failed → something like "message_ix_models.foo.bar.baz" → recurse c: "Condition" = _Recurse(spec) else: c = _PackageVersion(package, v_min) self.conditions.append(c)
[docs] def check_versions(self) -> str: """Evaluate all the :attr:`conditions. Return :py:`""` if all pass, else a :class:`str` describing failed conditions. """ return ", ".join(filter(None, [cond() for cond in self.conditions]))
[docs] def raise_for_version(self) -> None: """Raise :class:`.NotImplementedError` if :meth:`check_versions` fails.""" if result := self.check_versions(): raise NotImplementedError(f"{self.name} with {result}.")
[docs] def mark_test(self, obj): """Apply a pytest XFAIL mark to test function or class `obj`.""" import pytest # Evaluate the conditions msg = self.check_versions() # Create the Mark mark = pytest.mark.xfail( condition=bool(msg), raises=tuple(self.raises), reason=f"Not supported with {msg}", ) # Apply the mark to obj; return the result return mark(obj)
def __call__(self, to_wrap: Callable) -> "MinimumVersionDecorated": """Wrap `to_wrap`.""" # Store name for raise_for_version() self.name = f"{to_wrap.__module__}.{to_wrap.__name__}()" # Create a wrapper around `to_wrap` def wrapper(*args, **kwargs): self.raise_for_version() # MinimumVersionDecorator return to_wrap(*args, **kwargs) # Apply update_wrapper() from the standard library functools.update_wrapper(wrapper, to_wrap) # Set property minimum_version that can be used to mark test functions/classes setattr(wrapper, "minimum_version", self.mark_test) assert hasattr(wrapper, "minimum_version") # Store a reference to the current MinimumVersionDecorator for use by _Recurse setattr(wrapper, "_mvd", self) return wrapper
#: Alias for :class:`.MinimumVersionDecorator`. minimum_version = MinimumVersionDecorator