decorator.py (11816B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\decorator.py 2 from __future__ import print_function 3 import re 4 import sys 5 import inspect 6 import operator 7 import itertools 8 import collections 9 __version__ = '4.4.0' 10 if sys.version >= '3': 11 from inspect import getfullargspec 12 13 def get_init(cls): 14 return cls.__init__ 15 16 17 else: 18 FullArgSpec = collections.namedtuple('FullArgSpec', 'args varargs varkw defaults kwonlyargs kwonlydefaults annotations') 19 20 def getfullargspec(f): 21 return FullArgSpec._make(inspect.getargspec(f) + ([], None, {})) 22 23 24 def get_init(cls): 25 return cls.__init__.__func__ 26 27 28 try: 29 iscoroutinefunction = inspect.iscoroutinefunction 30 except AttributeError: 31 32 def iscoroutinefunction(f): 33 return False 34 35 36 try: 37 from inspect import isgeneratorfunction 38 except ImportError: 39 40 def isgeneratorfunction(caller): 41 return False 42 43 44 DEF = re.compile('\\s*def\\s*([_\\w][_\\w\\d]*)\\s*\\(') 45 46 class FunctionMaker(object): 47 _compile_count = itertools.count() 48 args = varargs = varkw = defaults = kwonlyargs = kwonlydefaults = () 49 50 def __init__(self, func = None, name = None, signature = None, defaults = None, doc = None, module = None, funcdict = None): 51 self.shortsignature = signature 52 if func: 53 self.name = func.__name__ 54 if self.name == '<lambda>': 55 self.name = '_lambda_' 56 self.doc = func.__doc__ 57 self.module = func.__module__ 58 if inspect.isfunction(func): 59 argspec = getfullargspec(func) 60 self.annotations = getattr(func, '__annotations__', {}) 61 for a in ('args', 'varargs', 'varkw', 'defaults', 'kwonlyargs', 'kwonlydefaults'): 62 setattr(self, a, getattr(argspec, a)) 63 64 for i, arg in enumerate(self.args): 65 setattr(self, 'arg%d' % i, arg) 66 67 allargs = list(self.args) 68 allshortargs = list(self.args) 69 if self.varargs: 70 allargs.append('*' + self.varargs) 71 allshortargs.append('*' + self.varargs) 72 elif self.kwonlyargs: 73 allargs.append('*') 74 for a in self.kwonlyargs: 75 allargs.append('%s=None' % a) 76 allshortargs.append('%s=%s' % (a, a)) 77 78 if self.varkw: 79 allargs.append('**' + self.varkw) 80 allshortargs.append('**' + self.varkw) 81 self.signature = ', '.join(allargs) 82 self.shortsignature = ', '.join(allshortargs) 83 self.dict = func.__dict__.copy() 84 if name: 85 self.name = name 86 if signature is not None: 87 self.signature = signature 88 if defaults: 89 self.defaults = defaults 90 if doc: 91 self.doc = doc 92 if module: 93 self.module = module 94 if funcdict: 95 self.dict = funcdict 96 if not hasattr(self, 'signature'): 97 raise TypeError('You are decorating a non function: %s' % func) 98 99 def update(self, func, **kw): 100 func.__name__ = self.name 101 func.__doc__ = getattr(self, 'doc', None) 102 func.__dict__ = getattr(self, 'dict', {}) 103 func.__defaults__ = self.defaults 104 func.__kwdefaults__ = self.kwonlydefaults or None 105 func.__annotations__ = getattr(self, 'annotations', None) 106 try: 107 frame = sys._getframe(3) 108 except AttributeError: 109 callermodule = '?' 110 else: 111 callermodule = frame.f_globals.get('__name__', '?') 112 113 func.__module__ = getattr(self, 'module', callermodule) 114 func.__dict__.update(kw) 115 116 def make(self, src_templ, evaldict = None, addsource = False, **attrs): 117 src = src_templ % vars(self) 118 evaldict = evaldict or {} 119 mo = DEF.search(src) 120 if mo is None: 121 raise SyntaxError('not a valid function template\n%s' % src) 122 name = mo.group(1) 123 names = set([name] + [ arg.strip(' *') for arg in self.shortsignature.split(',') ]) 124 for n in names: 125 if n in ('_func_', '_call_'): 126 raise NameError('%s is overridden in\n%s' % (n, src)) 127 128 if not src.endswith('\n'): 129 src += '\n' 130 filename = '<%s:decorator-gen-%d>' % (__file__, next(self._compile_count)) 131 try: 132 code = compile(src, filename, 'single') 133 exec code in evaldict 134 except Exception: 135 print('Error in generated code:', file=sys.stderr) 136 print(src, file=sys.stderr) 137 raise 138 139 func = evaldict[name] 140 if addsource: 141 attrs['__source__'] = src 142 self.update(func, **attrs) 143 return func 144 145 @classmethod 146 def create(cls, obj, body, evaldict, defaults = None, doc = None, module = None, addsource = True, **attrs): 147 if isinstance(obj, str): 148 name, rest = obj.strip().split('(', 1) 149 signature = rest[:-1] 150 func = None 151 else: 152 name = None 153 signature = None 154 func = obj 155 self = cls(func, name, signature, defaults, doc, module) 156 ibody = '\n'.join((' ' + line for line in body.splitlines())) 157 caller = evaldict.get('_call_') 158 if caller and iscoroutinefunction(caller): 159 body = ('async def %(name)s(%(signature)s):\n' + ibody).replace('return', 'return await') 160 else: 161 body = 'def %(name)s(%(signature)s):\n' + ibody 162 return self.make(body, evaldict, addsource, **attrs) 163 164 165 def decorate(func, caller, extras = ()): 166 evaldict = dict(_call_=caller, _func_=func) 167 es = '' 168 for i, extra in enumerate(extras): 169 ex = '_e%d_' % i 170 evaldict[ex] = extra 171 es += ex + ', ' 172 173 if '3.5' <= sys.version < '3.6': 174 generatorcaller = isgeneratorfunction(caller) and not iscoroutinefunction(caller) 175 else: 176 generatorcaller = isgeneratorfunction(caller) 177 if generatorcaller: 178 fun = FunctionMaker.create(func, 'for res in _call_(_func_, %s%%(shortsignature)s):\n yield res' % es, evaldict, __wrapped__=func) 179 else: 180 fun = FunctionMaker.create(func, 'return _call_(_func_, %s%%(shortsignature)s)' % es, evaldict, __wrapped__=func) 181 if hasattr(func, '__qualname__'): 182 fun.__qualname__ = func.__qualname__ 183 return fun 184 185 186 def decorator(caller, _func = None): 187 if _func is not None: 188 return decorate(_func, caller) 189 defaultargs, defaults = '', () 190 if inspect.isclass(caller): 191 name = caller.__name__.lower() 192 doc = 'decorator(%s) converts functions/generators into factories of %s objects' % (caller.__name__, caller.__name__) 193 elif inspect.isfunction(caller): 194 if caller.__name__ == '<lambda>': 195 name = '_lambda_' 196 else: 197 name = caller.__name__ 198 doc = caller.__doc__ 199 nargs = caller.__code__.co_argcount 200 ndefs = len(caller.__defaults__ or ()) 201 defaultargs = ', '.join(caller.__code__.co_varnames[nargs - ndefs:nargs]) 202 if defaultargs: 203 defaultargs += ',' 204 defaults = caller.__defaults__ 205 else: 206 name = caller.__class__.__name__.lower() 207 doc = caller.__call__.__doc__ 208 evaldict = dict(_call=caller, _decorate_=decorate) 209 dec = FunctionMaker.create('%s(func, %s)' % (name, defaultargs), 'if func is None: return lambda func: _decorate_(func, _call, (%s))\nreturn _decorate_(func, _call, (%s))' % (defaultargs, defaultargs), evaldict, doc=doc, module=caller.__module__, __wrapped__=caller) 210 if defaults: 211 dec.__defaults__ = (None,) + defaults 212 return dec 213 214 215 try: 216 from contextlib import _GeneratorContextManager 217 except ImportError: 218 from contextlib import GeneratorContextManager as _GeneratorContextManager 219 220 class ContextManager(_GeneratorContextManager): 221 222 def __call__(self, func): 223 return FunctionMaker.create(func, 'with _self_: return _func_(%(shortsignature)s)', dict(_self_=self, _func_=func), __wrapped__=func) 224 225 226 init = getfullargspec(_GeneratorContextManager.__init__) 227 n_args = len(init.args) 228 if n_args == 2 and not init.varargs: 229 230 def __init__(self, g, *a, **k): 231 return _GeneratorContextManager.__init__(self, g(*a, **k)) 232 233 234 ContextManager.__init__ = __init__ 235 elif n_args == 2 and init.varargs: 236 pass 237 elif n_args == 4: 238 239 def __init__(self, g, *a, **k): 240 return _GeneratorContextManager.__init__(self, g, a, k) 241 242 243 ContextManager.__init__ = __init__ 244 _contextmanager = decorator(ContextManager) 245 246 def contextmanager(func): 247 return _contextmanager(func) 248 249 250 def append(a, vancestors): 251 add = True 252 for j, va in enumerate(vancestors): 253 if issubclass(va, a): 254 add = False 255 break 256 if issubclass(a, va): 257 vancestors[j] = a 258 add = False 259 260 if add: 261 vancestors.append(a) 262 263 264 def dispatch_on(*dispatch_args): 265 dispatch_str = '(%s,)' % ', '.join(dispatch_args) 266 267 def check(arguments, wrong = operator.ne, msg = ''): 268 if wrong(len(arguments), len(dispatch_args)): 269 raise TypeError('Expected %d arguments, got %d%s' % (len(dispatch_args), len(arguments), msg)) 270 271 def gen_func_dec(func): 272 argset = set(getfullargspec(func).args) 273 if not set(dispatch_args) <= argset: 274 raise NameError('Unknown dispatch arguments %s' % dispatch_str) 275 typemap = {} 276 277 def vancestors(*types): 278 check(types) 279 ras = [ [] for _ in range(len(dispatch_args)) ] 280 for types_ in typemap: 281 for t, type_, ra in zip(types, types_, ras): 282 if issubclass(t, type_) and type_ not in t.mro(): 283 append(type_, ra) 284 285 return [ set(ra) for ra in ras ] 286 287 def ancestors(*types): 288 check(types) 289 lists = [] 290 for t, vas in zip(types, vancestors(*types)): 291 n_vas = len(vas) 292 if n_vas > 1: 293 raise RuntimeError('Ambiguous dispatch for %s: %s' % (t, vas)) 294 elif n_vas == 1: 295 va, = vas 296 mro = type('t', (t, va), {}).mro()[1:] 297 else: 298 mro = t.mro() 299 lists.append(mro[:-1]) 300 301 return lists 302 303 def register(*types): 304 check(types) 305 306 def dec(f): 307 check(getfullargspec(f).args, operator.lt, ' in ' + f.__name__) 308 typemap[types] = f 309 return f 310 311 return dec 312 313 def dispatch_info(*types): 314 check(types) 315 lst = [] 316 for anc in itertools.product(*ancestors(*types)): 317 lst.append(tuple((a.__name__ for a in anc))) 318 319 return lst 320 321 def _dispatch(dispatch_args, *args, **kw): 322 types = tuple((type(arg) for arg in dispatch_args)) 323 try: 324 f = typemap[types] 325 except KeyError: 326 pass 327 else: 328 return f(*args, **kw) 329 330 combinations = itertools.product(*ancestors(*types)) 331 next(combinations) 332 for types_ in combinations: 333 f = typemap.get(types_) 334 if f is not None: 335 return f(*args, **kw) 336 337 return func(*args, **kw) 338 339 return FunctionMaker.create(func, 'return _f_(%s, %%(shortsignature)s)' % dispatch_str, dict(_f_=_dispatch), register=register, default=func, typemap=typemap, vancestors=vancestors, ancestors=ancestors, dispatch_info=dispatch_info, __wrapped__=func) 340 341 gen_func_dec.__name__ = 'dispatch_on' + dispatch_str 342 return gen_func_dec