Skip to content

Commit 2d78d93

Browse files
authored
Merge pull request #26 from simula/fix/plugin_mechanism
fix/plugin mechanism
2 parents 3c2a4a0 + 67820c1 commit 2d78d93

16 files changed

Lines changed: 868 additions & 236 deletions

File tree

‎docs/cli.rst‎

Lines changed: 36 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -345,7 +345,7 @@ Generating small synthetic datasets to try it on:
345345

346346
# JoinByTimestamp is a local plugin transformer (see Plugins below) - needed both to
347347
# build the pipeline above and to load it back for damast process
348-
export DAMAST_PLUGIN_PATH=docs/examples/plugins
348+
export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins
349349
damast process --pipeline pipelines/osint_ais_preparation.damast.ppl \
350350
--input-data df=docs/examples/data/ais.parquet \
351351
--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
365365

366366
::
367367

368-
export DAMAST_PLUGIN_PATH=docs/examples/plugins
368+
export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins
369369
damast process --pipeline pipelines/osint_ais_preparation.damast.ppl --describe
370370

371371
.. highlight:: none
@@ -624,20 +624,35 @@ registering such plugin transformers are supported - see :class:`damast.core.tra
624624
for the full API:
625625

626626
- installable packages that declare their :class:`damast.core.transformations.PipelineElement` subclasses via the
627-
``damast.transformers`` entry-point group in their own ``pyproject.toml``::
627+
``damast.transformers`` entry-point group in their own ``pyproject.toml`` - either one entry per class, or
628+
one entry per module, which registers every transformer defined in that module (or, for a package, in its
629+
top-level submodules)::
628630

629631
[project.entry-points."damast.transformers"]
630632
MyTransformer = "acme_pkg.transformers:MyTransformer"
633+
acme_pkg = "acme_pkg.transformers"
631634

632-
- local, ad-hoc ``*.py`` files that are not part of any installed package, made discoverable by
633-
pointing the ``DAMAST_PLUGIN_PATH`` environment variable at the directory (or directories,
634-
separated with ``os.pathsep``) that contains them
635+
- local directories that are not part of any installed package, made discoverable via the
636+
``DAMAST_PLUGIN_PATH`` environment variable (several entries separated with ``os.pathsep``). An entry
637+
``name=path`` loads the directory as a package called ``name``: its top-level ``*.py`` files become
638+
``name.<file>``, and may use relative imports (``from .helpers import x``), also into subdirectories.
639+
The same can be done in code via ``plugin_manager.register_plugin_package(name, path)``.
640+
Choose a distinctive name - one that is already importable is rejected. A bare ``path`` entry
641+
(deprecated) loads each file as a flat module named after the file, without relative imports.
642+
643+
The package name of a local directory is recorded in saved pipelines (e.g.
644+
``module_name: my_plugins.my_transformers``), so a pipeline can only be replayed with the directory
645+
registered under the same name - or with an installed package of that name, e.g. once the directory has
646+
been turned into an installable package with a module entry-point.
635647

636648
Regardless of which of the two a transformer comes from, it is resolvable in code the same way,
637-
via the ``damast.plugins`` namespace::
649+
via ``damast.plugins.<package>`` - where ``<package>`` is the top-level package of the module
650+
defining it, i.e. ``acme_pkg`` for ``acme_pkg.transformers:MyTransformer``, or the name a local
651+
directory was registered under::
638652

639-
from damast.plugins import MyTransformer
653+
from damast.plugins.acme_pkg import MyTransformer
640654

655+
Scoping by package means two plugins can provide a transformer of the same name without clashing.
641656
``damast.plugins`` resolves names lazily on first access, so nothing beyond the requested class
642657
is ever imported - see :mod:`damast.plugins` for details.
643658

@@ -650,27 +665,29 @@ subclass in a loose ``*.py`` file, written like any other transformer:
650665
.. literalinclude:: ./examples/plugins/my_transformers.py
651666
:language: Python
652667

653-
With ``DAMAST_PLUGIN_PATH`` pointing at the directory containing that file, ``damast plugins``
654-
lists it without requiring any further Python code:
668+
With ``DAMAST_PLUGIN_PATH`` registering the directory containing that file (here as package
669+
``my_plugins``), ``damast plugins`` lists it without requiring any further Python code:
655670

656671
.. highlight:: none
657672

658673
::
659674

660-
$ export DAMAST_PLUGIN_PATH=./examples/plugins
675+
$ export DAMAST_PLUGIN_PATH=my_plugins=./examples/plugins
661676
$ damast plugins
662677

663-
MyTripler: my_transformers:MyTripler
678+
my_plugins (local: examples/plugins)
679+
JoinByTimestamp .osint_ais_transformers
680+
MyTripler .my_transformers
664681

665-
``MyTripler`` is now resolvable via ``damast.plugins`` and can be used in a pipeline like any
666-
other transformer:
682+
``MyTripler`` is now resolvable via ``damast.plugins.my_plugins`` and can be used in a pipeline
683+
like any other transformer:
667684

668685
.. literalinclude:: ./examples/damast-plugin-pipeline.py
669686
:language: Python
670687

671688
The resulting pipeline can be applied like any other, e.g. via ``damast process`` (see `Process`_
672-
above), as long as ``DAMAST_PLUGIN_PATH`` is still set to a directory containing
673-
``my_transformers.py``:
689+
above), as long as ``DAMAST_PLUGIN_PATH`` still registers the directory containing
690+
``my_transformers.py`` as ``my_plugins``:
674691

675692
.. highlight:: none
676693

@@ -679,7 +696,7 @@ above), as long as ``DAMAST_PLUGIN_PATH`` is still set to a directory containing
679696
damast process --input-data data.parquet --pipeline pipelines/my-plugin-pipeline.damast.ppl
680697

681698
Pipelines saved with a plugin transformer record where it came from under ``requires`` (the
682-
installed distribution and version, or the original local file path), so that loading the pipeline
683-
elsewhere fails with an actionable message - naming the missing package to ``pip install``, or the
684-
``DAMAST_PLUGIN_PATH`` directory to add - instead of a bare import error.
699+
installed distribution and version, or the local package name and directory), so that loading the
700+
pipeline elsewhere fails with an actionable message - naming the missing package to ``pip install``, or
701+
the ``DAMAST_PLUGIN_PATH`` entry to add - instead of a bare import error.
685702

‎docs/examples/damast-osint-ais-pipeline.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,10 @@
1616

1717
# JoinByTimestamp lives in its own package/file in a real project - see docs/cli.rst >
1818
# Plugins. Here it's a local plugin file, resolved the same way.
19-
os.environ["DAMAST_PLUGIN_PATH"] = str(Path(__file__).parent / "plugins")
19+
os.environ["DAMAST_PLUGIN_PATH"] = f"ais_osint={Path(__file__).parent / 'plugins'}"
2020

2121
from damast.core.dataprocessing import DataProcessingPipeline
22-
from damast.plugins import JoinByTimestamp
22+
from damast.plugins.ais_osint import JoinByTimestamp
2323

2424
pipeline = DataProcessingPipeline(name="osint_ais_preparation",
2525
description="Join AIS pings with OSINT events by timestamp",

‎docs/examples/damast-plugin-pipeline.py‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,13 @@
11
import os
22
from pathlib import Path
33

4-
# DAMAST_PLUGIN_PATH points at the directory holding my_transformers.py, so that
5-
# 'MyTripler' becomes resolvable via damast.plugins - see docs/examples/plugins/.
6-
os.environ["DAMAST_PLUGIN_PATH"] = str(Path(__file__).parent / "plugins")
4+
# DAMAST_PLUGIN_PATH registers the directory holding my_transformers.py as package
5+
# 'my_plugins', so that 'MyTripler' becomes resolvable via damast.plugins.my_plugins - see
6+
# docs/examples/plugins/.
7+
os.environ["DAMAST_PLUGIN_PATH"] = f"my_plugins={Path(__file__).parent / 'plugins'}"
78

89
from damast.core import DataProcessingPipeline
9-
from damast.plugins import MyTripler
10+
from damast.plugins.my_plugins import MyTripler
1011

1112
pipeline = DataProcessingPipeline(name="my-plugin-pipeline", base_dir=".")
1213
pipeline.add("Triple mmsi",

‎docs/usage.ipynb‎

Lines changed: 18 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -337,30 +337,34 @@
337337
"Python package at all. Damast supports two ways to use `PipelineElement` classes that live\n",
338338
"outside of `damast` itself, so that pipelines using them stay reproducible when shared with\n",
339339
"others. Regardless of which of the two a transformer comes from, it is always resolvable the\n",
340-
"same way in code - `from damast.plugins import MyTransformer` - without needing to import (or\n",
341-
"even know) the underlying module or package that defines it.\n",
340+
"same way in code - `from damast.plugins.<package> import MyTransformer`, where `<package>` is\n",
341+
"the plugin's top-level package - without needing to import (or even know) the underlying module\n",
342+
"that defines it. Scoping by package means two plugins can provide a transformer of the same name\n",
343+
"without clashing.\n",
342344
"\n",
343345
"### Local, ad-hoc transformers via `DAMAST_PLUGIN_PATH`\n",
344346
"\n",
345347
"For a quick transformer that does not warrant its own package, put it in a plain `*.py` file and\n",
346-
"point the `DAMAST_PLUGIN_PATH` environment variable at the directory containing it (multiple\n",
347-
"directories can be separated with `os.pathsep`, just like `PATH`). Every top-level file found\n",
348-
"there is imported once - using its filename stem as the module name - so any `PipelineElement`\n",
349-
"subclasses it defines become resolvable exactly like classes from an installed package."
348+
"register the directory containing it under a package name via the `DAMAST_PLUGIN_PATH`\n",
349+
"environment variable, as `name=path` (multiple entries can be separated with `os.pathsep`, just\n",
350+
"like `PATH`). Every top-level file found there is imported once as `name.<file>` - so files may\n",
351+
"use relative imports among each other - and any `PipelineElement` subclasses they define become\n",
352+
"resolvable exactly like classes from an installed package. Choose a distinctive name: it is\n",
353+
"recorded in saved pipelines, and a name that is already importable is rejected."
350354
]
351355
},
352356
{
353357
"cell_type": "code",
354358
"execution_count": null,
355359
"metadata": {},
356360
"outputs": [],
357-
"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()"
361+
"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()"
358362
},
359363
{
360364
"cell_type": "markdown",
361365
"metadata": {},
362366
"source": [
363-
"`Doubler` is now resolvable via `damast.plugins`, and can be used in a pipeline like any other\n",
367+
"`Doubler` is now resolvable via `damast.plugins.my_plugins`, and can be used in a pipeline like any other\n",
364368
"transformer. `damast.plugins` resolves names lazily on first access - nothing beyond the\n",
365369
"requested class is imported, and it works no matter which of the two plugin sources defines\n",
366370
"it. Saving the pipeline records where the transformer came from, under `requires`, so a pipeline\n",
@@ -375,7 +379,7 @@
375379
"outputs": [],
376380
"source": [
377381
"from damast.core.dataprocessing import DataProcessingPipeline\n",
378-
"from damast.plugins import (\n",
382+
"from damast.plugins.my_plugins import (\n",
379383
" Doubler, # resolved from the file we just wrote to `plugin_dir`\n",
380384
")\n",
381385
"\n",
@@ -401,12 +405,16 @@
401405
"MyTransformer = \"acme_pkg.transformers:MyTransformer\"\n",
402406
"```\n",
403407
"\n",
408+
"Instead of one entry per class, an entry may also name a whole module, e.g.\n",
409+
"`acme_pkg = \"acme_pkg.transformers\"`: every `PipelineElement` defined in that module - or, for a\n",
410+
"package, in its top-level submodules - is then registered.\n",
411+
"\n",
404412
"Once installed, it is discovered the same way as a local plugin - `PipelineElement.list_plugins()`\n",
405413
"merges both sources - and pipelines saved with it additionally record the distribution name and\n",
406414
"version under `requires`, so a version mismatch on reload is logged as a warning rather than\n",
407415
"silently changing behavior. Run `damast plugins` from the command line to list everything that is\n",
408416
"currently discoverable, from either source, without writing any Python. In code, it resolves the\n",
409-
"same way as a local plugin too - `from damast.plugins import MyTransformer`."
417+
"same way as a local plugin too - `from damast.plugins.acme_pkg import MyTransformer`."
410418
]
411419
},
412420
{

‎pyproject.toml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,8 @@ repository = "https://github.com/simula/damast"
7171
# Third-party packages can register PipelineElement subclasses as damast plugins via
7272
# their own pyproject.toml, e.g.:
7373
# [project.entry-points."damast.transformers"]
74-
# MyTransformer = "acme_pkg.transformers:MyTransformer"
74+
# MyTransformer = "acme_pkg.transformers:MyTransformer" # a single class
75+
# acme_pkg = "acme_pkg.transformers" # all classes of a module/package
7576
# Discoverable via `damast plugins` or PipelineElement.list_plugins().
7677

7778
[project.optional-dependencies]

‎src/damast/cli/plugins.py‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
from argparse import ArgumentParser
33

44
from damast.cli.base import BaseParser
5-
from damast.core.transformations import PipelineElement, PluginManager
5+
from damast.core.transformations import PipelineElement, PluginManager, plugin_manager
66

77

88
class PluginsParser(BaseParser):
@@ -21,5 +21,26 @@ def execute(self, args):
2121
f"{PluginManager.PLUGIN_PATH_ENV}={plugin_path})")
2222
return
2323

