Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1""" 

2Internal hook annotation, representation and calling machinery. 

3""" 

4import inspect 

5import sys 

6import warnings 

7from .callers import _legacymulticall, _multicall 

8 

9 

10class HookspecMarker(object): 

11 """ Decorator helper class for marking functions as hook specifications. 

12 

13 You can instantiate it with a project_name to get a decorator. 

14 Calling PluginManager.add_hookspecs later will discover all marked functions 

15 if the PluginManager uses the same project_name. 

16 """ 

17 

18 def __init__(self, project_name): 

19 self.project_name = project_name 

20 

21 def __call__( 

22 self, function=None, firstresult=False, historic=False, warn_on_impl=None 

23 ): 

24 """ if passed a function, directly sets attributes on the function 

25 which will make it discoverable to add_hookspecs(). If passed no 

26 function, returns a decorator which can be applied to a function 

27 later using the attributes supplied. 

28 

29 If firstresult is True the 1:N hook call (N being the number of registered 

30 hook implementation functions) will stop at I<=N when the I'th function 

31 returns a non-None result. 

32 

33 If historic is True calls to a hook will be memorized and replayed 

34 on later registered plugins. 

35 

36 """ 

37 

38 def setattr_hookspec_opts(func): 

39 if historic and firstresult: 

40 raise ValueError("cannot have a historic firstresult hook") 

41 setattr( 

42 func, 

43 self.project_name + "_spec", 

44 dict( 

45 firstresult=firstresult, 

46 historic=historic, 

47 warn_on_impl=warn_on_impl, 

48 ), 

49 ) 

50 return func 

51 

52 if function is not None: 

53 return setattr_hookspec_opts(function) 

54 else: 

55 return setattr_hookspec_opts 

56 

57 

58class HookimplMarker(object): 

59 """ Decorator helper class for marking functions as hook implementations. 

60 

61 You can instantiate with a project_name to get a decorator. 

62 Calling PluginManager.register later will discover all marked functions 

63 if the PluginManager uses the same project_name. 

64 """ 

65 

66 def __init__(self, project_name): 

67 self.project_name = project_name 

68 

69 def __call__( 

70 self, 

71 function=None, 

72 hookwrapper=False, 

73 optionalhook=False, 

74 tryfirst=False, 

75 trylast=False, 

76 ): 

77 

78 """ if passed a function, directly sets attributes on the function 

79 which will make it discoverable to register(). If passed no function, 

80 returns a decorator which can be applied to a function later using 

81 the attributes supplied. 

82 

83 If optionalhook is True a missing matching hook specification will not result 

84 in an error (by default it is an error if no matching spec is found). 

85 

86 If tryfirst is True this hook implementation will run as early as possible 

87 in the chain of N hook implementations for a specfication. 

88 

89 If trylast is True this hook implementation will run as late as possible 

90 in the chain of N hook implementations. 

91 

92 If hookwrapper is True the hook implementations needs to execute exactly 

93 one "yield". The code before the yield is run early before any non-hookwrapper 

94 function is run. The code after the yield is run after all non-hookwrapper 

95 function have run. The yield receives a ``_Result`` object representing 

96 the exception or result outcome of the inner calls (including other hookwrapper 

97 calls). 

98 

99 """ 

100 

101 def setattr_hookimpl_opts(func): 

102 setattr( 

103 func, 

104 self.project_name + "_impl", 

105 dict( 

106 hookwrapper=hookwrapper, 

107 optionalhook=optionalhook, 

108 tryfirst=tryfirst, 

109 trylast=trylast, 

110 ), 

111 ) 

112 return func 

113 

114 if function is None: 

115 return setattr_hookimpl_opts 

116 else: 

117 return setattr_hookimpl_opts(function) 

118 

119 

120def normalize_hookimpl_opts(opts): 

121 opts.setdefault("tryfirst", False) 

122 opts.setdefault("trylast", False) 

123 opts.setdefault("hookwrapper", False) 

124 opts.setdefault("optionalhook", False) 

125 

126 

