localstack-plugin-loader

localstack-plugin-loader是LocalStack中使用的动态代码加载框架。

概述

localstack-plugin-loader在Python的入口点机制周围构建了一个更高级的插件机制。它提供工具在运行时从入口点加载插件，并在构建时从插件发现入口点（因此您不需要在setup.py中静态声明入口点）。

核心概念

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

加载插件

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

发现入口点

在构建时（例如，使用python setup.py develop/install/sdist），一个特殊的PluginFinder收集可以解释为PluginSpec的任何内容，并从中创建setuptools入口点。在setup.py中，我们可以使用plugin.setuptools.load_entry_points方法收集一个字典，用于setup()的entry_points值。

from plugin.setuptools import load_entry_points

setup(
    entry_points=load_entry_points(exclude=("tests", "tests.*",))
)

请注意，load_entry_points将尝试从.egg-info目录解析缓存版本的entry_points.txt，以避免在从源分布构建包时解析入口点。

示例

要使用插件框架构建某些内容，您首先需要引入一个在加载时执行某些操作的插件。然后，在运行时，您需要一个使用PluginManager获取这些插件的组件。

每个插件一个类

这是我们的LocalstackCliPlugin采取的方法。每个插件类（例如，ProCliPlugin）本质上是一个单例。这很简单，因为这些类作为插件被发现。只需创建一个具有名称和命名空间的插件类，它就会被构建时的PluginFinder发现。

# 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类执行相同的事情，但使用它的多个实例。我们可以使用PluginFactory以及模块级别定义的PluginSpec实例的可发现性（受pluggy启发）来实现这一点。

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来构建一个监督器

class Supervisor:
    manager: PluginManager[ServicePlugin]

    def start(self, service_name):
        plugin = 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，您将收到FunctionPlugin实例，您可以像调用函数一样调用它们

runtime = LocalstackRuntime()

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

安装

pip install localstack-plugin-loader

开发

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

make venv
make test

运行代码格式化器

make format

使用twine上传pypi包

make upload