24-
for name, target in sorted(plugins.items()):
25-
print(f"{name}: {target}")
24+
# '<package>.<class>' -> 'module:class', grouped per plugin package
25+
packages: dict[str, list[tuple[str, str]]] = {}
26+
for qualified_name, target in plugins.items():
27+
package, class_name = qualified_name.split(".", 1)
28+
packages.setdefault(package, []).append((class_name, target.split(":")[0]))
29+
30+
for package, transformers in sorted(packages.items()):
31+
source = self._describe_source(plugin_manager.resolve_requirement(transformers[0][1]))
32+
print(f"{package}{f' ({source})' if source else ''}")
33+
34+
width = max(len(class_name) for class_name, _ in transformers)
35+
for class_name, module_name in sorted(transformers):
36+
# module relative to the package - omitted if defined in the package module itself
37+
relative_module = module_name.removeprefix(package)
38+
print(f" {class_name:<{width}} {relative_module}".rstrip())
39+
40+
@staticmethod
41+
def _describe_source(requirement: dict[str, str] | None) -> str:
42+
if not requirement:
43+
return ""
44+
if requirement.get("distribution"):
45+
return f"{requirement['distribution']}=={requirement['version']}"
46+
return f"local: {requirement['path']}"

‎src/damast/core/polars_dataframe.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -631,6 +631,8 @@ def read_batches(with_columns: list[str] | None,
631631
n_rows -= df.height
632632
yield df
633633

634+
# According to polars documentation this functionality is considered unstable
635+
# https://docs.pola.rs/api/python/stable/reference/api/polars.io.plugins.register_io_source.html
634636
return register_io_source(read_batches, schema=schema), variables
635637

636638
@staticmethod

0 commit comments

Comments
 (0)