127if hasattr(inspect, "getfullargspec"): 

128 

129 def _getargspec(func): 

130 return inspect.getfullargspec(func) 

131 

132 

133else: 

134 

135 def _getargspec(func): 

136 return inspect.getargspec(func) 

137 

138 

139_PYPY3 = hasattr(sys, "pypy_version_info") and sys.version_info.major == 3 

140 

141 

142def varnames(func): 

143 """Return tuple of positional and keywrord argument names for a function, 

144 method, class or callable. 

145 

146 In case of a class, its ``__init__`` method is considered. 

147 For methods the ``self`` parameter is not included. 

148 """ 

149 cache = getattr(func, "__dict__", {}) 

150 try: 

151 return cache["_varnames"] 

152 except KeyError: 

153 pass 

154 

155 if inspect.isclass(func): 

156 try: 

157 func = func.__init__ 

158 except AttributeError: 

159 return (), () 

160 elif not inspect.isroutine(func): # callable object? 

161 try: 

162 func = getattr(func, "__call__", func) 

163 except Exception: 

164 return () 

165 

166 try: # func MUST be a function or method here or we won't parse any args 

167 spec = _getargspec(func) 

168 except TypeError: 

169 return (), () 

170 

171 args, defaults = tuple(spec.args), spec.defaults 

172 if defaults: 

173 index = -len(defaults) 

174 args, defaults = args[:index], tuple(args[index:]) 

175 else: 

176 defaults = () 

177 

178 # strip any implicit instance arg 

179 # pypy3 uses "obj" instead of "self" for default dunder methods 

180 implicit_names = ("self",) if not _PYPY3 else ("self", "obj") 

181 if args: 

182 if inspect.ismethod(func) or ( 

183 "." in getattr(func, "__qualname__", ()) and args[0] in implicit_names 

184 ): 

185 args = args[1:] 

186 

187 try: 

188 cache["_varnames"] = args, defaults 

189 except TypeError: 

190 pass 

191 return args, defaults 

192 

193 

194class _HookRelay(object): 

195 """ hook holder object for performing 1:N hook calls where N is the number 

196 of registered plugins. 

197 

198 """ 

199 

200 def __init__(self, trace): 

201 self._trace = trace 

202 

203 

204class _HookCaller(object): 

205 def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None): 

206 self.name = name 

207 self._wrappers = [] 

208 self._nonwrappers = [] 

209 self._hookexec = hook_execute 

210 self.argnames = None 

211 self.kwargnames = None 

212 self.multicall = _multicall 

213 self.spec = None 

214 if specmodule_or_class is not None: 

215 assert spec_opts is not None 

216 self.set_specification(specmodule_or_class, spec_opts) 

217 

218 def has_spec(self): 

219 return self.spec is not None 

220 

221 def set_specification(self, specmodule_or_class, spec_opts): 

222 assert not self.has_spec() 

223 self.spec = HookSpec(specmodule_or_class, self.name, spec_opts) 

224 if spec_opts.get("historic"): 

225 self._call_history = [] 

226 

227 def is_historic(self): 

228 return hasattr(self, "_call_history") 

229 

230 def _remove_plugin(self, plugin): 

231 def remove(wrappers): 

232 for i, method in enumerate(wrappers): 

233 if method.plugin == plugin: 

234 del wrappers[i] 

235 return True 

236 

237 if remove(self._wrappers) is None: 

238 if remove(self._nonwrappers) is None: 

239 raise ValueError("plugin %r not found" % (plugin,)) 

240 

241 def get_hookimpls(self): 

242 # Order is important for _hookexec 

243 return self._nonwrappers + self._wrappers 

244 

245 def _add_hookimpl(self, hookimpl): 

246 """Add an implementation to the callback chain. 

247 """ 

248 if hookimpl.hookwrapper: 

249 methods = self._wrappers 

250 else: 

251 methods = self._nonwrappers 

252 

253 if hookimpl.trylast: 

254 methods.insert(0, hookimpl) 

255 elif hookimpl.tryfirst: 

