Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ with the `@plugin` decorator, you can expose functions as plugins. They will be
into `FunctionPlugin` instances, which satisfy both the contract of a Plugin, and that of the function.

```python
from plugin import plugin
from plux import plugin

@plugin(namespace="localstack.configurators")
def configure_logging(runtime):
Expand Down
179 changes: 110 additions & 69 deletions plux/build/discovery.py
Original file line number Diff line number Diff line change
@@ -1,24 +1,110 @@
"""
Buildtool independent utils to discover plugins from the codebase, and write index files.
Buildtool independent utils to discover plugins from the project's source code.
"""

import configparser
import importlib
import inspect
import json
import logging
import sys
import typing as t
from fnmatch import fnmatchcase
from types import ModuleType
import os
import pkgutil

from plux import PluginFinder, PluginSpecResolver, PluginSpec
from plux.core.entrypoint import discover_entry_points, EntryPointDict

if t.TYPE_CHECKING:
from _typeshed import SupportsWrite

LOG = logging.getLogger(__name__)


class PackageFinder:
"""
Generate a list of Python packages. How these are generated depends on the implementation.

Why this abstraction? The naive way to find packages is to list all directories with an ``__init__.py`` file.
However, this approach does not work for distributions that have namespace packages. How do we know whether
something is a namespace package or just a directory? Typically, the build configuration will tell us. For example,
setuptools has the following directives for ``pyproject.toml``::

[tool.setuptools]
packages = ["mypkg", "mypkg.subpkg1", "mypkg.subpkg2"]

Or in hatch::

[tool.hatch.build.targets.wheel]
packages = ["src/foo"]

So this abstraction allows us to use the build tool internals to generate a list of packages that we should be
scanning for plugins.
"""

def find_packages(self) -> t.Iterable[str]:
"""
Returns an Iterable of Python packages. Each item is a string-representation of a Python package (for example,
``plux.core``, ``myproject.mypackage.utils``, ...)

:return: An Iterable of Packages
"""
raise NotImplementedError

@property
def path(self) -> str:
"""
The root file path under which the packages are located.

:return: A file path
"""
raise NotImplementedError


class PluginFromPackageFinder(PluginFinder):
"""
Finds Plugins from packages that are resolved by the given ``PackageFinder``. Under the hood this uses a
``ModuleScanningPluginFinder``, which, for each package returned by the ``PackageFinder``, imports the package using
``importlib``, and scans the module for plugins.
"""

finder: PackageFinder

def __init__(self, finder: PackageFinder):
self.finder = finder

def find_plugins(self) -> list[PluginSpec]:
collector = ModuleScanningPluginFinder(self._load_modules())
return collector.find_plugins()

def _load_modules(self) -> t.Generator[ModuleType, None, None]:
"""
Generator to load all imported modules that are part of the packages returned by the ``PackageFinder``.

:return: A generator of python modules
"""
for module_name in self._list_module_names():
try:
yield importlib.import_module(module_name)
except Exception as e:
LOG.error("error importing module %s: %s", module_name, e)

def _list_module_names(self) -> set[str]:
"""
This method creates a set of module names by iterating over the packages detected by the ``PackageFinder``. It
includes top-level packages, as well as submodules found within those packages.

:return: A set of strings where each string represents a module name.
"""
# adapted from https://stackoverflow.com/a/54323162/804840

modules = set()

for pkg in self.finder.find_packages():
modules.add(pkg)
pkgpath = self.finder.path.rstrip(os.sep) + os.sep + pkg.replace(".", os.sep)
for info in pkgutil.iter_modules([pkgpath]):
if not info.ispkg:
modules.add(pkg + "." + info.name)

return modules


class ModuleScanningPluginFinder(PluginFinder):
"""
A PluginFinder that scans the members of given modules for available PluginSpecs. Each member is evaluated with a
Expand Down Expand Up @@ -51,72 +137,27 @@ def find_plugins(self) -> list[PluginSpec]:
return plugins


class PluginIndexBuilder:
class Filter:
"""
Builds an index file containing all discovered plugins. The index file can be written to stdout, or to a file.
The writer supports two formats: json and ini.
Given a list of patterns, create a callable that will be true only if
the input matches at least one of the patterns.
This is from `setuptools.discovery._Filter`
"""

def __init__(
self,
plugin_finder: PluginFinder,
output_format: t.Literal["json", "ini"] = "json",
):
self.plugin_finder = plugin_finder
self.output_format = output_format
def __init__(self, patterns: t.Iterable[str]):
self._patterns = patterns

def write(self, fp: "SupportsWrite[str]" = sys.stdout) -> EntryPointDict:
"""
Discover entry points using the configured ``PluginFinder``, and write the entry points into a file.
def __call__(self, item: str):
return any(fnmatchcase(item, pat) for pat in self._patterns)

:param fp: The file-like object to write to.
:return: The discovered entry points that were written into the file.
"""
ep = discover_entry_points(self.plugin_finder)

# sort entrypoints alphabetically in each group first
for group in ep:
ep[group].sort()

if self.output_format == "json":
json.dump(ep, fp, sort_keys=True, indent=2)
elif self.output_format == "ini":
cfg = configparser.ConfigParser()
cfg.read_dict(self.convert_to_nested_entry_point_dict(ep))
cfg.write(fp)
else:
raise ValueError(f"unknown plugin index output format {self.output_format}")
class MatchAllFilter(Filter):
"""
Filter that is equivalent to ``_Filter(["*"])``.
"""

return ep
def __init__(self):
super().__init__([])

@staticmethod
def convert_to_nested_entry_point_dict(ep: EntryPointDict) -> dict[str, dict[str, str]]:
"""
Converts and ``EntryPointDict`` to a nested dict, where the keys are the section names and values are
dictionaries. Each dictionary maps entry point names to their values. It also sorts the output alphabetically.

Example:
Input EntryPointDict:
{
'console_scripts': ['app=module:main', 'tool=module:cli'],
'plux.plugins': ['plugin1=pkg.module:Plugin1']
}

Output nested dict:
{
'console_scripts': {
'app': 'module:main',
'tool': 'module:cli'
},
'plux.plugins': {
'plugin1': 'pkg.module:Plugin1'
}
}
"""
result = {}
for section_name in sorted(ep.keys()):
result[section_name] = {}
for entry_point in sorted(ep[section_name]):
name, value = entry_point.split("=")
result[section_name][name] = value
return result
def __call__(self, item: str):
return True
6 changes: 6 additions & 0 deletions plux/build/hatchling.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
from plux.build.project import Project


class HatchlingProject(Project):
# TODO: implement me
pass
89 changes: 89 additions & 0 deletions plux/build/index.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
"""
Code to manage the plux plugin index file. The index file contains all discovered plugins, which later is used to
generate entry points.
"""

import configparser
import json
import sys
import typing as t

from plux.core.plugin import PluginFinder
from plux.core.entrypoint import EntryPointDict, discover_entry_points

if t.TYPE_CHECKING:
from _typeshed import SupportsWrite


class PluginIndexBuilder:
"""
Builds an index file containing all discovered plugins. The index file can be written to stdout, or to a file.
The writer supports two formats: json and ini.
"""

def __init__(
self,
plugin_finder: PluginFinder,
):
self.plugin_finder = plugin_finder

def write(
self,
fp: "SupportsWrite[str]" = sys.stdout,
output_format: t.Literal["json", "ini"] = "json",
) -> EntryPointDict:
"""
Discover entry points using the configured ``PluginFinder``, and write the entry points into a file.

:param fp: The file-like object to write to.
:param output_format: The format to write the entry points in. Can be either "json" or "ini".
:return: The discovered entry points that were written into the file.
"""
ep = discover_entry_points(self.plugin_finder)

# sort entrypoints alphabetically in each group first
for group in ep:
ep[group].sort()

if output_format == "json":
json.dump(ep, fp, sort_keys=True, indent=2)
elif output_format == "ini":
cfg = configparser.ConfigParser()
cfg.read_dict(self.convert_to_nested_entry_point_dict(ep))
cfg.write(fp)
else:
raise ValueError(f"unknown plugin index output format {output_format}")

return ep

@staticmethod
def convert_to_nested_entry_point_dict(ep: EntryPointDict) -> dict[str, dict[str, str]]:
"""
Converts and ``EntryPointDict`` to a nested dict, where the keys are the section names and values are
dictionaries. Each dictionary maps entry point names to their values. It also sorts the output alphabetically.

Example:
Input EntryPointDict:
{
'console_scripts': ['app=module:main', 'tool=module:cli'],
'plux.plugins': ['plugin1=pkg.module:Plugin1']
}

Output nested dict:
{
'console_scripts': {
'app': 'module:main',
'tool': 'module:cli'
},
'plux.plugins': {
'plugin1': 'pkg.module:Plugin1'
}
}
"""
result = {}
for section_name in sorted(ep.keys()):
result[section_name] = {}
for entry_point in sorted(ep[section_name]):
name, value = entry_point.split("=")
result[section_name][name] = value
return result
76 changes: 76 additions & 0 deletions plux/build/project.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
import os
from pathlib import Path

from plux.build.config import PluxConfiguration, read_plux_config_from_workdir
from plux.build.discovery import PackageFinder, PluginFromPackageFinder
from plux.build.index import PluginIndexBuilder


class Project:
"""
Abstraction for a python project to hide the details of a build tool from the CLI. A Project provides access to
the project's configuration, package discovery, and the entrypoint build mechanism.
"""

workdir: Path
config: PluxConfiguration

def __init__(self, workdir: str = None):
self.workdir = Path(workdir or os.curdir)
self.config = self.read_static_plux_config()

def find_entry_point_file(self) -> Path:
"""
Finds the entry_point.txt file of the current project. In case of setuptools, this may be in the
``<package>.egg-info`` directory, in case of hatch, where ``pip install -e .`` has become the standard, the
entrypoints file lives in the ``.dist-info`` directory of the venv.

:return: A path pointing to the entrypoints file. The file might not exist.
"""
raise NotImplementedError

def find_plux_index_file(self) -> Path:
"""
Returns the plux index file location. This may depend on the build tool, for similar reasons described in
``find_entry_point_file``. For example, in setuptools, the plux index file by default is in
``.egg-info/plux.json``.

:return: A path pointing to the plux index file.
"""
raise NotImplementedError

def create_package_finder(self) -> PackageFinder:
"""
Returns a build tool-specific PackageFinder instance that can be used to discover packages to scan for plugins.

:return: A PackageFinder instance
"""
raise NotImplementedError

def build_entrypoints(self):
"""
Routine to build the entrypoints file using ``EntryPointBuildMode.BUILD_HOOK``. This is called by the CLI
frontend. It's build tool-specific since we need to hook into the build process that generates the
``entry_points.txt``.
"""
raise NotImplementedError

def create_plugin_index_builder(self) -> PluginIndexBuilder:
"""
Returns a PluginIndexBuilder instance that can be used to build the plugin index.

The default implementation creates a PluginFromPackageFinder instance using ``create_package_finder``.

:return: A PluginIndexBuilder instance.
"""
plugin_finder = PluginFromPackageFinder(self.create_package_finder())
return PluginIndexBuilder(plugin_finder)

def read_static_plux_config(self) -> PluxConfiguration:
"""
Reads the static configuration (``pyproject.toml``) from the Project's working directory using
``read_read_plux_config_from_workdir``.

:return: A PluxConfiguration object
"""
return read_plux_config_from_workdir(str(self.workdir))
Loading