Plux

plux是LocalStack中使用的动态代码加载框架。

概述

Plux 在 Python 的入口点机制周围构建了一个更高层次的插件机制。它提供了在运行时从入口点加载插件和在构建时从插件中发现入口点（因此你不必在 setup.py 中静态声明入口点）的工具。

核心概念

• PluginSpec：描述了一个 Plugin。每个插件都有一个命名空间，该命名空间中唯一的名称，以及一个 PluginFactory（创建 Plugin 的对象。在 simplest 情况下，那可能只是 Plugin 的类）。
• Plugin：一个公开 should_load 和 load 方法的对象。请注意，它不作为领域对象（它不持有插件的生命周期状态，如已初始化、已加载等，或插件的其他元数据）。
• PluginFinder：在构建时（通过使用 pkgutil 和 setuptools 扫描模块）或运行时（使用 importlib 读取分发的入口点）查找插件。
• PluginManager：管理插件的运行时生命周期，它有三个状态：
  • 已解析：指向 PluginSpec 的入口点已导入，并创建了 PluginSpec 实例。
  • 初始化：PluginSpec 的 PluginFactory 已成功调用。
  • 已加载：Plugin 的 load 方法已成功调用。

加载插件

在运行时，PluginManager 使用一个 PluginFinder，该 PluginFinder 又使用 importlib 扫描可用的入口点以查找看起来像 PluginSpec 的东西。使用 PluginManager.load(name: str) 或 PluginManager.load_all()，可以加载命名空间内的插件，这些插件在入口点中可被发现。如果在生命周期的任何状态下发生错误，则 PluginManager 会通知 PluginLifecycleListener，但会继续运行。

发现入口点

要构建带有插件作为入口点的源分发和 wheel，只需运行 python setup.py plugins sdist bdist_wheel。如果您没有 setup.py，则可以使用 plux 构建前端并运行 python -m plux entrypoints。

工作原理：为了在构建时发现插件，plux 提供了一个自定义 setuptools 命令 plugins，通过 python setup.py plugins 调用。该命令使用一个特殊的 PluginFinder，从代码库中收集任何可以解释为 PluginSpec 的东西，并从中创建一个插件索引文件 plux.json，该文件放置在 .egg-info 分发元数据目录中。当使用 setuptools 命令创建分发时（例如，python setup.py sdist/bdist_wheel/...），plux 会找到 plux.json 插件索引并自动扩展入口点列表（收集到 .egg-info/entry_points.txt）。plux.json 文件成为分发的一部分，因此插件不必每次在别处安装分发时都进行发现。使用 python -m build 时也支持构建时发现，因为它调用已注册的 setuptools 脚本。

示例

要使用插件框架构建某些东西，您首先想要引入一个在加载时做某事的插件。然后，在运行时，您需要一个使用 PluginManager 来获取这些插件的功能组件。

每个插件一个类

这是我们在 LocalstackCliPlugin 中的做法。每个插件类（例如，ProCliPlugin）本质上是一个单例。这是很容易的，因为类可以被发现为插件。只需创建一个带有名称和命名空间的 Plugin 类，它就会被构建时的 PluginFinder 发现。

from plux import Plugin

# abstract case (not discovered at build time, missing name)
class CliPlugin(Plugin):
    namespace = "my.plugins.cli"

    def load(self, cli):
        self.attach(cli)

    def attach(self, cli):
        raise NotImplementedError

# discovered at build time (has a namespace, name, and is a Plugin)
class MyCliPlugin(CliPlugin):
    name = "my"

    def attach(self, cli):
        # ... attach commands to cli object

现在我们需要一个 PluginManager（它有一个泛型类型）来帮我们加载插件

cli = # ... needs to come from somewhere

manager: PluginManager[CliPlugin] = PluginManager("my.plugins.cli", load_args=(cli,))

plugins: List[CliPlugin] = manager.load_all()

# todo: do stuff with the plugins, if you want/need
#  in this example, we simply use the plugin mechanism to run a one-shot function (attach) on a load argument

可重用插件

当您有很多以相似方式组织的插件时，我们可能不想为每个插件创建一个单独的Plugin类。相反，我们希望使用相同的Plugin类来完成相同的事情，但使用它的几个实例。可以使用PluginFactory以及模块级别定义的PluginSpec实例的可发现性（灵感来自pluggy）来实现这一点。

from plux import Plugin, PluginFactory, PluginSpec
import importlib

class ServicePlugin(Plugin):

    def __init__(self, service_name):
        self.service_name = service_name
        self.service = None

    def should_load(self):
        return self.service_name in config.SERVICES

    def load(self):
        module = importlib.import_module("localstack.services.%s" % self.service_name)
        # suppose we define a convention that each service module has a Service class, like moto's `Backend`
        self.service = module.Service()

def service_plugin_factory(name) -> PluginFactory:
    def create():
        return ServicePlugin(name)

    return create

# discoverable
s3 = PluginSpec("localstack.plugins.services", "s3", service_plugin_factory("s3"))

# discoverable
dynamodb = PluginSpec("localstack.plugins.services", "dynamodb", service_plugin_factory("dynamodb"))

# ... could be simplified with convenience framework code, but the principle will stay the same

然后我们可以使用PluginManager来构建一个Supervisor

from plux import PluginManager

class Supervisor:
    manager: PluginManager[ServicePlugin]

    def start(self, service_name):
        plugin = self.manager.load(service_name)
        service = plugin.service
        service.start()

作为插件的函数

使用@plugin装饰器，您可以将函数暴露为插件。它们将被框架包装为FunctionPlugin实例，这些实例满足插件和函数的约定。

from plugin import plugin

@plugin(namespace="localstack.configurators")
def configure_logging(runtime):
    logging.basicConfig(level=runtime.config.loglevel)

    
@plugin(namespace="localstack.configurators")
def configure_somethingelse(runtime):
    # do other stuff with the runtime object
    pass

通过load_all使用PluginManager，您将收到可以像函数一样调用的FunctionPlugin实例。

runtime = LocalstackRuntime()

for configurator in PluginManager("localstack.configurators").load_all():
    configurator(runtime)

配置您的发行版

如果您正在构建一个暴露由plux发现的插件的Python发行版，您需要配置项目的构建系统，以便在安装您的发行版时创建entry_points.txt文件。

对于pyproject.toml模板，这涉及到添加build-system部分

[build-system]
requires = ['setuptools', 'wheel', 'plux>=1.3.1']
build-backend = "setuptools.build_meta"

# ...

安装

pip install plux

开发

创建虚拟环境，安装依赖项，并运行测试

make venv
make test

运行代码格式化器

make format

使用twine上传pypi包

make upload