256 methods.append(hookimpl) 

257 else: 

258 # find last non-tryfirst method 

259 i = len(methods) - 1 

260 while i >= 0 and methods[i].tryfirst: 

261 i -= 1 

262 methods.insert(i + 1, hookimpl) 

263 

264 if "__multicall__" in hookimpl.argnames: 

265 warnings.warn( 

266 "Support for __multicall__ is now deprecated and will be" 

267 "removed in an upcoming release.", 

268 DeprecationWarning, 

269 ) 

270 self.multicall = _legacymulticall 

271 

272 def __repr__(self): 

273 return "<_HookCaller %r>" % (self.name,) 

274 

275 def __call__(self, *args, **kwargs): 

276 if args: 

277 raise TypeError("hook calling supports only keyword arguments") 

278 assert not self.is_historic() 

279 if self.spec and self.spec.argnames: 

280 notincall = ( 

281 set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys()) 

282 ) 

283 if notincall: 

284 warnings.warn( 

285 "Argument(s) {} which are declared in the hookspec " 

286 "can not be found in this hook call".format(tuple(notincall)), 

287 stacklevel=2, 

288 ) 

289 return self._hookexec(self, self.get_hookimpls(), kwargs) 

290 

291 def call_historic(self, result_callback=None, kwargs=None, proc=None): 

292 """Call the hook with given ``kwargs`` for all registered plugins and 

293 for all plugins which will be registered afterwards. 

294 

295 If ``result_callback`` is not ``None`` it will be called for for each 

296 non-None result obtained from a hook implementation. 

297 

298 .. note:: 

299 The ``proc`` argument is now deprecated. 

300 """ 

301 if proc is not None: 

302 warnings.warn( 

303 "Support for `proc` argument is now deprecated and will be" 

304 "removed in an upcoming release.", 

305 DeprecationWarning, 

306 ) 

307 result_callback = proc 

308 

309 self._call_history.append((kwargs or {}, result_callback)) 

310 # historizing hooks don't return results 

311 res = self._hookexec(self, self.get_hookimpls(), kwargs) 

312 if result_callback is None: 

313 return 

314 # XXX: remember firstresult isn't compat with historic 

315 for x in res or []: 

316 result_callback(x) 

317 

318 def call_extra(self, methods, kwargs): 

319 """ Call the hook with some additional temporarily participating 

320 methods using the specified kwargs as call parameters. """ 

321 old = list(self._nonwrappers), list(self._wrappers) 

322 for method in methods: 

323 opts = dict(hookwrapper=False, trylast=False, tryfirst=False) 

324 hookimpl = HookImpl(None, "<temp>", method, opts) 

325 self._add_hookimpl(hookimpl) 

326 try: 

327 return self(**kwargs) 

328 finally: 

329 self._nonwrappers, self._wrappers = old 

330 

331 def _maybe_apply_history(self, method): 

332 """Apply call history to a new hookimpl if it is marked as historic. 

333 """ 

334 if self.is_historic(): 

335 for kwargs, result_callback in self._call_history: 

336 res = self._hookexec(self, [method], kwargs) 

337 if res and result_callback is not None: 

338 result_callback(res[0]) 

339 

340 

341class HookImpl(object): 

342 def __init__(self, plugin, plugin_name, function, hook_impl_opts): 

343 self.function = function 

344 self.argnames, self.kwargnames = varnames(self.function) 

345 self.plugin = plugin 

346 self.opts = hook_impl_opts 

347 self.plugin_name = plugin_name 

348 self.__dict__.update(hook_impl_opts) 

349 

350 def __repr__(self): 

351 return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin) 

352 

353 

354class HookSpec(object): 

355 def __init__(self, namespace, name, opts): 

356 self.namespace = namespace 

357 self.function = function = getattr(namespace, name) 

358 self.name = name 

359 self.argnames, self.kwargnames = varnames(function) 

360 self.opts = opts 

361 self.argnames = ["__multicall__"] + list(self.argnames) 

362 self.warn_on_impl = opts.get("warn_on_impl")