python实现一个通用的插件类

w123938

目录

• 定义插件管理器
  
• 定义元类
  
• 定义元类的一个抽象派生类
  
• 定义装饰器
  
• 单元测试
  
• 参考
  

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

定义插件管理器

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

1. import abc
   
2. from functools import wraps
   
3. from typing import Callable, Dict
   
4. 
   
5. from pydantic import (
   
6.     BaseModel,
   
7.     validate_arguments,
   
8.     ValidationError as PydanticValidationError,
   
9. )
   
10. 
    
11. def import_string(dotted_path: str) -> Callable:
    
12.     """Import a dotted module path and return the attribute/class designated by the
    
13.     last name in the path. Raise ImportError if the import failed.
    
14. 
    
15.     Args:
    
16.         dotted_path: 字符串表示的模块类，module.class
    
17.     Returns:
    
18.         返回加载的模块中的对象
    
19.     """
    
20.     try:
    
21.         module_path, class_name = dotted_path.rsplit(".", 1)
    
22.     except ValueError:
    
23.         raise ImportError("{} doesn't look like a module path".format(dotted_path))
    
24. 
    
25.     module: ModuleType = import_module(module_path)
    
26. 
    
27.     try:
    
28.         # 返回模块中的类
    
29.         return getattr(module, class_name)
    
30.     except AttributeError:
    
31.         raise ImportError(
    
32.             'Module "{}" does not define a "{}" attribute/class'.format(
    
33.                 module_path, class_name
    
34.             )
    
35.         )
    
36. 
    
37. 
    
38. class FunctionsManager:
    
39.     """函数管理器 ."""
    
40.     # 存放注册的可执行对象
    
41.     __hub = {}  # type: ignore
    
42. 
    
43.     @classmethod
    
44.     def register_invocation_cls(cls, invocation_cls: InvocationMeta, name=None) -> None:
    
45.         if not name:
    
46.             func_name = invocation_cls.Meta.func_name
    
47.         else:
    
48.             func_name = name
    
49.         if not isinstance(func_name, str):
    
50.             raise ValueError(f"func_name {func_name} should be string")
    
51.         existed_invocation_cls = cls.__hub.get(func_name)
    
52.         if existed_invocation_cls:
    
53.             raise RuntimeError(
    
54.                 "func register error, {}'s func_name {} conflict with {}".format(
    
55.                     existed_invocation_cls, func_name, invocation_cls
    
56.                 )
    
57.             )
    
58. 
    
59.         # 存放类的实例
    
60.         cls.__hub[func_name] = invocation_cls()
    
61. 
    
62.     @classmethod
    
63.     def register_funcs(cls, func_dict) -> None:
    
64.         for func_name, func_obj in func_dict.items():
    
65.             if not isinstance(func_name, str):
    
66.                 raise ValueError(f"func_name {func_name} should be string")
    
67.             if func_name in cls.__hub:
    
68.                 raise ValueError(
    
69.                     "func register error, {}'s func_name {} conflict with {}".format(
    
70.                         func_obj, func_name, cls.__hub[func_name]
    
71.                     )
    
72.                 )
    
73.             if isinstance(func_obj, str):
    
74.                 func = import_string(func_obj)
    
75.             elif isinstance(func_obj, Callable):
    
76.                 func = func_obj
    
77.             else:
    
78.                 raise ValueError(
    
79.                     "func register error, {} is not be callable".format(
    
80.                         func_obj, func_name
    
81.                     )
    
82.                 )
    
83. 
    
84.             cls.__hub[func_name] = func
    
85. 
    
86.     @classmethod
    
87.     def clear(cls) -> None:
    
88.         """清空注册信息 ."""
    
89.         cls.__hub = {}
    
90. 
    
91.     @classmethod
    
92.     def all_funcs(cls) -> Dict:
    
93.         """获得所有的注册信息. """
    
94.         return cls.__hub
    
95. 
    
96.     @classmethod
    
97.     def get_func(cls, func_name: str) -> Callable:
    
98.         """获得注册的函数 ."""
    
99.         func_obj = cls.__hub.get(func_name)
    
100.         if not func_obj:
     
101.             raise ValueError("func object {} not found".format(func_name))
     
102.         return func_obj
     
103. 
     
104.     @classmethod
     
105.     def func_call(cls, func_name: str, *args, **kwargs):
     
106.         """根据函数名执行注册的函数 ."""
     
107.         func = cls.get_func(func_name)
     
108.         return func(*args, **kwargs)

复制代码

定义元类

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

1. class InvocationMeta(type):
   
2.     """
   
3.     Metaclass for function invocation
   
4.     """
   
5. 
   
6.     def __new__(cls, name, bases, dct):
   
7.         # ensure initialization is only performed for subclasses of Plugin
   
8.         parents = [b for b in bases if isinstance(b, InvocationMeta)]
   
9.         if not parents:
   
10.             return super().__new__(cls, name, bases, dct)
    
11. 
    
12.         new_cls = super().__new__(cls, name, bases, dct)
    
13. 
    
14.         # meta validation
    
15.         meta_obj = getattr(new_cls, "Meta", None)
    
16.         if not meta_obj:
    
