python实现一个通用的插件类

2024-04-18 0 158
目录
  • 定义插件管理器
  • 定义元类
  • 定义元类的一个抽象派生类
  • 定义装饰器
  • 单元测试
  • 参考

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

定义插件管理器

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

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 通用插件类内容请搜索悠久资源网以前的文章或继续浏览下面的相关文章希望大家以后多多支持悠久资源网!

您可能感兴趣的文章:

  • 8个Python必备的PyCharm插件(附下载地址)
  • Python中表格插件Tabulate的用法小结
  • VsCode中超好用的8个python插件推荐
  • Python利用LyScript插件实现批量打开关闭进程
  • 使用Python开发游戏运行脚本成功调用大漠插件
  • Python 带你快速上手 Apache APISIX 插件开发

收藏 (0) 打赏

感谢您的支持,我会继续努力的!

打开微信/支付宝扫一扫,即可进行扫码打赏哦,分享从这里开始,精彩与您同在
点赞 (0)

悠久资源 Python python实现一个通用的插件类 https://www.u-9.cn/jiaoben/python/186939.html

常见问题

相关文章

发表评论
暂无评论
官方客服团队

为您解决烦忧 - 24小时在线 专业服务