1 # ######################### LICENSE ############################ # 2 3 # Copyright (c) 2005-2025, Michele Simionato 4 # All rights reserved. 5 6 # Redistribution and use in source and binary forms, with or without 7 # modification, are permitted provided that the following conditions are 8 # met: 9 10 # Redistributions of source code must retain the above copyright 11 # notice, this list of conditions and the following disclaimer. 12 # Redistributions in bytecode form must reproduce the above copyright 13 # notice, this list of conditions and the following disclaimer in 14 # the documentation and/or other materials provided with the 15 # distribution. 16 17 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 18 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 19 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 20 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 21 # HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 22 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 23 # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS 24 # OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND 25 # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR 26 # TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE 27 # USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH 28 # DAMAGE. 29 30 """ 31 Decorator module, see 32 https://github.com/micheles/decorator/blob/master/docs/documentation.md 33 for the documentation. 34 """ 35 import re 36 import sys 37 import inspect 38 import operator 39 import itertools 40 import functools 41 from contextlib import _GeneratorContextManager 42 from inspect import getfullargspec, iscoroutinefunction, isgeneratorfunction 43 44 __version__ = '5.2.1' 45 46 DEF = re.compile(r'\s*def\s*([_\w][_\w\d]*)\s*\(') 47 POS = inspect.Parameter.POSITIONAL_OR_KEYWORD 48 EMPTY = inspect.Parameter.empty 49 50 51 # this is not used anymore in the core, but kept for backward compatibility 52 class FunctionMaker(object): 53 """ 54 An object with the ability to create functions with a given signature. 55 It has attributes name, doc, module, signature, defaults, dict and 56 methods update and make. 57 """ 58 59 # Atomic get-and-increment provided by the GIL 60 _compile_count = itertools.count() 61 62 # make pylint happy 63 args = varargs = varkw = defaults = kwonlyargs = kwonlydefaults = () 64 65 def __init__(self, func=None, name=None, signature=None, 66 defaults=None, doc=None, module=None, funcdict=None): 67 self.shortsignature = signature 68 if func: 69 # func can be a class or a callable, but not an instance method 70 self.name = func.__name__ 71 if self.name == '<lambda>': # small hack for lambda functions 72 self.name = '_lambda_' 73 self.doc = func.__doc__ 74 self.module = func.__module__ 75 if inspect.isroutine(func) or isinstance(func, functools.partial): 76 argspec = getfullargspec(func) 77 self.annotations = getattr(func, '__annotations__', {}) 78 for a in ('args', 'varargs', 'varkw', 'defaults', 'kwonlyargs', 79 'kwonlydefaults'): 80 setattr(self, a, getattr(argspec, a)) 81 for i, arg in enumerate(self.args): 82 setattr(self, 'arg%d' % i, arg) 83 allargs = list(self.args) 84 allshortargs = list(self.args) 85 if self.varargs: 86 allargs.append('*' + self.varargs) 87 allshortargs.append('*' + self.varargs) 88 elif self.kwonlyargs: 89 allargs.append('*') # single star syntax 90 for a in self.kwonlyargs: 91 allargs.append('%s=None' % a) 92 allshortargs.append('%s=%s' % (a, a)) 93 if self.varkw: 94 allargs.append('**' + self.varkw) 95 allshortargs.append('**' + self.varkw) 96 self.signature = ', '.join(allargs) 97 self.shortsignature = ', '.join(allshortargs) 98 self.dict = func.__dict__.copy() 99 # func=None happens when decorating a caller 100 if name: 101 self.name = name 102 if signature is not None: 103 self.signature = signature 104 if defaults: 105 self.defaults = defaults 106 if doc: 107 self.doc = doc 108 if module: 109 self.module = module 110 if funcdict: 111 self.dict = funcdict 112 # check existence required attributes 113 assert hasattr(self, 'name') 114 if not hasattr(self, 'signature'): 115 raise TypeError('You are decorating a non function: %s' % func) 116 117 def update(self, func, **kw): 118 """ 119 Update the signature of func with the data in self 120 """ 121 func.__name__ = self.name 122 func.__doc__ = getattr(self, 'doc', None) 123 func.__dict__ = getattr(self, 'dict', {}) 124 func.__defaults__ = self.defaults 125 func.__kwdefaults__ = self.kwonlydefaults or None 126 func.__annotations__ = getattr(self, 'annotations', None) 127 try: 128 frame = sys._getframe(3) 129 except AttributeError: # for IronPython and similar implementations 130 callermodule = '?' 131 else: 132 callermodule = frame.f_globals.get('__name__', '?') 133 func.__module__ = getattr(self, 'module', callermodule) 134 func.__dict__.update(kw) 135 136 def make(self, src_templ, evaldict=None, addsource=False, **attrs): 137 """ 138 Make a new function from a given template and update the signature 139 """ 140 src = src_templ % vars(self) # expand name and signature 141 evaldict = evaldict or {} 142 mo = DEF.search(src) 143 if mo is None: 144 raise SyntaxError('not a valid function template\n%s' % src) 145 name = mo.group(1) # extract the function name 146 names = set([name] + [arg.strip(' *') for arg in 147 self.shortsignature.split(',')]) 148 for n in names: 149 if n in ('_func_', '_call_'): 150 raise NameError('%s is overridden in\n%s' % (n, src)) 151 152 if not src.endswith('\n'): # add a newline for old Pythons 153 src += '\n' 154 155 # Ensure each generated function has a unique filename for profilers 156 # (such as cProfile) that depend on the tuple of (<filename>, 157 # <definition line>, <function name>) being unique. 158 filename = '<decorator-gen-%d>' % next(self._compile_count) 159 try: 160 code = compile(src, filename, 'single') 161 exec(code, evaldict) 162 except Exception: 163 print('Error in generated code:', file=sys.stderr) 164 print(src, file=sys.stderr) 165 raise 166 func = evaldict[name] 167 if addsource: 168 attrs['__source__'] = src 169 self.update(func, **attrs) 170 return func 171 172 @classmethod 173 def create(cls, obj, body, evaldict, defaults=None, 174 doc=None, module=None, addsource=True, **attrs): 175 """ 176 Create a function from the strings name, signature and body. 177 evaldict is the evaluation dictionary. If addsource is true an 178 attribute __source__ is added to the result. The attributes attrs 179 are added, if any. 180 """ 181 if isinstance(obj, str): # "name(signature)" 182 name, rest = obj.strip().split('(', 1) 183 signature = rest[:-1] # strip a right parens 184 func = None 185 else: # a function 186 name = None 187 signature = None 188 func = obj 189 self = cls(func, name, signature, defaults, doc, module) 190 ibody = '\n'.join(' ' + line for line in body.splitlines()) 191 caller = evaldict.get('_call_') # when called from `decorate` 192 if caller and iscoroutinefunction(caller): 193 body = ('async def %(name)s(%(signature)s):\n' + ibody).replace( 194 'return', 'return await') 195 else: 196 body = 'def %(name)s(%(signature)s):\n' + ibody 197 return self.make(body, evaldict, addsource, **attrs) 198 199 200 def fix(args, kwargs, sig): 201 """ 202 Fix args and kwargs to be consistent with the signature 203 """ 204 ba = sig.bind(*args, **kwargs) 205 ba.apply_defaults() # needed for test_dan_schult 206 return ba.args, ba.kwargs 207 208 209 def decorate(func, caller, extras=(), kwsyntax=False): 210 """ 211 Decorates a function/generator/coroutine using a caller. 212 If kwsyntax is True calling the decorated functions with keyword 213 syntax will pass the named arguments inside the ``kw`` dictionary, 214 even if such argument are positional, similarly to what functools.wraps 215 does. By default kwsyntax is False and the the arguments are untouched. 216 """ 217 sig = inspect.signature(func) 218 if isinstance(func, functools.partial): 219 func = functools.update_wrapper(func, func.func) 220 if iscoroutinefunction(caller): 221 async def fun(*args, **kw): 222 if not kwsyntax: 223 args, kw = fix(args, kw, sig) 224 return await caller(func, *(extras + args), **kw) 225 elif isgeneratorfunction(caller): 226 def fun(*args, **kw): 227 if not kwsyntax: 228 args, kw = fix(args, kw, sig) 229 for res in caller(func, *(extras + args), **kw): 230 yield res 231 else: 232 def fun(*args, **kw): 233 if not kwsyntax: 234 args, kw = fix(args, kw, sig) 235 return caller(func, *(extras + args), **kw) 236 237 fun.__name__ = func.__name__ 238 fun.__doc__ = func.__doc__ 239 fun.__wrapped__ = func 240 fun.__signature__ = sig 241 fun.__qualname__ = func.__qualname__ 242 # builtin functions like defaultdict.__setitem__ lack many attributes 243 try: 244 fun.__defaults__ = func.__defaults__ 245 except AttributeError: 246 pass 247 try: 248 fun.__kwdefaults__ = func.__kwdefaults__ 249 except AttributeError: 250 pass 251 try: 252 fun.__annotations__ = func.__annotations__ 253 except AttributeError: 254 pass 255 try: 256 fun.__module__ = func.__module__ 257 except AttributeError: 258 pass 259 try: 260 fun.__name__ = func.__name__ 261 except AttributeError: # happens with old versions of numpy.vectorize 262 func.__name__ == 'noname' 263 try: 264 fun.__dict__.update(func.__dict__) 265 except AttributeError: 266 pass 267 return fun 268 269 270 def decoratorx(caller): 271 """ 272 A version of "decorator" implemented via "exec" and not via the 273 Signature object. Use this if you are want to preserve the `.__code__` 274 object properties (https://github.com/micheles/decorator/issues/129). 275 """ 276 def dec(func): 277 return FunctionMaker.create( 278 func, 279 "return _call_(_func_, %(shortsignature)s)", 280 dict(_call_=caller, _func_=func), 281 __wrapped__=func, __qualname__=func.__qualname__) 282 return dec 283 284 285 def decorator(caller, _func=None, kwsyntax=False): 286 """ 287 decorator(caller) converts a caller function into a decorator 288 """ 289 if _func is not None: # return a decorated function 290 # this is obsolete behavior; you should use decorate instead 291 return decorate(_func, caller, (), kwsyntax) 292 # else return a decorator function 293 sig = inspect.signature(caller) 294 dec_params = [p for p in sig.parameters.values() if p.kind is POS] 295 296 def dec(func=None, *args, **kw): 297 na = len(args) + 1 298 extras = args + tuple(kw.get(p.name, p.default) 299 for p in dec_params[na:] 300 if p.default is not EMPTY) 301 if func is None: 302 return lambda func: decorate(func, caller, extras, kwsyntax) 303 else: 304 return decorate(func, caller, extras, kwsyntax) 305 dec.__signature__ = sig.replace(parameters=dec_params) 306 dec.__name__ = caller.__name__ 307 dec.__doc__ = caller.__doc__ 308 dec.__wrapped__ = caller 309 dec.__qualname__ = caller.__qualname__ 310 dec.__kwdefaults__ = getattr(caller, '__kwdefaults__', None) 311 dec.__dict__.update(caller.__dict__) 312 return dec 313 314 315 # ####################### contextmanager ####################### # 316 317 318 class ContextManager(_GeneratorContextManager): 319 def __init__(self, g, *a, **k): 320 _GeneratorContextManager.__init__(self, g, a, k) 321 322 def __call__(self, func): 323 def caller(f, *a, **k): 324 with self.__class__(self.func, *self.args, **self.kwds): 325 return f(*a, **k) 326 return decorate(func, caller) 327 328 329 _contextmanager = decorator(ContextManager) 330 331 332 def contextmanager(func): 333 # Enable Pylint config: contextmanager-decorators=decorator.contextmanager 334 return _contextmanager(func) 335 336 337 # ############################ dispatch_on ############################ # 338 339 def append(a, vancestors): 340 """ 341 Append ``a`` to the list of the virtual ancestors, unless it is already 342 included. 343 """ 344 add = True 345 for j, va in enumerate(vancestors): 346 if issubclass(va, a): 347 add = False 348 break 349 if issubclass(a, va): 350 vancestors[j] = a 351 add = False 352 if add: 353 vancestors.append(a) 354 355 356 # inspired from simplegeneric by P.J. Eby and functools.singledispatch 357 def dispatch_on(*dispatch_args): 358 """ 359 Factory of decorators turning a function into a generic function 360 dispatching on the given arguments. 361 """ 362 assert dispatch_args, 'No dispatch args passed' 363 dispatch_str = '(%s,)' % ', '.join(dispatch_args) 364 365 def check(arguments, wrong=operator.ne, msg=''): 366 """Make sure one passes the expected number of arguments""" 367 if wrong(len(arguments), len(dispatch_args)): 368 raise TypeError('Expected %d arguments, got %d%s' % 369 (len(dispatch_args), len(arguments), msg)) 370 371 def gen_func_dec(func): 372 """Decorator turning a function into a generic function""" 373 374 # first check the dispatch arguments 375 argset = set(getfullargspec(func).args) 376 if not set(dispatch_args) <= argset: 377 raise NameError('Unknown dispatch arguments %s' % dispatch_str) 378 379 typemap = {} 380 381 def vancestors(*types): 382 """ 383 Get a list of sets of virtual ancestors for the given types 384 """ 385 check(types) 386 ras = [[] for _ in range(len(dispatch_args))] 387 for types_ in typemap: 388 for t, type_, ra in zip(types, types_, ras): 389 if issubclass(t, type_) and type_ not in t.mro(): 390 append(type_, ra) 391 return [set(ra) for ra in ras] 392 393 def ancestors(*types): 394 """ 395 Get a list of virtual MROs, one for each type 396 """ 397 check(types) 398 lists = [] 399 for t, vas in zip(types, vancestors(*types)): 400 n_vas = len(vas) 401 if n_vas > 1: 402 raise RuntimeError( 403 'Ambiguous dispatch for %s: %s' % (t, vas)) 404 elif n_vas == 1: 405 va, = vas 406 mro = type('t', (t, va), {}).mro()[1:] 407 else: 408 mro = t.mro() 409 lists.append(mro[:-1]) # discard t and object 410 return lists 411 412 def register(*types): 413 """ 414 Decorator to register an implementation for the given types 415 """ 416 check(types) 417 418 def dec(f): 419 check(getfullargspec(f).args, operator.lt, ' in ' + f.__name__) 420 typemap[types] = f 421 return f 422 return dec 423 424 def dispatch_info(*types): 425 """ 426 An utility to introspect the dispatch algorithm 427 """ 428 check(types) 429 lst = [] 430 for ancs in itertools.product(*ancestors(*types)): 431 lst.append(tuple(a.__name__ for a in ancs)) 432 return lst 433 434 def _dispatch(dispatch_args, *args, **kw): 435 types = tuple(type(arg) for arg in dispatch_args) 436 try: # fast path 437 f = typemap[types] 438 except KeyError: 439 pass 440 else: 441 return f(*args, **kw) 442 combinations = itertools.product(*ancestors(*types)) 443 next(combinations) # the first one has been already tried 444 for types_ in combinations: 445 f = typemap.get(types_) 446 if f is not None: 447 return f(*args, **kw) 448 449 # else call the default implementation 450 return func(*args, **kw) 451 452 return FunctionMaker.create( 453 func, 'return _f_(%s, %%(shortsignature)s)' % dispatch_str, 454 dict(_f_=_dispatch), register=register, default=func, 455 typemap=typemap, vancestors=vancestors, ancestors=ancestors, 456 dispatch_info=dispatch_info, __wrapped__=func) 457 458 gen_func_dec.__name__ = 'dispatch_on' + dispatch_str 459 return gen_func_dec 460