17.             raise AttributeError("Meta class is required")
    
18. 
    
19.         func_name = getattr(meta_obj, "func_name", None)
    
20.         if not func_name:
    
21.             raise AttributeError("func_name is required in Meta")
    
22. 
    
23.         desc = getattr(meta_obj, "desc", None)
    
24.         if desc is not None and not isinstance(desc, str):
    
25.             raise AttributeError("desc in Meta should be str")
    
26. 
    
27.         # register func
    
28.         FunctionsManager.register_invocation_cls(new_cls)
    
29. 
    
30.         return new_cls

复制代码

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

支持参数验证。

1. class BaseInvocation(metaclass=InvocationMeta):
   
2.     """
   
3.     Base class for function invocation
   
4.     """
   
5. 
   
6.     class Inputs(BaseModel):
   
7.         """
   
8.         输入校验器
   
9.         """
   
10. 
    
11.         pass
    
12. 
    
13.     @validate_arguments  # type: ignore
    
14.     def __call__(self, *args, **kwargs):
    
15. 
    
16.         # 输入参数校验, 仅可能是 args 或 kwargs 之一
    
17.         try:
    
18.             params = {}
    
19.             if args:
    
20.                 inputs_meta = getattr(self.Inputs, "Meta", None)
    
21.                 inputs_ordering = getattr(inputs_meta, "ordering", None)
    
22.                 if isinstance(inputs_ordering, list):
    
23.                     if len(args) > len(inputs_ordering):
    
24.                         raise Exception(f"Too many arguments for inputs: {args}")
    
25.                     params = dict(zip(inputs_ordering, args))
    
26.             elif kwargs:
    
27.                 params = kwargs
    
28. 
    
29.             # 参数校验
    
30.             if params:
    
31.                 self.Inputs(**params)
    
32.         except PydanticValidationError as e:
    
33.             raise Exception(e)
    
34. 
    
35.         # 执行自定义业务逻辑
    
36.         return self.invoke(*args, **kwargs)
    
37. 
    
38.     @abc.abstractmethod
    
39.     def invoke(self, *args, **kwargs):
    
40.         """自定义业务逻辑 ."""
    
41.         raise NotImplementedError()

复制代码

定义装饰器

1. def register_class(name: str):
   
2.     def _register_class(cls: BaseInvocation):
   
3.         FunctionsManager.register_invocation_cls(cls, name=name)
   
4. 
   
5.         @wraps(cls)
   
6.         def wrapper():
   
7.             return cls()
   
8. 
   
9.         return wrapper
   
10. 
    
11.     return _register_class
    
12. 
    
13. 
    
14. def register_func(name: str):
    
15.     def _register_func(func: Callable):
    
16.         FunctionsManager.register_funcs({name: func})
    
17. 
    
18.         @wraps(func)
    
19.         def wrapper(*args, **kwargs):
    
20.             return func(*args, **kwargs)
    
21. 
    
22.         return wrapper
    
23. 
    
24.     return _register_func

复制代码

单元测试

1. from pydantic import BaseModel
   
2. 
   
3. from .register import FunctionsManager, register_func, register_class, BaseInvocation
   
4. 
   
5. 
   
6. @register_func("add")
   
7. def add(x: int, y: int) -> int:
   
8.     return x + y
   
9. 
   
10. 
    
11. class Add(BaseInvocation):
    
12.     class Meta:
    
13.         func_name = "multiply"
    
14. 
    
15.     class Inputs(BaseModel):
    
16.         """
    
17.         输入校验器
    
18.         """
    
19.         x: int
    
20.         y: int
    
21. 
    
22.         class Meta:
    
23.             ordering = ["x", "y"]
    
24. 
    
25.     def invoke(self, x: int, y: int) -> int:
    
26.         return x * y
    
27. 
    
28. 
    
29. @register_class("subtract")
    
30. class Subtract:
    
31.     class Inputs(BaseModel):
    
32.         """
    
33.         输入校验器
    
34.         """
    
35.         x: int
    
36.         y: int
    
37. 
    
38.         class Meta:
    
39.             ordering = ["x", "y"]
    
40. 
    
41.     def __call__(self, x: int, y: int) -> int:
    
42.         return x - y
    
43. 
    
44. 
    
45. class TestFunctionsManager:
    
46.     def test_register_func(self):
    
47.         func = FunctionsManager.get_func("add")
    
48.         assert func(2, 3) == 5
    
49. 
    
50.     def test_register_class(self):
    
51.         func = FunctionsManager.get_func("subtract")
    
52.         assert func(2, 3) == -1
    
53. 
    
54.     def test_metaclass(self):
    
55.         func = FunctionsManager.get_func("multiply")
    
56.         assert func(2, 3) == 6

复制代码

参考

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

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

大地回春666

感谢分享，受益匪浅！

万物有灵

我们一起努力，共同解决问题吧。

电竞番茄妹

让我们一起努力

爱染星空

我们一起努力，共同解决问题吧。

悠乐卡盟

顶一个，观点非常中肯！

明月小豆

说得太好了，完全同意！

小卒灬sky

非常感谢你的观点，让我受益良多！

萝卜头111315

我想了解更多

游戏大妹子呀

我完全同意你的观点