python - 实现一个通用的插件类

本文提供了一种插件类的实现方案。

定义插件管理器

插件管理器用于注册、销毁、执行插件。


import abc
from functools import wraps
from typing import Callable, Dictfrom pydantic import (BaseModel,validate_arguments,ValidationError as PydanticValidationError,
)def import_string(dotted_path: str) -> Callable:"""Import a dotted module path and return the attribute/class designated by thelast name in the path. Raise ImportError if the import failed.Args:dotted_path: 字符串表示的模块类，module.classReturns:返回加载的模块中的对象"""try:module_path, class_name = dotted_path.rsplit(".", 1)except ValueError:raise ImportError("{} doesn't look like a module path".format(dotted_path))module: ModuleType = import_module(module_path)try:# 返回模块中的类return getattr(module, class_name)except AttributeError:raise ImportError('Module "{}" does not define a "{}" attribute/class'.format(module_path, class_name))class FunctionsManager:"""函数管理器 ."""# 存放注册的可执行对象__hub = {}  # type: ignore@classmethoddef register_invocation_cls(cls, invocation_cls: InvocationMeta, name=None) -> None:if not name:func_name = invocation_cls.Meta.func_nameelse:func_name = nameif not isinstance(func_name, str):raise ValueError(f"func_name {func_name} should be string")existed_invocation_cls = cls.__hub.get(func_name)if existed_invocation_cls:raise RuntimeError("func register error, {}'s func_name {} conflict with {}".format(existed_invocation_cls, func_name, invocation_cls))# 存放类的实例cls.__hub[func_name] = invocation_cls()@classmethoddef register_funcs(cls, func_dict) -> None:for func_name, func_obj in func_dict.items():if not isinstance(func_name, str):raise ValueError(f"func_name {func_name} should be string")if func_name in cls.__hub:raise ValueError("func register error, {}'s func_name {} conflict with {}".format(func_obj, func_name, cls.__hub[func_name]))if isinstance(func_obj, str):func = import_string(func_obj)elif isinstance(func_obj, Callable):func = func_objelse:raise ValueError("func register error, {} is not be callable".format(func_obj, func_name))cls.__hub[func_name] = func@classmethoddef clear(cls) -> None:"""清空注册信息 ."""cls.__hub = {}@classmethoddef all_funcs(cls) -> Dict:"""获得所有的注册信息. """return cls.__hub@classmethoddef get_func(cls, func_name: str) -> Callable:"""获得注册的函数 ."""func_obj = cls.__hub.get(func_name)if not func_obj:raise ValueError("func object {} not found".format(func_name))return func_obj@classmethoddef func_call(cls, func_name: str, *args, **kwargs):"""根据函数名执行注册的函数 ."""func = cls.get_func(func_name)return func(*args, **kwargs)

定义元类

派生的类可自行注册到插件管理器。

class InvocationMeta(type):"""Metaclass for function invocation"""def __new__(cls, name, bases, dct):# ensure initialization is only performed for subclasses of Pluginparents = [b for b in bases if isinstance(b, InvocationMeta)]if not parents:return super().__new__(cls, name, bases, dct)new_cls = super().__new__(cls, name, bases, dct)# meta validationmeta_obj = getattr(new_cls, "Meta", None)if not meta_obj:raise AttributeError("Meta class is required")func_name = getattr(meta_obj, "func_name", None)if not func_name:raise AttributeError("func_name is required in Meta")desc = getattr(meta_obj, "desc", None)if desc is not None and not isinstance(desc, str):raise AttributeError("desc in Meta should be str")# register funcFunctionsManager.register_invocation_cls(new_cls)return new_cls

定义元类的一个抽象派生类

支持参数验证。

class BaseInvocation(metaclass=InvocationMeta):"""Base class for function invocation"""class Inputs(BaseModel):"""输入校验器"""pass@validate_arguments  # type: ignoredef __call__(self, *args, **kwargs):# 输入参数校验, 仅可能是 args 或 kwargs 之一try:params = {}if args:inputs_meta = getattr(self.Inputs, "Meta", None)inputs_ordering = getattr(inputs_meta, "ordering", None)if isinstance(inputs_ordering, list):if len(args) > len(inputs_ordering):raise Exception(f"Too many arguments for inputs: {args}")params = dict(zip(inputs_ordering, args))elif kwargs:params = kwargs# 参数校验if params:self.Inputs(**params)except PydanticValidationError as e:raise Exception(e)# 执行自定义业务逻辑return self.invoke(*args, **kwargs)@abc.abstractmethoddef invoke(self, *args, **kwargs):"""自定义业务逻辑 ."""raise NotImplementedError()

定义装饰器

def register_class(name: str):def _register_class(cls: BaseInvocation):FunctionsManager.register_invocation_cls(cls, name=name)@wraps(cls)def wrapper():return cls()return wrapperreturn _register_classdef register_func(name: str):def _register_func(func: Callable):FunctionsManager.register_funcs({name: func})@wraps(func)def wrapper(*args, **kwargs):return func(*args, **kwargs)return wrapperreturn _register_func

单元测试

from pydantic import BaseModelfrom .register import FunctionsManager, register_func, register_class, BaseInvocation@register_func("add")
def add(x: int, y: int) -> int:return x + yclass Add(BaseInvocation):class Meta:func_name = "multiply"class Inputs(BaseModel):"""输入校验器"""x: inty: intclass Meta:ordering = ["x", "y"]def invoke(self, x: int, y: int) -> int:return x * y@register_class("subtract")
class Subtract:class Inputs(BaseModel):"""输入校验器"""x: inty: intclass Meta:ordering = ["x", "y"]def __call__(self, x: int, y: int) -> int:return x - yclass TestFunctionsManager:def test_register_func(self):func = FunctionsManager.get_func("add")assert func(2, 3) == 5def test_register_class(self):func = FunctionsManager.get_func("subtract")assert func(2, 3) == -1def test_metaclass(self):func = FunctionsManager.get_func("multiply")assert func(2, 3) == 6