设为首页收藏本站
天天打卡

 找回密码
 立即注册
搜索
查看: 40|回复: 10

python实现一个通用的插件类

[复制链接]

2

主题

41

回帖

127

积分

注册会员

积分
127
发表于 2024-4-20 09:45:56 | 显示全部楼层 |阅读模式
目录

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

定义插件管理器

插件管理器用于注册、销毁、执行插件。
  1. import abc
  2. from functools import wraps
  3. from typing import Callable, Dict

  4. from pydantic import (
  5.     BaseModel,
  6.     validate_arguments,
  7.     ValidationError as PydanticValidationError,
  8. )

  9. def import_string(dotted_path: str) -> Callable:
  10.     """Import a dotted module path and return the attribute/class designated by the
  11.     last name in the path. Raise ImportError if the import failed.

  12.     Args:
  13.         dotted_path: 字符串表示的模块类,module.class
  14.     Returns:
  15.         返回加载的模块中的对象
  16.     """
  17.     try:
  18.         module_path, class_name = dotted_path.rsplit(".", 1)
  19.     except ValueError:
  20.         raise ImportError("{} doesn't look like a module path".format(dotted_path))

  21.     module: ModuleType = import_module(module_path)

  22.     try:
  23.         # 返回模块中的类
  24.         return getattr(module, class_name)
  25.     except AttributeError:
  26.         raise ImportError(
  27.             'Module "{}" does not define a "{}" attribute/class'.format(
  28.                 module_path, class_name
  29.             )
  30.         )


  31. class FunctionsManager:
  32.     """函数管理器 ."""
  33.     # 存放注册的可执行对象
  34.     __hub = {}  # type: ignore

  35.     @classmethod
  36.     def register_invocation_cls(cls, invocation_cls: InvocationMeta, name=None) -> None:
  37.         if not name:
  38.             func_name = invocation_cls.Meta.func_name
  39.         else:
  40.             func_name = name
  41.         if not isinstance(func_name, str):
  42.             raise ValueError(f"func_name {func_name} should be string")
  43.         existed_invocation_cls = cls.__hub.get(func_name)
  44.         if existed_invocation_cls:
  45.             raise RuntimeError(
  46.                 "func register error, {}'s func_name {} conflict with {}".format(
  47.                     existed_invocation_cls, func_name, invocation_cls
  48.                 )
  49.             )

  50.         # 存放类的实例
  51.         cls.__hub[func_name] = invocation_cls()

  52.     @classmethod
  53.     def register_funcs(cls, func_dict) -> None:
  54.         for func_name, func_obj in func_dict.items():
  55.             if not isinstance(func_name, str):
  56.                 raise ValueError(f"func_name {func_name} should be string")
  57.             if func_name in cls.__hub:
  58.                 raise ValueError(
  59.                     "func register error, {}'s func_name {} conflict with {}".format(
  60.                         func_obj, func_name, cls.__hub[func_name]
  61.                     )
  62.                 )
  63.             if isinstance(func_obj, str):
  64.                 func = import_string(func_obj)
  65.             elif isinstance(func_obj, Callable):
  66.                 func = func_obj
  67.             else:
  68.                 raise ValueError(
  69.                     "func register error, {} is not be callable".format(
  70.                         func_obj, func_name
  71.                     )
  72.                 )

  73.             cls.__hub[func_name] = func

  74.     @classmethod
  75.     def clear(cls) -> None:
  76.         """清空注册信息 ."""
  77.         cls.__hub = {}

  78.     @classmethod
  79.     def all_funcs(cls) -> Dict:
  80.         """获得所有的注册信息. """
  81.         return cls.__hub

  82.     @classmethod
  83.     def get_func(cls, func_name: str) -> Callable:
  84.         """获得注册的函数 ."""
  85.         func_obj = cls.__hub.get(func_name)
  86.         if not func_obj:
  87.             raise ValueError("func object {} not found".format(func_name))
  88.         return func_obj

  89.     @classmethod
  90.     def func_call(cls, func_name: str, *args, **kwargs):
  91.         """根据函数名执行注册的函数 ."""
  92.         func = cls.get_func(func_name)
  93.         return func(*args, **kwargs)
复制代码
定义元类

派生的类可自行注册到插件管理器。
  1. class InvocationMeta(type):
  2.     """
  3.     Metaclass for function invocation
  4.     """

  5.     def __new__(cls, name, bases, dct):
  6.         # ensure initialization is only performed for subclasses of Plugin
  7.         parents = [b for b in bases if isinstance(b, InvocationMeta)]
  8.         if not parents:
  9.             return super().__new__(cls, name, bases, dct)

  10.         new_cls = super().__new__(cls, name, bases, dct)

  11.         # meta validation
  12.         meta_obj = getattr(new_cls, "Meta", None)
  13.         if not meta_obj:
  14.             raise AttributeError("Meta class is required")

  15.         func_name = getattr(meta_obj, "func_name", None)
  16.         if not func_name:
  17.             raise AttributeError("func_name is required in Meta")

  18.         desc = getattr(meta_obj, "desc", None)
  19.         if desc is not None and not isinstance(desc, str):
  20.             raise AttributeError("desc in Meta should be str")

  21.         # register func
  22.         FunctionsManager.register_invocation_cls(new_cls)

  23.         return new_cls
复制代码
定义元类的一个抽象派生类

