diff --git a/docs/cli.rst b/docs/cli.rst index 2eae5a2..fa5763a 100644 --- a/docs/cli.rst +++ b/docs/cli.rst @@ -345,7 +345,7 @@ Generating small synthetic datasets to try it on: # JoinByTimestamp is a local plugin transformer (see Plugins below) - needed both to # build the pipeline above and to load it back for damast process - export DAMAST_PLUGIN_PATH=docs/examples/plugins + export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins damast process --pipeline pipelines/osint_ais_preparation.damast.ppl \ --input-data df=docs/examples/data/ais.parquet \ --input-data osint_events=docs/examples/data/osint.parquet \ @@ -365,7 +365,7 @@ resolvable, i.e., ``DAMAST_PLUGIN_PATH`` must be set for a local plugin, while i :: - export DAMAST_PLUGIN_PATH=docs/examples/plugins + export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins damast process --pipeline pipelines/osint_ais_preparation.damast.ppl --describe .. highlight:: none @@ -624,20 +624,35 @@ registering such plugin transformers are supported - see :class:`damast.core.tra for the full API: - installable packages that declare their :class:`damast.core.transformations.PipelineElement` subclasses via the - ``damast.transformers`` entry-point group in their own ``pyproject.toml``:: + ``damast.transformers`` entry-point group in their own ``pyproject.toml`` - either one entry per class, or + one entry per module, which registers every transformer defined in that module (or, for a package, in its + top-level submodules):: [project.entry-points."damast.transformers"] MyTransformer = "acme_pkg.transformers:MyTransformer" + acme_pkg = "acme_pkg.transformers" -- local, ad-hoc ``*.py`` files that are not part of any installed package, made discoverable by - pointing the ``DAMAST_PLUGIN_PATH`` environment variable at the directory (or directories, - separated with ``os.pathsep``) that contains them +- local directories that are not part of any installed package, made discoverable via the + ``DAMAST_PLUGIN_PATH`` environment variable (several entries separated with ``os.pathsep``). An entry + ``name=path`` loads the directory as a package called ``name``: its top-level ``*.py`` files become + ``name.``, and may use relative imports (``from .helpers import x``), also into subdirectories. + The same can be done in code via ``plugin_manager.register_plugin_package(name, path)``. + Choose a distinctive name - one that is already importable is rejected. A bare ``path`` entry + (deprecated) loads each file as a flat module named after the file, without relative imports. + +The package name of a local directory is recorded in saved pipelines (e.g. +``module_name: my_plugins.my_transformers``), so a pipeline can only be replayed with the directory +registered under the same name - or with an installed package of that name, e.g. once the directory has +been turned into an installable package with a module entry-point. Regardless of which of the two a transformer comes from, it is resolvable in code the same way, -via the ``damast.plugins`` namespace:: +via ``damast.plugins.`` - where ```` is the top-level package of the module +defining it, i.e. ``acme_pkg`` for ``acme_pkg.transformers:MyTransformer``, or the name a local +directory was registered under:: - from damast.plugins import MyTransformer + from damast.plugins.acme_pkg import MyTransformer +Scoping by package means two plugins can provide a transformer of the same name without clashing. ``damast.plugins`` resolves names lazily on first access, so nothing beyond the requested class is ever imported - see :mod:`damast.plugins` for details. @@ -650,27 +665,29 @@ subclass in a loose ``*.py`` file, written like any other transformer: .. literalinclude:: ./examples/plugins/my_transformers.py :language: Python -With ``DAMAST_PLUGIN_PATH`` pointing at the directory containing that file, ``damast plugins`` -lists it without requiring any further Python code: +With ``DAMAST_PLUGIN_PATH`` registering the directory containing that file (here as package +``my_plugins``), ``damast plugins`` lists it without requiring any further Python code: .. highlight:: none :: - $ export DAMAST_PLUGIN_PATH=./examples/plugins + $ export DAMAST_PLUGIN_PATH=my_plugins=./examples/plugins $ damast plugins - MyTripler: my_transformers:MyTripler + my_plugins (local: examples/plugins) + JoinByTimestamp .osint_ais_transformers + MyTripler .my_transformers -``MyTripler`` is now resolvable via ``damast.plugins`` and can be used in a pipeline like any -other transformer: +``MyTripler`` is now resolvable via ``damast.plugins.my_plugins`` and can be used in a pipeline +like any other transformer: .. literalinclude:: ./examples/damast-plugin-pipeline.py :language: Python The resulting pipeline can be applied like any other, e.g. via ``damast process`` (see `Process`_ -above), as long as ``DAMAST_PLUGIN_PATH`` is still set to a directory containing -``my_transformers.py``: +above), as long as ``DAMAST_PLUGIN_PATH`` still registers the directory containing +``my_transformers.py`` as ``my_plugins``: .. highlight:: none @@ -679,7 +696,7 @@ above), as long as ``DAMAST_PLUGIN_PATH`` is still set to a directory containing damast process --input-data data.parquet --pipeline pipelines/my-plugin-pipeline.damast.ppl Pipelines saved with a plugin transformer record where it came from under ``requires`` (the -installed distribution and version, or the original local file path), so that loading the pipeline -elsewhere fails with an actionable message - naming the missing package to ``pip install``, or the -``DAMAST_PLUGIN_PATH`` directory to add - instead of a bare import error. +installed distribution and version, or the local package name and directory), so that loading the +pipeline elsewhere fails with an actionable message - naming the missing package to ``pip install``, or +the ``DAMAST_PLUGIN_PATH`` entry to add - instead of a bare import error. diff --git a/docs/examples/damast-osint-ais-pipeline.py b/docs/examples/damast-osint-ais-pipeline.py index b5ea367..610a8e3 100644 --- a/docs/examples/damast-osint-ais-pipeline.py +++ b/docs/examples/damast-osint-ais-pipeline.py @@ -16,10 +16,10 @@ # JoinByTimestamp lives in its own package/file in a real project - see docs/cli.rst > # Plugins. Here it's a local plugin file, resolved the same way. -os.environ["DAMAST_PLUGIN_PATH"] = str(Path(__file__).parent / "plugins") +os.environ["DAMAST_PLUGIN_PATH"] = f"ais_osint={Path(__file__).parent / 'plugins'}" from damast.core.dataprocessing import DataProcessingPipeline -from damast.plugins import JoinByTimestamp +from damast.plugins.ais_osint import JoinByTimestamp pipeline = DataProcessingPipeline(name="osint_ais_preparation", description="Join AIS pings with OSINT events by timestamp", diff --git a/docs/examples/damast-plugin-pipeline.py b/docs/examples/damast-plugin-pipeline.py index 6583599..9a476ab 100644 --- a/docs/examples/damast-plugin-pipeline.py +++ b/docs/examples/damast-plugin-pipeline.py @@ -1,12 +1,13 @@ import os from pathlib import Path -# DAMAST_PLUGIN_PATH points at the directory holding my_transformers.py, so that -# 'MyTripler' becomes resolvable via damast.plugins - see docs/examples/plugins/. -os.environ["DAMAST_PLUGIN_PATH"] = str(Path(__file__).parent / "plugins") +# DAMAST_PLUGIN_PATH registers the directory holding my_transformers.py as package +# 'my_plugins', so that 'MyTripler' becomes resolvable via damast.plugins.my_plugins - see +# docs/examples/plugins/. +os.environ["DAMAST_PLUGIN_PATH"] = f"my_plugins={Path(__file__).parent / 'plugins'}" from damast.core import DataProcessingPipeline -from damast.plugins import MyTripler +from damast.plugins.my_plugins import MyTripler pipeline = DataProcessingPipeline(name="my-plugin-pipeline", base_dir=".") pipeline.add("Triple mmsi", diff --git a/docs/usage.ipynb b/docs/usage.ipynb index e912567..979df96 100644 --- a/docs/usage.ipynb +++ b/docs/usage.ipynb @@ -337,16 +337,20 @@ "Python package at all. Damast supports two ways to use `PipelineElement` classes that live\n", "outside of `damast` itself, so that pipelines using them stay reproducible when shared with\n", "others. Regardless of which of the two a transformer comes from, it is always resolvable the\n", - "same way in code - `from damast.plugins import MyTransformer` - without needing to import (or\n", - "even know) the underlying module or package that defines it.\n", + "same way in code - `from damast.plugins. import MyTransformer`, where `` is\n", + "the plugin's top-level package - without needing to import (or even know) the underlying module\n", + "that defines it. Scoping by package means two plugins can provide a transformer of the same name\n", + "without clashing.\n", "\n", "### Local, ad-hoc transformers via `DAMAST_PLUGIN_PATH`\n", "\n", "For a quick transformer that does not warrant its own package, put it in a plain `*.py` file and\n", - "point the `DAMAST_PLUGIN_PATH` environment variable at the directory containing it (multiple\n", - "directories can be separated with `os.pathsep`, just like `PATH`). Every top-level file found\n", - "there is imported once - using its filename stem as the module name - so any `PipelineElement`\n", - "subclasses it defines become resolvable exactly like classes from an installed package." + "register the directory containing it under a package name via the `DAMAST_PLUGIN_PATH`\n", + "environment variable, as `name=path` (multiple entries can be separated with `os.pathsep`, just\n", + "like `PATH`). Every top-level file found there is imported once as `name.` - so files may\n", + "use relative imports among each other - and any `PipelineElement` subclasses they define become\n", + "resolvable exactly like classes from an installed package. Choose a distinctive name: it is\n", + "recorded in saved pipelines, and a name that is already importable is rejected." ] }, { @@ -354,13 +358,13 @@ "execution_count": null, "metadata": {}, "outputs": [], - "source": "import os\nimport tempfile\nfrom pathlib import Path\n\nplugin_dir = Path(tempfile.mkdtemp())\n(plugin_dir / \"my_transformers.py\").write_text(\"\"\"\nimport polars\n\nfrom damast.core.dataframe import AnnotatedDataFrame\nfrom damast.core.decorators import describe, input, output\nfrom damast.core.transformations import PipelineElement\n\n\nclass Doubler(PipelineElement):\n @describe(\"Doubles a column\")\n @input({\"x\": {}})\n @output({\"{{x}}_doubled\": {}})\n def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame:\n feature = self.get_name(\"x\")\n df.lazyframe = df.lazyframe.with_columns(\n (polars.col(feature) * 2).alias(f\"{feature}_doubled\")\n )\n return df\n\"\"\")\n\nos.environ[\"DAMAST_PLUGIN_PATH\"] = str(plugin_dir)\n\nfrom damast.core.transformations import PipelineElement # noqa: E402\n\n# directories are only scanned once per process, so force a rescan after\n# changing DAMAST_PLUGIN_PATH or adding/editing files\nPipelineElement.reload_plugins()\nPipelineElement.list_plugins()" + "source": "import os\nimport tempfile\nfrom pathlib import Path\n\nplugin_dir = Path(tempfile.mkdtemp())\n(plugin_dir / \"my_transformers.py\").write_text(\"\"\"\nimport polars\n\nfrom damast.core.dataframe import AnnotatedDataFrame\nfrom damast.core.decorators import describe, input, output\nfrom damast.core.transformations import PipelineElement\n\n\nclass Doubler(PipelineElement):\n @describe(\"Doubles a column\")\n @input({\"x\": {}})\n @output({\"{{x}}_doubled\": {}})\n def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame:\n feature = self.get_name(\"x\")\n df.lazyframe = df.lazyframe.with_columns(\n (polars.col(feature) * 2).alias(f\"{feature}_doubled\")\n )\n return df\n\"\"\")\n\nos.environ[\"DAMAST_PLUGIN_PATH\"] = f\"my_plugins={plugin_dir}\"\n\nfrom damast.core.transformations import PipelineElement # noqa: E402\n\n# directories are only scanned once per process, so force a rescan after\n# changing DAMAST_PLUGIN_PATH or adding/editing files\nPipelineElement.reload_plugins()\nPipelineElement.list_plugins()" }, { "cell_type": "markdown", "metadata": {}, "source": [ - "`Doubler` is now resolvable via `damast.plugins`, and can be used in a pipeline like any other\n", + "`Doubler` is now resolvable via `damast.plugins.my_plugins`, and can be used in a pipeline like any other\n", "transformer. `damast.plugins` resolves names lazily on first access - nothing beyond the\n", "requested class is imported, and it works no matter which of the two plugin sources defines\n", "it. Saving the pipeline records where the transformer came from, under `requires`, so a pipeline\n", @@ -375,7 +379,7 @@ "outputs": [], "source": [ "from damast.core.dataprocessing import DataProcessingPipeline\n", - "from damast.plugins import (\n", + "from damast.plugins.my_plugins import (\n", " Doubler, # resolved from the file we just wrote to `plugin_dir`\n", ")\n", "\n", @@ -401,12 +405,16 @@ "MyTransformer = \"acme_pkg.transformers:MyTransformer\"\n", "```\n", "\n", + "Instead of one entry per class, an entry may also name a whole module, e.g.\n", + "`acme_pkg = \"acme_pkg.transformers\"`: every `PipelineElement` defined in that module - or, for a\n", + "package, in its top-level submodules - is then registered.\n", + "\n", "Once installed, it is discovered the same way as a local plugin - `PipelineElement.list_plugins()`\n", "merges both sources - and pipelines saved with it additionally record the distribution name and\n", "version under `requires`, so a version mismatch on reload is logged as a warning rather than\n", "silently changing behavior. Run `damast plugins` from the command line to list everything that is\n", "currently discoverable, from either source, without writing any Python. In code, it resolves the\n", - "same way as a local plugin too - `from damast.plugins import MyTransformer`." + "same way as a local plugin too - `from damast.plugins.acme_pkg import MyTransformer`." ] }, { diff --git a/pyproject.toml b/pyproject.toml index a547cd9..b77779b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,7 +71,8 @@ repository = "https://github.com/simula/damast" # Third-party packages can register PipelineElement subclasses as damast plugins via # their own pyproject.toml, e.g.: # [project.entry-points."damast.transformers"] -# MyTransformer = "acme_pkg.transformers:MyTransformer" +# MyTransformer = "acme_pkg.transformers:MyTransformer" # a single class +# acme_pkg = "acme_pkg.transformers" # all classes of a module/package # Discoverable via `damast plugins` or PipelineElement.list_plugins(). [project.optional-dependencies] diff --git a/src/damast/cli/plugins.py b/src/damast/cli/plugins.py index 23fc83b..e96a223 100644 --- a/src/damast/cli/plugins.py +++ b/src/damast/cli/plugins.py @@ -2,7 +2,7 @@ from argparse import ArgumentParser from damast.cli.base import BaseParser -from damast.core.transformations import PipelineElement, PluginManager +from damast.core.transformations import PipelineElement, PluginManager, plugin_manager class PluginsParser(BaseParser): @@ -21,5 +21,26 @@ def execute(self, args): f"{PluginManager.PLUGIN_PATH_ENV}={plugin_path})") return - for name, target in sorted(plugins.items()): - print(f"{name}: {target}") + # '.' -> 'module:class', grouped per plugin package + packages: dict[str, list[tuple[str, str]]] = {} + for qualified_name, target in plugins.items(): + package, class_name = qualified_name.split(".", 1) + packages.setdefault(package, []).append((class_name, target.split(":")[0])) + + for package, transformers in sorted(packages.items()): + source = self._describe_source(plugin_manager.resolve_requirement(transformers[0][1])) + print(f"{package}{f' ({source})' if source else ''}") + + width = max(len(class_name) for class_name, _ in transformers) + for class_name, module_name in sorted(transformers): + # module relative to the package - omitted if defined in the package module itself + relative_module = module_name.removeprefix(package) + print(f" {class_name:<{width}} {relative_module}".rstrip()) + + @staticmethod + def _describe_source(requirement: dict[str, str] | None) -> str: + if not requirement: + return "" + if requirement.get("distribution"): + return f"{requirement['distribution']}=={requirement['version']}" + return f"local: {requirement['path']}" diff --git a/src/damast/core/polars_dataframe.py b/src/damast/core/polars_dataframe.py index 7946664..248b837 100644 --- a/src/damast/core/polars_dataframe.py +++ b/src/damast/core/polars_dataframe.py @@ -631,6 +631,8 @@ def read_batches(with_columns: list[str] | None, n_rows -= df.height yield df + # According to polars documentation this functionality is considered unstable + # https://docs.pola.rs/api/python/stable/reference/api/polars.io.plugins.register_io_source.html return register_io_source(read_batches, schema=schema), variables @staticmethod diff --git a/src/damast/core/transformations.py b/src/damast/core/transformations.py index f2490a8..f3e2e84 100644 --- a/src/damast/core/transformations.py +++ b/src/damast/core/transformations.py @@ -3,12 +3,16 @@ import copy import importlib import importlib.metadata +import importlib.machinery import importlib.util import inspect +import keyword import os +import pkgutil import re import sys from abc import abstractmethod +from collections.abc import Callable from logging import getLogger from pathlib import Path from types import ModuleType @@ -67,20 +71,29 @@ class PluginManager: Discovers and resolves :class:`PipelineElement` 'plugin' transformers, i.e. transformers that are not necessarily part of the damast package itself. - Two plugin sources are supported: + Supported plugin sources: - - packages that register :class:`PipelineElement` subclasses via the - ``damast.transformers`` entry-point group, e.g. in their own pyproject.toml:: + - installed packages that register :class:`PipelineElement` subclasses via the + ``damast.transformers`` entry-point group in their own pyproject.toml - either one + entry per class, or one entry per module (every PipelineElement defined in that module, + or in the top-level submodules of that package, is registered):: [project.entry-points."damast.transformers"] MyTransformer = "acme_pkg.transformers:MyTransformer" + acme_pkg = "acme_pkg.transformers" - - loose ``*.py`` files in directories listed in the ``DAMAST_PLUGIN_PATH`` - environment variable (os.pathsep-separated), for local/ad-hoc transformers that - are not part of an installed package. Every top-level file found there is - imported once (using its filename stem as 'module_name'), so that any - :class:`PipelineElement` subclasses it defines become resolvable exactly like - classes from an installed package. + - local directories listed in the ``DAMAST_PLUGIN_PATH`` environment variable + (os.pathsep-separated), for transformers that are not part of an installed package: + + - ``name=path`` (or :func:`register_plugin_package`) imports the directory as a package + called ``name``, so its top-level files become ``name.`` and may use relative + imports (``from .helpers import x``), including into subpackages + - a bare ``path`` (deprecated) imports each top-level file as a flat module named after + its filename stem - relative imports are not possible there + + In either case every top-level, non-underscore file is imported once, so that the + :class:`PipelineElement` subclasses it defines become resolvable exactly like classes from + an installed package. """ #: Entry-point group that plugin packages use to advertise PipelineElement subclasses @@ -90,10 +103,18 @@ class PluginManager: PLUGIN_PATH_ENV = "DAMAST_PLUGIN_PATH" def __init__(self): - #: module_name -> loaded module, for modules imported from PLUGIN_PATH_ENV + #: module_name -> loaded module, for modules imported from local plugin directories self._local_modules: dict[str, ModuleType] = {} #: module_name -> source file, used to detect/warn about name collisions self._local_files: dict[str, Path] = {} + #: package name -> directory, for named local plugin directories that were loaded + self._local_packages: dict[str, Path] = {} + #: package name -> directory, registered via register_plugin_package() + self._registered_packages: dict[str, Path] = {} + #: module entry-point value -> modules found in it (see scan_module_entry_point) + self._entry_point_modules: dict[str, dict[str, ModuleType]] = {} + #: unnamed plugin directories a deprecation warning was already logged for + self._warned_unnamed: set[Path] = set() self._loaded = False self._requirement_cache: dict[str, dict[str, str] | None] = {} @@ -105,15 +126,57 @@ def local_modules(self) -> dict[str, ModuleType]: def local_files(self) -> dict[str, Path]: return dict(self._local_files) + @property + def local_packages(self) -> dict[str, Path]: + return dict(self._local_packages) + def plugin_path_dirs(self) -> list[Path]: + return [path for _, path in self.plugin_path_entries()] + + def plugin_path_entries(self) -> list[tuple[str | None, Path]]: + """ + Parse :attr:`PLUGIN_PATH_ENV` into ``(package_name, directory)`` pairs. + + An entry ``name=path`` names the package, a bare ``path`` yields ``None`` as name. An + entry is only treated as named if the part before the first ``=`` contains no path + separator, so plain paths that happen to contain ``=`` keep working. + """ raw = os.environ.get(self.PLUGIN_PATH_ENV, "") - return [Path(p) for p in raw.split(os.pathsep) if p.strip()] + entries = [] + for entry in raw.split(os.pathsep): + if not entry.strip(): + continue + name, sep, path = entry.partition("=") + if sep and not any(s in name for s in {"/", os.sep}): + entries.append((name.strip(), Path(path))) + else: + entries.append((None, Path(entry))) + return entries + + def register_plugin_package(self, name: str, path: str | Path): + """ + Register a local plugin directory as package ``name`` - same as adding ``name=path`` + to :attr:`PLUGIN_PATH_ENV`. It is loaded on the next plugin lookup. + + :param name: Package name, a valid Python identifier + :param path: Directory containing the plugin files + :raise ValueError: If ``name`` is not a valid package name + """ + if not self._is_valid_package_name(name): + raise ValueError(f"PluginManager: '{name}' is not a valid plugin package name") + self._registered_packages[name] = Path(path) + self._loaded = False + + @staticmethod + def _is_valid_package_name(name: str) -> bool: + return name.isidentifier() and not keyword.iskeyword(name) def load_local_plugins(self, force: bool = False) -> dict[str, ModuleType]: """ - Import loose '*.py' files found in :attr:`PLUGIN_PATH_ENV` directories, so that - any PipelineElement subclasses they define become resolvable by - 'module_name'/'class_name' - the same way as classes from an installed package. + Import the plugin files found in :attr:`PLUGIN_PATH_ENV` directories and in packages + registered via :func:`register_plugin_package`, so that any PipelineElement subclasses + they define become resolvable by 'module_name'/'class_name' - the same way as classes + from an installed package. :param force: Re-scan the configured directories and re-import their files, even if they were already loaded in this process @@ -122,44 +185,182 @@ def load_local_plugins(self, force: bool = False) -> dict[str, ModuleType]: return self._local_modules if force: - self._local_modules.clear() - self._local_files.clear() - self._requirement_cache.clear() - - for plugin_dir in self.plugin_path_dirs(): - if not plugin_dir.is_dir(): - logger.warning(f"PluginManager: {self.PLUGIN_PATH_ENV} entry '{plugin_dir}'" - " is not a directory - skipping") + self._unload() + # the import system caches directory listings - make added files visible + importlib.invalidate_caches() + + # Do not write bytecode for local plugin sources: .pyc files are validated by the + # source's mtime (in whole seconds) and size only, so an edit within the same second + # would otherwise be missed by a reload. + dont_write_bytecode = sys.dont_write_bytecode + sys.dont_write_bytecode = True + try: + entries = self.plugin_path_entries() + list(self._registered_packages.items()) + for name, plugin_dir in entries: + if not plugin_dir.is_dir(): + logger.warning(f"PluginManager: {self.PLUGIN_PATH_ENV} entry '{plugin_dir}'" + " is not a directory - skipping") + continue + + if name is None: + self._load_flat_directory(plugin_dir) + else: + self._load_package_directory(name, plugin_dir) + finally: + sys.dont_write_bytecode = dont_write_bytecode + + self._loaded = True + return self._local_modules + + def _load_flat_directory(self, plugin_dir: Path): + if plugin_dir not in self._warned_unnamed: + self._warned_unnamed.add(plugin_dir) + logger.warning( + f"PluginManager: unnamed {self.PLUGIN_PATH_ENV} entry '{plugin_dir}' is deprecated" + f" - use '={plugin_dir}' to load it as a package" + ) + + for py_file in sorted(plugin_dir.glob("*.py")): + module_name = py_file.stem + if module_name.startswith("_"): continue - for py_file in sorted(plugin_dir.glob("*.py")): - module_name = py_file.stem - if module_name.startswith("_"): - continue + existing_file = self._local_files.get(module_name) + if existing_file is not None: + if existing_file != py_file: + logger.warning( + f"PluginManager: plugin module '{module_name}' from '{py_file}' collides with" + f" already loaded '{existing_file}' - keeping the first one" + ) + continue - existing_file = self._local_files.get(module_name) - if existing_file is not None: - if existing_file != py_file: - logger.warning( - f"PluginManager: plugin module '{module_name}' from '{py_file}' collides with" - f" already loaded '{existing_file}' - keeping the first one" - ) - continue + spec = importlib.util.spec_from_file_location(module_name, py_file) + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + except Exception as e: + logger.warning(f"PluginManager: failed to load plugin '{py_file}': {e}") + continue - spec = importlib.util.spec_from_file_location(module_name, py_file) - module = importlib.util.module_from_spec(spec) - try: - spec.loader.exec_module(module) - except Exception as e: - logger.warning(f"PluginManager: failed to load plugin '{py_file}': {e}") - continue + sys.modules[module_name] = module + self._local_modules[module_name] = module + self._local_files[module_name] = py_file - sys.modules[module_name] = module - self._local_modules[module_name] = module - self._local_files[module_name] = py_file + def _load_package_directory(self, name: str, plugin_dir: Path): + existing_dir = self._local_packages.get(name) + if existing_dir is not None: + if existing_dir != plugin_dir: + logger.warning(f"PluginManager: plugin package '{name}' from '{plugin_dir}' collides" + f" with already loaded '{existing_dir}' - keeping the first one") + return - self._loaded = True - return self._local_modules + if not self._is_valid_package_name(name): + logger.warning(f"PluginManager: '{name}' in {self.PLUGIN_PATH_ENV} entry" + f" '{name}={plugin_dir}' is not a valid package name - skipping") + return + + # never shadow an importable module, e.g. from the standard library or an installed + # plugin package - which then takes precedence + if name in sys.modules or importlib.util.find_spec(name) is not None: + logger.warning(f"PluginManager: plugin package name '{name}' (for '{plugin_dir}') is" + " already importable - skipping") + return + + # The directory itself becomes the package: its __init__.py if present, an empty + # package otherwise. Submodules then resolve via the regular import system. + init_file = plugin_dir / "__init__.py" + if init_file.is_file(): + spec = importlib.util.spec_from_file_location( + name, init_file, submodule_search_locations=[str(plugin_dir)]) + else: + spec = importlib.machinery.ModuleSpec(name, None, is_package=True) + spec.submodule_search_locations = [str(plugin_dir)] + + package = importlib.util.module_from_spec(spec) + sys.modules[name] = package + try: + if spec.loader is not None: + spec.loader.exec_module(package) + except Exception as e: + sys.modules.pop(name, None) + logger.warning(f"PluginManager: failed to load plugin package '{name}' from '{init_file}': {e}") + return + + self._local_packages[name] = plugin_dir + for module_name, module in self._scan_package(package).items(): + self._local_modules[module_name] = module + self._local_files[module_name] = Path(module.__file__) if module.__file__ else plugin_dir + + @staticmethod + def _scan_package(module: ModuleType) -> dict[str, ModuleType]: + """ + The given module, plus - if it is a package - its top-level, non-underscore, + non-package submodules (imported here). A failing submodule is skipped with a warning. + """ + modules = {module.__name__: module} + for info in pkgutil.iter_modules(getattr(module, "__path__", [])): + if info.ispkg or info.name.startswith("_"): + continue + module_name = f"{module.__name__}.{info.name}" + try: + modules[module_name] = importlib.import_module(module_name) + except Exception as e: + logger.warning(f"PluginManager: failed to load plugin module '{module_name}': {e}") + return modules + + @staticmethod + def is_module_entry_point(entry_point) -> bool: + """Whether an entry point names a whole module ('pkg.mod') rather than a class ('pkg.mod:Class').""" + return ":" not in entry_point.value + + def scan_module_entry_point(self, entry_point) -> dict[str, ModuleType]: + """ + Import the module named by a module entry point - see :func:`is_module_entry_point` - + and, if it is a package, its top-level submodules. + + :return: module_name -> module, empty if the module could not be imported + """ + value = entry_point.value.strip() + if value not in self._entry_point_modules: + try: + module = importlib.import_module(value) + except Exception as e: + logger.warning(f"PluginManager: failed to load plugin module '{value}' of" + f" entry-point '{entry_point.name}': {e}") + self._entry_point_modules[value] = {} + else: + self._entry_point_modules[value] = self._scan_package(module) + return self._entry_point_modules[value] + + @staticmethod + def pipeline_elements(modules: dict[str, ModuleType]) -> list[tuple[str, str, type]]: + """ + :return: (module_name, class_name, class) for each PipelineElement subclass *defined* + in one of the given modules - re-exported classes are skipped + """ + return [ + (module_name, attr_name, obj) + for module_name, module in modules.items() + for attr_name, obj in vars(module).items() + if (inspect.isclass(obj) + and issubclass(obj, PipelineElement) + and obj is not PipelineElement + and obj.__module__ == module_name) + ] + + def _unload(self): + """Forget all loaded local plugins and module entry points, and drop them from sys.modules.""" + for module_name in self._local_files: + sys.modules.pop(module_name, None) + for name in self._local_packages: + for module_name in [m for m in sys.modules if m == name or m.startswith(f"{name}.")]: + sys.modules.pop(module_name, None) + self._local_modules.clear() + self._local_files.clear() + self._local_packages.clear() + self._entry_point_modules.clear() + self._requirement_cache.clear() + self._loaded = False def reload(self): """ @@ -181,9 +382,11 @@ def resolve_requirement(self, module_name: str) -> dict[str, str] | None: :param module_name: Dotted module path of a :class:`PipelineElement` subclass :return: Dict with 'distribution' and 'version' for an installed package; a dict with - 'hint': 'local' and 'path' for a transformer loaded from :attr:`PLUGIN_PATH_ENV`; - or None if it could not be resolved at all (e.g. the class is defined in a script or - notebook that is neither installed nor on the plugin path) + 'hint': 'local', 'package' and 'path' (the directory) for a transformer from a named + local plugin package; a dict with 'hint': 'local' and 'path' (the file) for one from + an unnamed :attr:`PLUGIN_PATH_ENV` directory; or None if it could not be resolved at + all (e.g. the class is defined in a script or notebook that is neither installed nor + on the plugin path) """ if module_name in self._requirement_cache: return self._requirement_cache[module_name] @@ -191,11 +394,13 @@ def resolve_requirement(self, module_name: str) -> dict[str, str] | None: self.load_local_plugins() result = None + top_level = module_name.split(".")[0] local_file = self._local_files.get(module_name) - if local_file is not None: + if top_level in self._local_packages: + result = {"hint": "local", "package": top_level, "path": str(self._local_packages[top_level])} + elif local_file is not None: result = {"hint": "local", "path": str(local_file)} else: - top_level = module_name.split(".")[0] try: distributions = importlib.metadata.packages_distributions().get(top_level) except Exception: @@ -212,42 +417,99 @@ def resolve_requirement(self, module_name: str) -> dict[str, str] | None: self._requirement_cache[module_name] = result return result + @staticmethod + def plugin_package(module_name: str) -> str: + """ + The plugin package a module belongs to, i.e. its top-level package - e.g. 'acme' for + 'acme.transformers', or the package name of a named local plugin directory. Plugin + transformers are exposed per plugin package, as ``damast.plugins..``. + """ + return module_name.split(".")[0] + + def _entry_points(self) -> tuple[list, list]: + """:return: (class entry-points, module entry-points) of :attr:`ENTRY_POINT_GROUP`""" + entry_points = list(importlib.metadata.entry_points(group=self.ENTRY_POINT_GROUP)) + return ([ep for ep in entry_points if not self.is_module_entry_point(ep)], + [ep for ep in entry_points if self.is_module_entry_point(ep)]) + + def plugin_packages(self) -> set[str]: + """Names of all plugin packages - local ones and those of entry-points (not imported here).""" + packages = {self.plugin_package(module_name) for module_name in self.load_local_plugins()} + packages |= {self.plugin_package(ep.value) for ep in importlib.metadata.entry_points(group=self.ENTRY_POINT_GROUP)} + return packages + + def resolve_plugin(self, package: str, name: str) -> type[PipelineElement]: + """ + Resolve transformer ``name`` within plugin ``package`` - see :func:`plugin_package`. + + Sources are checked in this order, the first one wins (with a warning, if several + provide it): local plugin files, class entry-points, then module entry-points - the + latter are only imported if nothing else provides ``name``. + + :raise AttributeError: If the package provides no transformer ``name`` + """ + matches: dict[str, Callable[[], type]] = { + f"{module_name}:{attr_name}": (lambda obj=obj: obj) + for module_name, attr_name, obj in self.pipeline_elements(self.load_local_plugins()) + if self.plugin_package(module_name) == package and attr_name == name + } + class_entry_points, module_entry_points = self._entry_points() + for ep in class_entry_points: + if self.plugin_package(ep.value) == package and ep.name == name: + matches.setdefault(ep.value, ep.load) + + if not matches: + for ep in module_entry_points: + if self.plugin_package(ep.value) != package: + continue + for module_name, attr_name, obj in self.pipeline_elements(self.scan_module_entry_point(ep)): + if attr_name == name: + matches.setdefault(f"{module_name}:{attr_name}", lambda obj=obj: obj) + + if not matches: + raise AttributeError(f"plugin package '{package}' has no transformer '{name}'") + + targets = list(matches) + if len(targets) > 1: + logger.warning(f"PluginManager: transformer '{package}.{name}' is provided by more than one" + f" source ({', '.join(targets)}) - using '{targets[0]}'") + return matches[targets[0]]() + def list_plugins(self) -> dict[str, str]: """ Discover transformer plugins from both the entry-point group and local plugin path. This is purely a discovery/documentation aid - :func:`PipelineElement.create_new` resolves classes by ``module_name``/``class_name`` regardless of whether they are - registered here. If the same class name is registered by more than one source (two - local plugin files, two entry-points, or a local plugin and an entry-point), a warning - is logged and the first source encountered wins - local plugin files are checked before - entry-points, matching the precedence used when resolving a single name (e.g. via - :mod:`damast.plugins`). + registered here. If the same transformer is provided by more than one source with a + different target, a warning is logged and the first one wins - see + :func:`resolve_plugin` for the order. - :return: Mapping of class name to its 'module_name:class_name' target + :return: Mapping of '.' (see :func:`plugin_package`) to its + 'module_name:class_name' target """ plugins: dict[str, str] = {} - def register(attr_name: str, target: str, source: str) -> None: - if attr_name in plugins: + def register(qualified_name: str, target: str, source: str) -> None: + existing = plugins.setdefault(qualified_name, target) + if existing != target: logger.warning( - f"PluginManager: plugin name '{attr_name}' is registered by more than one" - f" source ('{plugins[attr_name]}' and '{target}' from {source}) - keeping" - " the first one" + f"PluginManager: plugin '{qualified_name}' is registered by more than one" + f" source ('{existing}' and '{target}' from {source}) - keeping the first one" ) - return - plugins[attr_name] = target - for module_name, module in self.load_local_plugins().items(): - for attr_name, obj in vars(module).items(): - if (inspect.isclass(obj) - and issubclass(obj, PipelineElement) - and obj is not PipelineElement - and obj.__module__ == module_name): - register(attr_name, f"{module_name}:{attr_name}", "a local plugin file") + for module_name, attr_name, _ in self.pipeline_elements(self.load_local_plugins()): + register(f"{self.plugin_package(module_name)}.{attr_name}", f"{module_name}:{attr_name}", + "a local plugin file") - for ep in importlib.metadata.entry_points(group=self.ENTRY_POINT_GROUP): - register(ep.name, ep.value, "an entry-point") + class_entry_points, module_entry_points = self._entry_points() + for ep in class_entry_points: + register(f"{self.plugin_package(ep.value)}.{ep.name}", ep.value, "an entry-point") + + for ep in module_entry_points: + for module_name, attr_name, _ in self.pipeline_elements(self.scan_module_entry_point(ep)): + register(f"{self.plugin_package(module_name)}.{attr_name}", f"{module_name}:{attr_name}", + f"module entry-point '{ep.name}'") return plugins @@ -439,6 +701,14 @@ def _missing_plugin_message(cls, f"Install it with: pip install {pip_spec}") plugin_path = os.environ.get(PluginManager.PLUGIN_PATH_ENV, "") + if requires and requires.get("hint") == "local" and requires.get("package"): + package = requires["package"] + return (f"{cls.__name__}.create_new: could not load '{class_name}' from '{module_name}'." + f" It was saved as part of the local plugin package '{package}', originally loaded" + f" from '{requires.get('path')}'. Register that directory under the same name, e.g." + f" {PluginManager.PLUGIN_PATH_ENV}=\"{package}={requires.get('path')}\"" + f" (currently: {plugin_path}).") + origin_hint = "" if requires and requires.get("hint") == "local": origin_hint = f" It was saved as a local transformer, originally loaded from '{requires.get('path')}'." @@ -540,7 +810,7 @@ def list_plugins(cls) -> dict[str, str]: supported sources (installed packages via entry-points, and local files via ``DAMAST_PLUGIN_PATH``). - :return: Mapping of class name to its 'module_name:class_name' target + :return: Mapping of '.' to its 'module_name:class_name' target """ return plugin_manager.list_plugins() diff --git a/src/damast/data_handling/transformers/augmenters.py b/src/damast/data_handling/transformers/augmenters.py index 3fda0d7..ca885e1 100644 --- a/src/damast/data_handling/transformers/augmenters.py +++ b/src/damast/data_handling/transformers/augmenters.py @@ -172,6 +172,9 @@ def update_balltree(self, x: npt.NDArray[np.float64]): def __call__(self, x: npt.NDArray[np.float64], y: npt.NDArray[np.float64]) -> npt.NDArray[np.float64]: """ Compute distances between the Balltree and each entry in `x`""" + # polars may pass empty batches, which BallTree.query rejects + if len(x) == 0: + return np.empty(0, dtype=np.float64) return self._tree.query(np.vstack([x, y]).T, return_distance=True)[ 0].reshape(-1) diff --git a/src/damast/plugins/__init__.py b/src/damast/plugins/__init__.py index 2712b62..65528a3 100644 --- a/src/damast/plugins/__init__.py +++ b/src/damast/plugins/__init__.py @@ -1,60 +1,78 @@ """ -Resolve :class:`damast.core.transformations.PipelineElement` 'plugin' transformers by name, so -that e.g. ``from damast.plugins import MyTransformer`` works for any transformer discoverable -by :class:`damast.core.transformations.PluginManager` - whether it comes from an installed -package's ``damast.transformers`` entry point, or a loose file on ``DAMAST_PLUGIN_PATH``. - -Names are resolved lazily on first access (via module ``__getattr__``, see :pep:`562`): nothing -is imported/loaded until a specific name is actually requested, so installed plugin packages -that are never referenced here are never imported, and a ``DAMAST_PLUGIN_PATH`` set after this -module was first imported is still picked up. - -If a name is registered by more than one source (two local plugin files, two entry-points, or a -local plugin and an entry-point sharing a name), a warning is logged and the first source found -wins - local plugin files are checked before entry-points, matching the precedence used by -:func:`damast.core.transformations.PluginManager.list_plugins`. +Resolve :class:`damast.core.transformations.PipelineElement` 'plugin' transformers per plugin +package, so that e.g. ``from damast.plugins.acme import MyTransformer`` works for any transformer +discoverable by :class:`damast.core.transformations.PluginManager` - whether it comes from an +installed package's ``damast.transformers`` entry point (for a single class, or for a whole +module), or a local plugin directory on ``DAMAST_PLUGIN_PATH``. + +The plugin package is the top-level package of the module defining the transformer (see +:func:`damast.core.transformations.PluginManager.plugin_package`): ``acme`` for an installed +``acme.transformers:MyTransformer``, or ``name`` for a local directory registered as +``name=path``. Scoping names by package means two plugins can provide a transformer of the same +name without clashing. + +Names are resolved lazily on first access: ``damast.plugins.`` is created on import +without importing anything, and a transformer is only looked up once it is actually requested - a +module entry-point is only imported if nothing else in that package provides the name. """ from __future__ import annotations -import importlib.metadata -import inspect -from logging import getLogger +import importlib +import importlib.abc +import importlib.util +import sys +from types import ModuleType -from damast.core.transformations import PipelineElement, PluginManager, plugin_manager +from damast.core.transformations import PipelineElement, plugin_manager __all__: list[str] = [] -logger = getLogger(__name__) +class _PluginPackageFinder(importlib.abc.MetaPathFinder, importlib.abc.Loader): + """Creates the ``damast.plugins.`` namespace modules on import.""" + + def find_spec(self, fullname: str, path=None, target=None): + package = fullname.removeprefix(f"{__name__}.") + if package == fullname or "." in package: + return None + if package not in plugin_manager.plugin_packages(): + raise ModuleNotFoundError(f"No plugin package '{package}' - available:" + f" {sorted(plugin_manager.plugin_packages())}", name=fullname) + return importlib.util.spec_from_loader(fullname, self) + + def create_module(self, spec): + return None + + def exec_module(self, module: ModuleType): + package = module.__name__.rpartition(".")[2] + + def __getattr__(name: str) -> type[PipelineElement]: + if name.startswith("__"): + raise AttributeError(name) + return plugin_manager.resolve_plugin(package, name) + + def __dir__() -> list[str]: + prefix = f"{package}." + return sorted(name.removeprefix(prefix) for name in plugin_manager.list_plugins() + if name.startswith(prefix)) -def __getattr__(name: str) -> type[PipelineElement]: - local_matches = [ - (module_name, obj) - for module_name, module in plugin_manager.load_local_plugins().items() - for obj in [vars(module).get(name)] - if (inspect.isclass(obj) and issubclass(obj, PipelineElement) - and obj is not PipelineElement and obj.__module__ == module_name) - ] - entry_point_matches = [ - ep for ep in importlib.metadata.entry_points(group=PluginManager.ENTRY_POINT_GROUP) - if ep.name == name - ] + module.__getattr__ = __getattr__ + module.__dir__ = __dir__ - if len(local_matches) + len(entry_point_matches) > 1: - sources = [module_name for module_name, _ in local_matches] + [ep.value for ep in entry_point_matches] - logger.warning( - f"damast.plugins: plugin name '{name}' is ambiguous - registered by more than one" - f" source ({', '.join(sources)}) - using '{sources[0]}'" - ) - if local_matches: - return local_matches[0][1] +# appended, so a real submodule of damast.plugins would still take precedence +if not any(isinstance(finder, _PluginPackageFinder) for finder in sys.meta_path): + sys.meta_path.append(_PluginPackageFinder()) - if entry_point_matches: - return entry_point_matches[0].load() - raise AttributeError(f"module 'damast.plugins' has no plugin named '{name}'") +def __getattr__(name: str) -> ModuleType: + if name.startswith("__"): + raise AttributeError(name) + try: + return importlib.import_module(f"{__name__}.{name}") + except ModuleNotFoundError as e: + raise AttributeError(f"module '{__name__}' has no plugin package '{name}'") from e def __dir__() -> list[str]: - return sorted(plugin_manager.list_plugins()) + return sorted(plugin_manager.plugin_packages()) diff --git a/tests/damast/cli/test_cli.py b/tests/damast/cli/test_cli.py index 5ee3d21..076f697 100644 --- a/tests/damast/cli/test_cli.py +++ b/tests/damast/cli/test_cli.py @@ -479,6 +479,28 @@ def fake_entry_points(*, group): parser.execute(args=None) captured = capsys.readouterr() - assert "AcmeTransformer: acme_pkg.transformers:AcmeTransformer" in captured.out + assert captured.out == "acme_pkg\n AcmeTransformer .transformers\n" + + +def test_plugins_lists_transformers_per_package(isolate_plugins, tmp_path, monkeypatch, capsys): + source = """ +from damast.core.transformations import PipelineElement + + +class {}(PipelineElement): + pass +""" + (tmp_path / "doublers.py").write_text(source.format("Doubler")) + (tmp_path / "long_named_triplers.py").write_text(source.format("LongNamedTripler")) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, f"acme_cli={tmp_path}") + + from damast.cli.plugins import PluginsParser + PluginsParser(parser=ArgumentParser()).execute(args=None) + + assert capsys.readouterr().out == ( + f"acme_cli (local: {tmp_path})\n" + " Doubler .doublers\n" + " LongNamedTripler .long_named_triplers\n" + ) diff --git a/tests/damast/cli/test_process_cli.py b/tests/damast/cli/test_process_cli.py index 13faf8e..a1e9e4a 100644 --- a/tests/damast/cli/test_process_cli.py +++ b/tests/damast/cli/test_process_cli.py @@ -1,6 +1,5 @@ import re import shutil -import sys import pytest @@ -226,7 +225,7 @@ def join_pipeline_path(tmp_path, monkeypatch): monkeypatch.setenv("DAMAST_PLUGIN_PATH", str(plugin_dir)) plugin_manager.reload() - from damast.plugins import JoinByTimestamp + from damast.plugins.join_transformer import JoinByTimestamp output_dir = tmp_path / "output" output_dir.mkdir() @@ -243,12 +242,8 @@ def join_pipeline_path(tmp_path, monkeypatch): # drop the cached plugin module so it doesn't leak into other tests sharing the # process-wide plugin_manager - mirrors _reset_plugin_manager() in test_plugins.py - for module_name in list(plugin_manager.local_files): - sys.modules.pop(module_name, None) - plugin_manager._local_modules.clear() - plugin_manager._local_files.clear() - plugin_manager._requirement_cache.clear() - plugin_manager._loaded = False + plugin_manager._unload() + plugin_manager._registered_packages.clear() def test_process_multi_datasource_input_data(data_path, join_pipeline_path, tmp_path, script_runner): diff --git a/tests/damast/conftest.py b/tests/damast/conftest.py index a0546b5..61b74c4 100644 --- a/tests/damast/conftest.py +++ b/tests/damast/conftest.py @@ -1,4 +1,3 @@ -import sys from pathlib import Path import pytest @@ -35,9 +34,5 @@ def fake_entry_points(*, group=None, **kwargs): monkeypatch.setattr(importlib_metadata, "entry_points", fake_entry_points) - for module_name in list(plugin_manager.local_files): - sys.modules.pop(module_name, None) - plugin_manager._local_modules.clear() - plugin_manager._local_files.clear() - plugin_manager._requirement_cache.clear() - plugin_manager._loaded = False + plugin_manager._unload() + plugin_manager._registered_packages.clear() diff --git a/tests/damast/core/test_transformations.py b/tests/damast/core/test_transformations.py index d70c877..dd57c6b 100644 --- a/tests/damast/core/test_transformations.py +++ b/tests/damast/core/test_transformations.py @@ -34,12 +34,8 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: def _reset_plugin_manager(): - for module_name in list(plugin_manager.local_files): - sys.modules.pop(module_name, None) - plugin_manager._local_modules.clear() - plugin_manager._local_files.clear() - plugin_manager._requirement_cache.clear() - plugin_manager._loaded = False + plugin_manager._unload() + plugin_manager._registered_packages.clear() @pytest.fixture @@ -198,12 +194,11 @@ def fake_entry_points(*, group): monkeypatch.setattr(importlib.metadata, "entry_points", fake_entry_points) - assert "AcmeTransformer" in PipelineElement.list_plugins() - assert PipelineElement.list_plugins()["AcmeTransformer"] == "acme_pkg.transformers:AcmeTransformer" + assert PipelineElement.list_plugins()["acme_pkg.AcmeTransformer"] == "acme_pkg.transformers:AcmeTransformer" def test_list_plugins_empty_by_default(): - assert "AcmeTransformer" not in PipelineElement.list_plugins() + assert "acme_pkg.AcmeTransformer" not in PipelineElement.list_plugins() def test_local_plugin_path_discovered_via_list_plugins(local_plugin_path): @@ -211,7 +206,7 @@ def test_local_plugin_path_discovered_via_list_plugins(local_plugin_path): PipelineElement.reload_plugins() plugins = PipelineElement.list_plugins() - assert plugins["LocalDoubler"] == "acme_local_transformer:LocalDoubler" + assert plugins["acme_local_transformer.LocalDoubler"] == "acme_local_transformer:LocalDoubler" def test_local_plugin_path_resolvable_via_create_new(local_plugin_path): @@ -277,7 +272,7 @@ def test_local_plugin_path_name_collision_warns_and_keeps_first(tmp_path, monkey _reset_plugin_manager() -def test_list_plugins_class_name_collision_across_local_files_warns_and_keeps_first( +def test_list_plugins_same_class_name_in_different_local_files_is_listed_per_package( local_plugin_path, caplog): (local_plugin_path / "acme_transformer_a.py").write_text(LOCAL_TRANSFORMER_SOURCE) (local_plugin_path / "acme_transformer_b.py").write_text(LOCAL_TRANSFORMER_SOURCE) @@ -286,8 +281,9 @@ def test_list_plugins_class_name_collision_across_local_files_warns_and_keeps_fi with caplog.at_level("WARNING"): plugins = PipelineElement.list_plugins() - assert plugins["LocalDoubler"] == "acme_transformer_a:LocalDoubler" - assert any("is registered by more than one source" in record.message for record in caplog.records) + assert plugins["acme_transformer_a.LocalDoubler"] == "acme_transformer_a:LocalDoubler" + assert plugins["acme_transformer_b.LocalDoubler"] == "acme_transformer_b:LocalDoubler" + assert not any("is registered by more than one source" in record.message for record in caplog.records) def test_list_plugins_local_and_entry_point_collision_warns_and_local_wins( @@ -297,14 +293,14 @@ def test_list_plugins_local_and_entry_point_collision_warns_and_local_wins( class FakeEntryPoint: name = "LocalDoubler" - value = "acme_pkg.transformers:LocalDoubler" + value = "acme_local_transformer.other:LocalDoubler" monkeypatch.setattr(importlib.metadata, "entry_points", lambda *, group: [FakeEntryPoint()]) with caplog.at_level("WARNING"): plugins = PipelineElement.list_plugins() - assert plugins["LocalDoubler"] == "acme_local_transformer:LocalDoubler" + assert plugins["acme_local_transformer.LocalDoubler"] == "acme_local_transformer:LocalDoubler" assert any("is registered by more than one source" in record.message for record in caplog.records) diff --git a/tests/damast/domains/maritime/ais/test_augmenters.py b/tests/damast/domains/maritime/ais/test_augmenters.py index 1faa481..0ae50ea 100644 --- a/tests/damast/domains/maritime/ais/test_augmenters.py +++ b/tests/damast/domains/maritime/ais/test_augmenters.py @@ -292,6 +292,28 @@ def test_add_distance_closest_anchorage(tmp_path): assert np.isclose(np.min(distances), closest_anchorages[idx]) +def test_add_distance_closest_anchorage_empty_input(tmp_path): + dataset = polars.DataFrame({"latitude": [34.84, 35.14], "longitude": [128.42, 128.60]}) + df = polars.DataFrame({ColumnName.LATITUDE: [], ColumnName.LONGITUDE: []}, + schema={ColumnName.LATITUDE: pl.Float64, ColumnName.LONGITUDE: pl.Float64}) + metadata = damast.core.MetaData( + columns=[damast.core.DataSpecification(ColumnName.LATITUDE, unit=units.deg, + representation_type=float), + damast.core.DataSpecification(ColumnName.LONGITUDE, unit=units.deg, + representation_type=float)]) + adf = damast.core.AnnotatedDataFrame(df, metadata) + + pipeline = damast.core.DataProcessingPipeline(name="Compute closest anchorage", + base_dir=tmp_path) + pipeline.add("Add distance to anchorage", ComputeClosestAnchorage(dataset, ["latitude", "longitude"]), + name_mappings={"x": ColumnName.LATITUDE, + "y": ColumnName.LONGITUDE, + "distance": ColumnName.DISTANCE_CLOSEST_ANCHORAGE}) + new_adf = pipeline.transform(adf) + + assert new_adf.collect()[ColumnName.DISTANCE_CLOSEST_ANCHORAGE].len() == 0 + + def test_message_index(tmp_path): mmsi_a = 400000000 mmsi_b = 500000000 diff --git a/tests/damast/test_plugins.py b/tests/damast/test_plugins.py index 22cb9db..fb523b8 100644 --- a/tests/damast/test_plugins.py +++ b/tests/damast/test_plugins.py @@ -1,11 +1,14 @@ +import importlib import importlib.metadata import os +import re import sys +from pathlib import Path import pytest import damast.plugins -from damast.core.transformations import PluginManager, plugin_manager +from damast.core.transformations import PipelineElement, PluginManager, plugin_manager LOCAL_TRANSFORMER_SOURCE = """ from damast.core.transformations import PipelineElement @@ -23,12 +26,8 @@ def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame: def _reset_plugin_manager(): - for module_name in list(plugin_manager.local_files): - sys.modules.pop(module_name, None) - plugin_manager._local_modules.clear() - plugin_manager._local_files.clear() - plugin_manager._requirement_cache.clear() - plugin_manager._loaded = False + plugin_manager._unload() + plugin_manager._registered_packages.clear() @pytest.fixture @@ -42,19 +41,20 @@ def local_plugin_path(tmp_path, monkeypatch): _reset_plugin_manager() -def test_getattr_resolves_local_plugin(local_plugin_path): +def test_getattr_resolves_local_plugin_via_its_package(local_plugin_path): (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) plugin_manager.reload() - LocalDoubler = damast.plugins.LocalDoubler + LocalDoubler = damast.plugins.acme_local_transformer.LocalDoubler assert LocalDoubler.__name__ == "LocalDoubler" assert LocalDoubler.__module__ == "acme_local_transformer" -def test_getattr_resolves_entry_point_plugin(monkeypatch): +def test_getattr_resolves_entry_point_plugin_via_its_package(monkeypatch): class FakeEntryPoint: name = "AcmeTransformer" + value = "acme_pkg.transformers:AcmeTransformer" @staticmethod def load(): @@ -66,33 +66,35 @@ def fake_entry_points(*, group): monkeypatch.setattr(importlib.metadata, "entry_points", fake_entry_points) - assert damast.plugins.AcmeTransformer == "loaded-acme-transformer" + assert damast.plugins.acme_pkg.AcmeTransformer == "loaded-acme-transformer" -def test_getattr_unknown_name_raises_attribute_error(): - with pytest.raises(AttributeError, match="no plugin named 'DoesNotExist'"): - damast.plugins.DoesNotExist - - -def test_local_plugin_takes_precedence_over_same_named_entry_point(local_plugin_path, monkeypatch): +def test_plain_class_name_is_not_resolvable(local_plugin_path): (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) plugin_manager.reload() - class FakeEntryPoint: - name = "LocalDoubler" - value = "acme_pkg.transformers:LocalDoubler" + with pytest.raises(AttributeError, match="no plugin package 'LocalDoubler'"): + damast.plugins.LocalDoubler + with pytest.raises(ImportError): + from damast.plugins import LocalDoubler # noqa: F401 - @staticmethod - def load(): - return "should-not-be-used" - monkeypatch.setattr(importlib.metadata, "entry_points", - lambda *, group: [FakeEntryPoint()]) +def test_unknown_plugin_package_raises(): + with pytest.raises(AttributeError, match="no plugin package 'does_not_exist'"): + damast.plugins.does_not_exist + with pytest.raises(ModuleNotFoundError, match="No plugin package 'does_not_exist'"): + importlib.import_module("damast.plugins.does_not_exist") + - assert damast.plugins.LocalDoubler.__module__ == "acme_local_transformer" +def test_unknown_transformer_in_known_package_raises(local_plugin_path): + (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) + plugin_manager.reload() + with pytest.raises(AttributeError, match="'acme_local_transformer' has no transformer 'DoesNotExist'"): + damast.plugins.acme_local_transformer.DoesNotExist -def test_getattr_warns_on_ambiguous_local_plugins(tmp_path, monkeypatch, caplog): + +def test_same_class_name_in_two_packages_does_not_clash(tmp_path, monkeypatch, caplog): dir_a = tmp_path / "a" dir_b = tmp_path / "b" dir_a.mkdir() @@ -103,47 +105,306 @@ def test_getattr_warns_on_ambiguous_local_plugins(tmp_path, monkeypatch, caplog) plugin_manager.reload() with caplog.at_level("WARNING"): - LocalDoubler = damast.plugins.LocalDoubler + from damast.plugins.acme_transformer_a import LocalDoubler as DoublerA + from damast.plugins.acme_transformer_b import LocalDoubler as DoublerB - assert LocalDoubler.__module__ == "acme_transformer_a" - assert any("is ambiguous" in record.message for record in caplog.records) + assert DoublerA.__module__ == "acme_transformer_a" + assert DoublerB.__module__ == "acme_transformer_b" + assert not any("more than one source" in record.message for record in caplog.records) _reset_plugin_manager() -def test_getattr_warns_on_local_and_entry_point_collision(local_plugin_path, monkeypatch, caplog): +def test_dir_lists_plugin_packages_and_their_transformers(local_plugin_path): (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) plugin_manager.reload() - class FakeEntryPoint: - name = "LocalDoubler" - value = "acme_pkg.transformers:LocalDoubler" + assert "acme_local_transformer" in dir(damast.plugins) + assert dir(damast.plugins.acme_local_transformer) == ["LocalDoubler"] - @staticmethod - def load(): - return "should-not-be-used" - monkeypatch.setattr(importlib.metadata, "entry_points", - lambda *, group: [FakeEntryPoint()]) +def test_from_import_syntax_resolves_local_plugin(local_plugin_path): + (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) + plugin_manager.reload() - with caplog.at_level("WARNING"): - LocalDoubler = damast.plugins.LocalDoubler + from damast.plugins.acme_local_transformer import LocalDoubler assert LocalDoubler.__module__ == "acme_local_transformer" - assert any("is ambiguous" in record.message for record in caplog.records) -def test_dir_lists_discovered_plugins(local_plugin_path): - (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) +# --- named local plugin packages (DAMAST_PLUGIN_PATH="name=path") --------------------------- + +NAMED_MAIN_SOURCE = LOCAL_TRANSFORMER_SOURCE + """ +from . import helpers +from .sub import VALUE +""" + +NAMED_PACKAGE_FILES = { + "main.py": NAMED_MAIN_SOURCE, + "helpers.py": "FACTOR = 2\n", + "_private.py": "raise RuntimeError('must not be imported by the scan')\n", + "sub/__init__.py": "from .util import VALUE\n", + "sub/util.py": "VALUE = 42\n", +} + + +def _write_files(root, files: dict[str, str]): + for relative_path, source in files.items(): + path = root / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source) + return root + + +@pytest.fixture +def named_plugin_dir(tmp_path, monkeypatch): + plugin_dir = _write_files(tmp_path / "transformers", NAMED_PACKAGE_FILES) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, f"acme_named={plugin_dir}") plugin_manager.reload() - assert "LocalDoubler" in dir(damast.plugins) + yield plugin_dir + _reset_plugin_manager() -def test_from_import_syntax_resolves_local_plugin(local_plugin_path): - (local_plugin_path / "acme_local_transformer.py").write_text(LOCAL_TRANSFORMER_SOURCE) + +def test_named_plugin_dir_is_loaded_as_package_with_relative_imports(named_plugin_dir): + assert PipelineElement.list_plugins()["acme_named.LocalDoubler"] == "acme_named.main:LocalDoubler" + + main = sys.modules["acme_named.main"] + assert main.VALUE == 42 + # the sibling imported via 'from . import helpers' is the very module the scan registered + assert main.helpers is plugin_manager.local_modules["acme_named.helpers"] + assert "acme_named._private" not in sys.modules + assert plugin_manager.local_packages == {"acme_named": named_plugin_dir} + + +def test_named_plugin_dir_resolvable_via_damast_plugins_and_create_new(named_plugin_dir): + from damast.plugins.acme_named import LocalDoubler + + assert LocalDoubler.__module__ == "acme_named.main" + + instance = PipelineElement.create_new(module_name="acme_named.main", class_name="LocalDoubler") + assert isinstance(instance, LocalDoubler) + assert dict(instance)["requires"] == {"hint": "local", "package": "acme_named", "path": str(named_plugin_dir)} + + +def test_named_plugin_dir_with_init_runs_it_as_the_package(tmp_path, monkeypatch): + plugin_dir = _write_files(tmp_path / "transformers", {"__init__.py": LOCAL_TRANSFORMER_SOURCE}) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, f"acme_named_init={plugin_dir}") plugin_manager.reload() - from damast.plugins import LocalDoubler + assert PipelineElement.list_plugins()["acme_named_init.LocalDoubler"] == "acme_named_init:LocalDoubler" + assert sys.modules["acme_named_init"].__path__ == [str(plugin_dir)] - assert LocalDoubler.__module__ == "acme_local_transformer" + _reset_plugin_manager() + + +def test_missing_named_plugin_package_error_suggests_name_and_path(named_plugin_dir, monkeypatch): + saved_step = dict(PipelineElement.create_new(module_name="acme_named.main", class_name="LocalDoubler")) + + # simulate loading the pipeline elsewhere, where this package is not registered + monkeypatch.delenv(PluginManager.PLUGIN_PATH_ENV) + plugin_manager.reload() + + with pytest.raises(ImportError, match=re.escape(f'DAMAST_PLUGIN_PATH="acme_named={named_plugin_dir}"')): + PipelineElement.create_new(**saved_step) + + +def test_register_plugin_package(tmp_path): + plugin_dir = _write_files(tmp_path / "transformers", NAMED_PACKAGE_FILES) + plugin_manager.register_plugin_package("acme_registered", plugin_dir) + + assert PipelineElement.list_plugins()["acme_registered.LocalDoubler"] == "acme_registered.main:LocalDoubler" + + # registrations survive a reload + plugin_manager.reload() + assert "acme_registered.main" in plugin_manager.local_modules + + _reset_plugin_manager() + + +def test_register_plugin_package_rejects_invalid_name(tmp_path): + with pytest.raises(ValueError, match="not a valid plugin package name"): + plugin_manager.register_plugin_package("acme-registered", tmp_path) + + +@pytest.mark.parametrize(["name", "message"], [ + ["class", "not a valid package name"], + ["json", "already importable"], +]) +def test_named_plugin_dir_invalid_or_importable_name_is_skipped(name, message, tmp_path, monkeypatch, caplog): + plugin_dir = _write_files(tmp_path / "transformers", NAMED_PACKAGE_FILES) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, f"{name}={plugin_dir}") + + with caplog.at_level("WARNING"): + plugin_manager.reload() + + assert plugin_manager.local_packages == {} + assert any(message in record.message for record in caplog.records) + + _reset_plugin_manager() + + +def test_named_plugin_dir_duplicate_name_warns_and_keeps_first(tmp_path, monkeypatch, caplog): + dir_a = _write_files(tmp_path / "a", NAMED_PACKAGE_FILES) + dir_b = _write_files(tmp_path / "b", NAMED_PACKAGE_FILES) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, + os.pathsep.join([f"acme_dup={dir_a}", f"acme_dup={dir_b}"])) + + with caplog.at_level("WARNING"): + plugin_manager.reload() + + assert plugin_manager.local_packages == {"acme_dup": dir_a} + assert any("collides with already loaded" in record.message for record in caplog.records) + + _reset_plugin_manager() + + +def test_named_plugin_dir_reload_picks_up_edits(named_plugin_dir): + main_file = named_plugin_dir / "main.py" + main_file.write_text(NAMED_MAIN_SOURCE.replace("LocalDoubler", "LocalTripler")) + plugin_manager.reload() + + plugins = PipelineElement.list_plugins() + assert plugins["acme_named.LocalTripler"] == "acme_named.main:LocalTripler" + assert "acme_named.LocalDoubler" not in plugins + + +def test_unnamed_plugin_dir_is_deprecated_but_still_loads_alongside_named(tmp_path, monkeypatch, caplog): + named_dir = _write_files(tmp_path / "named", NAMED_PACKAGE_FILES) + flat_dir = _write_files(tmp_path / "flat", {"acme_flat.py": LOCAL_TRANSFORMER_SOURCE.replace( + "LocalDoubler", "FlatDoubler")}) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, os.pathsep.join([f"acme_mixed={named_dir}", str(flat_dir)])) + plugin_manager._warned_unnamed.clear() + + with caplog.at_level("WARNING"): + plugins = PipelineElement.list_plugins() + + assert plugins["acme_mixed.LocalDoubler"] == "acme_mixed.main:LocalDoubler" + assert plugins["acme_flat.FlatDoubler"] == "acme_flat:FlatDoubler" + assert any("is deprecated" in record.message for record in caplog.records) + + _reset_plugin_manager() + + +# --- module entry-points ('name = "pkg.module"') ------------------------------------------- + +MODULE_ENTRY_FILES = { + "__init__.py": "", + "a.py": LOCAL_TRANSFORMER_SOURCE.replace("LocalDoubler", "ModuleDoubler"), + "reexport.py": "from .a import ModuleDoubler\n", + "broken.py": "raise RuntimeError('broken plugin module')\n", +} + + +@pytest.fixture +def installed_package(tmp_path, monkeypatch): + """Create an importable package, standing in for an installed distribution.""" + site_dir = tmp_path / "site" + site_dir.mkdir() + monkeypatch.syspath_prepend(str(site_dir)) + created = [] + + def make(name: str, files: dict[str, str]): + created.append(name) + _write_files(site_dir / name, files) + importlib.invalidate_caches() + + yield make + + for module_name in [m for m in sys.modules if m.split(".")[0] in created]: + sys.modules.pop(module_name) + _reset_plugin_manager() + + +@pytest.fixture +def fake_entry_points(monkeypatch): + entry_points: list[importlib.metadata.EntryPoint] = [] + + def add(name: str, value: str): + entry_points.append(importlib.metadata.EntryPoint(name=name, value=value, + group=PluginManager.ENTRY_POINT_GROUP)) + + monkeypatch.setattr(importlib.metadata, "entry_points", lambda *, group: list(entry_points)) + return add + + +def test_module_entry_point_for_package_registers_classes_of_its_submodules( + installed_package, fake_entry_points, caplog): + installed_package("acme_mod_pkg", MODULE_ENTRY_FILES) + fake_entry_points("acme_mod_pkg", "acme_mod_pkg") + + with caplog.at_level("WARNING"): + plugins = PipelineElement.list_plugins() + + # defined in 'a', only re-exported by 'reexport' - listed once, without an ambiguity warning + assert plugins["acme_mod_pkg.ModuleDoubler"] == "acme_mod_pkg.a:ModuleDoubler" + assert not any("more than one source" in record.message for record in caplog.records) + assert any("acme_mod_pkg.broken" in record.message for record in caplog.records) + + +def test_module_entry_point_for_single_module_skips_reexported_classes(installed_package, fake_entry_points): + installed_package("acme_mod_single", MODULE_ENTRY_FILES) + fake_entry_points("acme_mod_single", "acme_mod_single.reexport") + + assert "acme_mod_single.ModuleDoubler" not in PipelineElement.list_plugins() + + +def test_module_entry_point_resolvable_via_damast_plugins(installed_package, fake_entry_points): + installed_package("acme_mod_lookup", MODULE_ENTRY_FILES) + fake_entry_points("acme_mod_lookup", "acme_mod_lookup.a") + + from damast.plugins.acme_mod_lookup import ModuleDoubler + + assert ModuleDoubler.__module__ == "acme_mod_lookup.a" + + +def test_class_entry_point_wins_over_module_entry_point_of_same_package(installed_package, fake_entry_points): + installed_package("acme_mod_class", MODULE_ENTRY_FILES) + installed_package("acme_mod_lazy", MODULE_ENTRY_FILES) + fake_entry_points("ModuleDoubler", "acme_mod_class.a:ModuleDoubler") + fake_entry_points("acme_mod_class", "acme_mod_class") + fake_entry_points("acme_mod_lazy", "acme_mod_lazy") + + # answered by the class entry-point: no module entry-point package gets scanned/imported + assert damast.plugins.acme_mod_class.ModuleDoubler.__module__ == "acme_mod_class.a" + assert "acme_mod_class.reexport" not in sys.modules + assert "acme_mod_lazy" not in sys.modules + + # both entries point at the very same class - listed once + assert PipelineElement.list_plugins()["acme_mod_class.ModuleDoubler"] == "acme_mod_class.a:ModuleDoubler" + + +def test_same_class_name_twice_within_one_package_warns_and_keeps_first(tmp_path, monkeypatch, caplog): + plugin_dir = _write_files(tmp_path / "transformers", { + "main.py": LOCAL_TRANSFORMER_SOURCE, + "main2.py": LOCAL_TRANSFORMER_SOURCE, + }) + monkeypatch.setenv(PluginManager.PLUGIN_PATH_ENV, f"acme_twice={plugin_dir}") + plugin_manager.reload() + + with caplog.at_level("WARNING"): + plugins = PipelineElement.list_plugins() + LocalDoubler = damast.plugins.acme_twice.LocalDoubler + + assert plugins["acme_twice.LocalDoubler"] == "acme_twice.main:LocalDoubler" + assert LocalDoubler.__module__ == "acme_twice.main" + assert any("registered by more than one source" in record.message for record in caplog.records) + assert any("provided by more than one source" in record.message for record in caplog.records) + + _reset_plugin_manager() + + +def test_pipeline_saved_from_named_plugin_dir_replays_against_installed_package( + named_plugin_dir, installed_package, fake_entry_points, monkeypatch): + saved_step = dict(PipelineElement.create_new(module_name="acme_named.main", class_name="LocalDoubler")) + + # elsewhere: the same code is installed as package 'acme_named' instead of a local directory + monkeypatch.delenv(PluginManager.PLUGIN_PATH_ENV) + plugin_manager.reload() + installed_package("acme_named", NAMED_PACKAGE_FILES) + fake_entry_points("acme_named", "acme_named") + + instance = PipelineElement.create_new(**saved_step) + assert type(instance).__module__ == "acme_named.main" + assert "site" in Path(sys.modules["acme_named.main"].__file__).parts