测试开发技术网站
博客
设计
设计
开发
Python
测试
unittest
运维
Linux基础应用
CI/CD
CI/CD
数据库
数据库
云计算
云计算
云原生
云原生
爬虫
爬虫
数据分析
数据分析
人工智能
人工智能
登录
注册
Pluggy源码解读----register注册插件源码解析
收藏本文
作者:redrose2100 类别: 日期:2022-12-10 09:42:01 阅读:1277 次 消耗积分:0 分
[TOC]  首先看应用中注册的代码如下一行,即调用register函数,传入实现接口的插件类。 ```python pm.register(Plugin_1()) ``` register方法定义如下,这里首先是获取插件的名称,通过register函数的定义可以看出,在注册插件的时候,可以通过name形参指定插件的名称。如果没有指定则通过get_canonical_name方法获取。 ```python def register(self, plugin, name=None): """Register a plugin and return its canonical name or ``None`` if the name is blocked from registering. Raise a :py:class:`ValueError` if the plugin is already registered.""" plugin_name = name or self.get_canonical_name(plugin) if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers: if self._name2plugin.get(plugin_name, -1) is None: return # blocked plugin, return None to indicate no registration raise ValueError( "Plugin already registered: %s=%s\n%s" % (plugin_name, plugin, self._name2plugin) ) # XXX if an error happens we should make sure no state has been # changed at point of return self._name2plugin[plugin_name] = plugin # register matching hook implementations of the plugin self._plugin2hookcallers[plugin] = hookcallers = [] for name in dir(plugin): hookimpl_opts = self.parse_hookimpl_opts(plugin, name) if hookimpl_opts is not None: normalize_hookimpl_opts(hookimpl_opts) method = getattr(plugin, name) hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts) name = hookimpl_opts.get("specname") or name hook = getattr(self.hook, name, None) if hook is None: hook = _HookCaller(name, self._hookexec) setattr(self.hook, name, hook) elif hook.has_spec(): self._verify_hook(hook, hookimpl) hook._maybe_apply_history(hookimpl) hook._add_hookimpl(hookimpl) hookcallers.append(hook) return plugin_name ``` get_canonical_name方法的定义如下,可以看出这里首先获取插件类的__name__属性值,如果没有此属性,则返回插件类的id作为插件的名称。 ```python def get_canonical_name(self, plugin): """Return canonical name for a plugin object. Note that a plugin may be registered under a different name which was specified by the caller of :py:meth:`register(plugin, name) <.PluginManager.register>`. To obtain the name of an registered plugin use :py:meth:`get_name(plugin) <.PluginManager.get_name>` instead.""" return getattr(plugin, "__name__", None) or str(id(plugin)) ``` 回到register方法,接下来就是做了一个判断,如果插件名称已经存在_name2plugin属性中或者插件类对象已经存在_plugin2hookcallers属性中了,说明插件已经注册过了,则报错提醒。如果没有注册,如下一行代码可以看到_name2plugin属性果然是存放插件名称和插件类对象的一个属性。 ```python self._name2plugin[plugin_name] = plugin ``` 而接下来一行代码则是说明plugin2hookcallers属性时存放插件和hookcallers映射关系的属性,因为一个插件完全可以存在多个hookcallers,因此这里初始化为一个空的列表。 ```python self._plugin2hookcallers[plugin] = hookcallers = [] ``` 紧接下来的for循环部分则和add_hookspecs中的解析接口类属性的代码是类似的,即对HookimplMarker类的所有方法和属性进行循环遍历,然后针对每一个方法或属性首先解析hookimpl_opts的值,是通过parse_hookimpl_opts方法获得的,而parse_hookimpl_opts的定义代码如下,即同样还是看此方法或属性是否被hookimpl装饰器修饰,如果没有被修饰说明不是要找的方法,则直接返回进入下一个循环,如果是则再获取myproject_impl属性,同样通过前面对HookimplMarker类的初始化代码的分析得知,myproject_impl属性的值也是一个字典,而且此字典包含hookwrapper、optionalhook、tryfirst、trylast、specname五个key,即这里res里存的就是由这五个key组成的字典。 ```python def parse_hookimpl_opts(self, plugin, name): method = getattr(plugin, name) if not inspect.isroutine(method): return try: res = getattr(method, self.project_name + "_impl", None) except Exception: res = {} if res is not None and not isinstance(res, dict): # false positive res = None return res ``` 再次回到register方法,当hookimpl_opts不为空时,即此时的方法是被hookimpl装饰的亦即是我们所要寻找的myhook实现方法。然后这里又对hookimpl_opts进行了一次规范化处理,即normalize_hookimpl_opts函数,实现代码如下: ```python def normalize_hookimpl_opts(opts): opts.setdefault("tryfirst", False) opts.setdefault("trylast", False) opts.setdefault("hookwrapper", False) opts.setdefault("optionalhook", False) opts.setdefault("specname", None) ``` 接下来method即为myhook方法对象,hookimpl为HookImpl类的一个实例对象,HookImpl类的代码如下,可以看出,这里HookImpl类和HookSpec类是类似的,即主要用于存放插件类的属性的,比如这里就存放了插件类中的方法、方法参数、插件、插件名称以及修饰的参数等,这里的\_\_repr\_\_方法是python语言中的常用的一个魔法函数,用于打印对象时在控制台显示的内容。 ```python class HookImpl: def __init__(self, plugin, plugin_name, function, hook_impl_opts): self.function = function self.argnames, self.kwargnames = varnames(self.function) self.plugin = plugin self.opts = hook_impl_opts self.plugin_name = plugin_name self.__dict__.update(hook_impl_opts) def __repr__(self): return f"
" ``` 回到register方法,接下来name的取值首先看hookimpl_opts,如果hookimpl_opts字典中有specname则取其值,否则则直接去方法名,这里即myhook,然后获取hook,即hook从当前类的self.hook对象中获取myhook的值,由于在分析add_hookspecs的时候,曾经分析到,self.hook中增加了名为myhook的属性,myhook属性的值为add_hookspecs方法中的hc即_HookCaller类对象,因此这里hook即可以获取到add_hookspecs中的hc。然后通过_add_hookimpl方法继续对hook增加属性配置。 ```python hook._add_hookimpl(hookimpl) ``` 如下,_add_hookimpl方法实现如下,这里可以看到,如果hookimpl的trylast为True,则会在方法列表中将当前的hookimpl插入到第一个,这就是为什么在插件定义中如果使用了trylast=True事,插件会最后一个执行,因为这些方法是先进后出的队列,同样如果设置了tryfirst=True,则会将当前hookimpl使用append插入到最后,这样最后一个就是第一个出队列执行了。此外这里还有hookimpl的hookwrapper是否为True,若为True,则表示此方法中有yield关键字,则此时将hookimpl放入_wrappers属性中,而若为False,则直接将hookimpl放入_nonwrappers属性中。 ```python def _add_hookimpl(self, hookimpl): """Add an implementation to the callback chain.""" if hookimpl.hookwrapper: methods = self._wrappers else: methods = self._nonwrappers if hookimpl.trylast: methods.insert(0, hookimpl) elif hookimpl.tryfirst: methods.append(hookimpl) else: # find last non-tryfirst method i = len(methods) - 1 while i >= 0 and methods[i].tryfirst: i -= 1 methods.insert(i + 1, hookimpl) ``` 最后将hook放入_plugin2hookcallers属性的plugin类对象作为key的值中。 至此,register方法就解析完成了。
始终坚持开源开放共享精神,同时感谢您的充电鼓励和支持!
版权所有,转载本站文章请注明出处:redrose2100, http://blog.redrose2100.com/article/493
上一篇:
Pluggy源码解读----add_hookspecs增加自定义的接口类
下一篇:
Pluggy源码解读----hook函数调用执行过程分析
搜索
个人成就
出版书籍
《Pytest企业级应用实战》
测试开发技术全栈公众号
测试开发技术全栈公众号
DevOps技术交流微信群
加微信邀请进群
常用网站链接
开源软件洞察
云原生技术栈全景图
Python语言官方文档
Golang官方文档
Docker官方文档
Jenkins中文用户手册
Scrapy官方文档
VUE官方文档
Harbor官方文档
openQA官方文档
云原生开源社区
开源中国
Kubernetes中文文档
Markdown语法官方教程
Kubernetes中文社区
Kubersphere官方文档
BootStrap中文网站
JavaScript中文网
NumPy官方文档
Pandas官方文档
GitLink确实开源网站
数据库排名网站
编程语言排名网站
SEO综合查询网站
数学加减法练习自动生成网站
Kickstart Generator
文章分类
最新文章
最多阅读
特别推荐
×
Close
登录
注册
找回密码
登录邮箱:
登录密码:
图片验证码:
注册邮箱:
注册密码:
邮箱验证码:
发送邮件
注册邮箱:
新的密码:
邮箱验证码:
发送邮件