|
目录
本文提供了一种插件类的实现方案。
定义插件管理器
插件管理器用于注册、销毁、执行插件。- import abc
- from functools import wraps
- from typing import Callable, Dict
- from 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 the
- last name in the path. Raise ImportError if the import failed.
- Args:
- dotted_path: 字符串表示的模块类,module.class
- Returns:
- 返回加载的模块中的对象
- """
- 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
- @classmethod
- def register_invocation_cls(cls, invocation_cls: InvocationMeta, name=None) -> None:
- if not name:
- func_name = invocation_cls.Meta.func_name
- else:
- func_name = name
- if 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()
- @classmethod
- def 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_obj
- else:
- raise ValueError(
- "func register error, {} is not be callable".format(
- func_obj, func_name
- )
- )
- cls.__hub[func_name] = func
- @classmethod
- def clear(cls) -> None:
- """清空注册信息 ."""
- cls.__hub = {}
- @classmethod
- def all_funcs(cls) -> Dict:
- """获得所有的注册信息. """
- return cls.__hub
- @classmethod
- def 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
- @classmethod
- def 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 Plugin
- parents = [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 validation
- meta_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 func
- FunctionsManager.register_invocation_cls(new_cls)
- return new_cls
复制代码 定义元类的一个抽象派生类
支持参数验证。- class BaseInvocation(metaclass=InvocationMeta):
- """
- Base class for function invocation
- """
- class Inputs(BaseModel):
- """
- 输入校验器
- """
- pass
- @validate_arguments # type: ignore
- def __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.abstractmethod
- def 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 wrapper
- return _register_class
- def 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 wrapper
- return _register_func
复制代码 单元测试
- from pydantic import BaseModel
- from .register import FunctionsManager, register_func, register_class, BaseInvocation
- @register_func("add")
- def add(x: int, y: int) -> int:
- return x + y
- class Add(BaseInvocation):
- class Meta:
- func_name = "multiply"
- class Inputs(BaseModel):
- """
- 输入校验器
- """
- x: int
- y: int
- class Meta:
- ordering = ["x", "y"]
- def invoke(self, x: int, y: int) -> int:
- return x * y
- @register_class("subtract")
- class Subtract:
- class Inputs(BaseModel):
- """
- 输入校验器
- """
- x: int
- y: int
- class Meta:
- ordering = ["x", "y"]
- def __call__(self, x: int, y: int) -> int:
- return x - y
- class TestFunctionsManager:
- def test_register_func(self):
- func = FunctionsManager.get_func("add")
- assert func(2, 3) == 5
- def test_register_class(self):
- func = FunctionsManager.get_func("subtract")
- assert func(2, 3) == -1
- def test_metaclass(self):
- func = FunctionsManager.get_func("multiply")
- assert func(2, 3) == 6
复制代码 参考
https://github.com/TencentBlueKing/bkflow-feel/blob/main/bkflow_feel/utils.py
到此这篇关于python实现一个通用的插件类的文章就介绍到这了,更多相关python 通用插件类内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作! |
|