支持参数验证。
  1. class BaseInvocation(metaclass=InvocationMeta):
  2.     """
  3.     Base class for function invocation
  4.     """

  5.     class Inputs(BaseModel):
  6.         """
  7.         输入校验器
  8.         """

  9.         pass

  10.     @validate_arguments  # type: ignore
  11.     def __call__(self, *args, **kwargs):

  12.         # 输入参数校验, 仅可能是 args 或 kwargs 之一
  13.         try:
  14.             params = {}
  15.             if args:
  16.                 inputs_meta = getattr(self.Inputs, "Meta", None)
  17.                 inputs_ordering = getattr(inputs_meta, "ordering", None)
  18.                 if isinstance(inputs_ordering, list):
  19.                     if len(args) > len(inputs_ordering):
  20.                         raise Exception(f"Too many arguments for inputs: {args}")
  21.                     params = dict(zip(inputs_ordering, args))
  22.             elif kwargs:
  23.                 params = kwargs

  24.             # 参数校验
  25.             if params:
  26.                 self.Inputs(**params)
  27.         except PydanticValidationError as e:
  28.             raise Exception(e)

  29.         # 执行自定义业务逻辑
  30.         return self.invoke(*args, **kwargs)

  31.     @abc.abstractmethod
  32.     def invoke(self, *args, **kwargs):
  33.         """自定义业务逻辑 ."""
  34.         raise NotImplementedError()
复制代码
定义装饰器
  1. def register_class(name: str):
  2.     def _register_class(cls: BaseInvocation):
  3.         FunctionsManager.register_invocation_cls(cls, name=name)

  4.         @wraps(cls)
  5.         def wrapper():
  6.             return cls()

  7.         return wrapper

  8.     return _register_class


  9. def register_func(name: str):
  10.     def _register_func(func: Callable):
  11.         FunctionsManager.register_funcs({name: func})

  12.         @wraps(func)
  13.         def wrapper(*args, **kwargs):
  14.             return func(*args, **kwargs)

  15.         return wrapper

  16.     return _register_func
复制代码
单元测试
  1. from pydantic import BaseModel

  2. from .register import FunctionsManager, register_func, register_class, BaseInvocation


  3. @register_func("add")
  4. def add(x: int, y: int) -> int:
  5.     return x + y


  6. class Add(BaseInvocation):
  7.     class Meta:
  8.         func_name = "multiply"

  9.     class Inputs(BaseModel):
  10.         """
  11.         输入校验器
  12.         """
  13.         x: int
  14.         y: int

  15.         class Meta:
  16.             ordering = ["x", "y"]

  17.     def invoke(self, x: int, y: int) -> int:
  18.         return x * y


  19. @register_class("subtract")
  20. class Subtract:
  21.     class Inputs(BaseModel):
  22.         """
  23.         输入校验器
  24.         """
  25.         x: int
  26.         y: int

  27.         class Meta:
  28.             ordering = ["x", "y"]

  29.     def __call__(self, x: int, y: int) -> int:
  30.         return x - y


  31. class TestFunctionsManager:
  32.     def test_register_func(self):
  33.         func = FunctionsManager.get_func("add")
  34.         assert func(2, 3) == 5

  35.     def test_register_class(self):
  36.         func = FunctionsManager.get_func("subtract")
  37.         assert func(2, 3) == -1

  38.     def test_metaclass(self):
  39.         func = FunctionsManager.get_func("multiply")
  40.         assert func(2, 3) == 6
复制代码
参考

https://github.com/TencentBlueKing/bkflow-feel/blob/main/bkflow_feel/utils.py
到此这篇关于python实现一个通用的插件类的文章就介绍到这了,更多相关python 通用插件类内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!

3

主题

47

回帖

129

积分

注册会员

积分
129
发表于 2024-4-23 13:51:48 | 显示全部楼层
感谢分享,受益匪浅!

0

主题

61

回帖

123

积分

注册会员

积分
123
发表于 2024-5-1 22:58:30 | 显示全部楼层
我们一起努力,共同解决问题吧。

1

主题

42

回帖

108

积分

注册会员

积分
108
发表于 2024-5-19 18:36:11 | 显示全部楼层
让我们一起努力

1

主题

32

回帖

88

积分

注册会员

积分
88
发表于 2024-6-8 14:19:35 | 显示全部楼层
我们一起努力,共同解决问题吧。

2

主题

43

回帖

130

积分

注册会员

积分
130
发表于 2024-6-11 17:37:31 | 显示全部楼层
顶一个,观点非常中肯!

1

主题

50

回帖

122

积分

注册会员

积分
122
发表于 2024-6-14 15:23:51 | 显示全部楼层
说得太好了,完全同意!
  • 打卡等级:无名新人
  • 打卡总天数:2
  • 打卡月天数:0
  • 打卡总奖励:43
  • 最近打卡:2024-08-26 11:26:22

8

主题

54

回帖

285

积分

中级会员

积分
285
发表于 2024-8-16 08:50:15 | 显示全部楼层
非常感谢你的观点,让我受益良多!

2

主题

44

回帖

132

积分

注册会员

积分
132
发表于 2024-9-26 16:06:26 | 显示全部楼层
我想了解更多

4

主题

50

回帖

190

积分

注册会员

积分
190
发表于 2024-10-19 19:51:38 | 显示全部楼层
我完全同意你的观点
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

手机版|小黑屋|爱云论坛 - d.taiji888.cn - 技术学习 免费资源分享 ( 蜀ICP备2022010826号 )|天天打卡

GMT+8, 2024-11-15 05:34 , Processed in 0.087976 second(s), 26 queries .

Powered by i云网络 Licensed

© 2023-2028 正版授权

快速回复 返回顶部 返回列表