Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/wrapt/decorators.py: 46%
192 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1"""This module implements decorators for implementing other decorators
2as well as some commonly used decorators.
4"""
6import sys
8PY2 = sys.version_info[0] == 2
10if PY2:
11 string_types = basestring,
13 def exec_(_code_, _globs_=None, _locs_=None):
14 """Execute code in a namespace."""
15 if _globs_ is None:
16 frame = sys._getframe(1)
17 _globs_ = frame.f_globals
18 if _locs_ is None:
19 _locs_ = frame.f_locals
20 del frame
21 elif _locs_ is None:
22 _locs_ = _globs_
23 exec("""exec _code_ in _globs_, _locs_""")
25else:
26 string_types = str,
28 import builtins
30 exec_ = getattr(builtins, "exec")
31 del builtins
33from functools import partial
34from inspect import isclass
35from threading import Lock, RLock
37from .arguments import formatargspec
39try:
40 from inspect import signature
41except ImportError:
42 pass
44from .wrappers import (FunctionWrapper, BoundFunctionWrapper, ObjectProxy,
45 CallableObjectProxy)
47# Adapter wrapper for the wrapped function which will overlay certain
48# properties from the adapter function onto the wrapped function so that
49# functions such as inspect.getargspec(), inspect.getfullargspec(),
50# inspect.signature() and inspect.getsource() return the correct results
51# one would expect.
53class _AdapterFunctionCode(CallableObjectProxy):
55 def __init__(self, wrapped_code, adapter_code):
56 super(_AdapterFunctionCode, self).__init__(wrapped_code)
57 self._self_adapter_code = adapter_code
59 @property
60 def co_argcount(self):
61 return self._self_adapter_code.co_argcount
63 @property
64 def co_code(self):
65 return self._self_adapter_code.co_code
67 @property
68 def co_flags(self):
69 return self._self_adapter_code.co_flags
71 @property
72 def co_kwonlyargcount(self):
73 return self._self_adapter_code.co_kwonlyargcount
75 @property
76 def co_varnames(self):
77 return self._self_adapter_code.co_varnames
79class _AdapterFunctionSurrogate(CallableObjectProxy):
81 def __init__(self, wrapped, adapter):
82 super(_AdapterFunctionSurrogate, self).__init__(wrapped)
83 self._self_adapter = adapter
85 @property
86 def __code__(self):
87 return _AdapterFunctionCode(self.__wrapped__.__code__,
88 self._self_adapter.__code__)
90 @property
91 def __defaults__(self):
92 return self._self_adapter.__defaults__
94 @property
95 def __kwdefaults__(self):
96 return self._self_adapter.__kwdefaults__
98 @property
99 def __signature__(self):
100 if 'signature' not in globals():
101 return self._self_adapter.__signature__
102 else:
103 return signature(self._self_adapter)
105 if PY2:
106 func_code = __code__
107 func_defaults = __defaults__
109class _BoundAdapterWrapper(BoundFunctionWrapper):
111 @property
112 def __func__(self):
113 return _AdapterFunctionSurrogate(self.__wrapped__.__func__,
114 self._self_parent._self_adapter)
116 @property
117 def __signature__(self):
118 if 'signature' not in globals():
119 return self.__wrapped__.__signature__
120 else:
121 return signature(self._self_parent._self_adapter)
123 if PY2:
124 im_func = __func__
126class AdapterWrapper(FunctionWrapper):
128 __bound_function_wrapper__ = _BoundAdapterWrapper
130 def __init__(self, *args, **kwargs):
131 adapter = kwargs.pop('adapter')
132 super(AdapterWrapper, self).__init__(*args, **kwargs)
133 self._self_surrogate = _AdapterFunctionSurrogate(
134 self.__wrapped__, adapter)
135 self._self_adapter = adapter
137 @property
138 def __code__(self):
139 return self._self_surrogate.__code__
141 @property
142 def __defaults__(self):
143 return self._self_surrogate.__defaults__
145 @property
146 def __kwdefaults__(self):
147 return self._self_surrogate.__kwdefaults__
149 if PY2:
150 func_code = __code__
151 func_defaults = __defaults__
153 @property
154 def __signature__(self):
155 return self._self_surrogate.__signature__
157class AdapterFactory(object):
158 def __call__(self, wrapped):
159 raise NotImplementedError()
161class DelegatedAdapterFactory(AdapterFactory):
162 def __init__(self, factory):
163 super(DelegatedAdapterFactory, self).__init__()
164 self.factory = factory
165 def __call__(self, wrapped):
166 return self.factory(wrapped)
168adapter_factory = DelegatedAdapterFactory
170# Decorator for creating other decorators. This decorator and the
171# wrappers which they use are designed to properly preserve any name
172# attributes, function signatures etc, in addition to the wrappers
173# themselves acting like a transparent proxy for the original wrapped
174# function so the wrapper is effectively indistinguishable from the
175# original wrapped function.
177def decorator(wrapper=None, enabled=None, adapter=None, proxy=FunctionWrapper):
178 # The decorator should be supplied with a single positional argument
179 # which is the wrapper function to be used to implement the
180 # decorator. This may be preceded by a step whereby the keyword
181 # arguments are supplied to customise the behaviour of the
182 # decorator. The 'adapter' argument is used to optionally denote a
183 # separate function which is notionally used by an adapter
184 # decorator. In that case parts of the function '__code__' and
185 # '__defaults__' attributes are used from the adapter function
186 # rather than those of the wrapped function. This allows for the
187 # argument specification from inspect.getfullargspec() and similar
188 # functions to be overridden with a prototype for a different
189 # function than what was wrapped. The 'enabled' argument provides a
190 # way to enable/disable the use of the decorator. If the type of
191 # 'enabled' is a boolean, then it is evaluated immediately and the
192 # wrapper not even applied if it is False. If not a boolean, it will
193 # be evaluated when the wrapper is called for an unbound wrapper,
194 # and when binding occurs for a bound wrapper. When being evaluated,
195 # if 'enabled' is callable it will be called to obtain the value to
196 # be checked. If False, the wrapper will not be called and instead
197 # the original wrapped function will be called directly instead.
198 # The 'proxy' argument provides a way of passing a custom version of
199 # the FunctionWrapper class used in decorating the function.
201 if wrapper is not None:
202 # Helper function for creating wrapper of the appropriate
203 # time when we need it down below.
205 def _build(wrapped, wrapper, enabled=None, adapter=None):
206 if adapter:
207 if isinstance(adapter, AdapterFactory):
208 adapter = adapter(wrapped)
210 if not callable(adapter):
211 ns = {}
213 # Check if the signature argument specification has
214 # annotations. If it does then we need to remember
215 # it but also drop it when attempting to manufacture
216 # a standin adapter function. This is necessary else
217 # it will try and look up any types referenced in
218 # the annotations in the empty namespace we use,
219 # which will fail.
221 annotations = {}
223 if not isinstance(adapter, string_types):
224 if len(adapter) == 7:
225 annotations = adapter[-1]
226 adapter = adapter[:-1]
227 adapter = formatargspec(*adapter)
229 exec_('def adapter{}: pass'.format(adapter), ns, ns)
230 adapter = ns['adapter']
232 # Override the annotations for the manufactured
233 # adapter function so they match the original
234 # adapter signature argument specification.
236 if annotations:
237 adapter.__annotations__ = annotations
239 return AdapterWrapper(wrapped=wrapped, wrapper=wrapper,
240 enabled=enabled, adapter=adapter)
242 return proxy(wrapped=wrapped, wrapper=wrapper, enabled=enabled)
244 # The wrapper has been provided so return the final decorator.
245 # The decorator is itself one of our function wrappers so we
246 # can determine when it is applied to functions, instance methods
247 # or class methods. This allows us to bind the instance or class
248 # method so the appropriate self or cls attribute is supplied
249 # when it is finally called.
251 def _wrapper(wrapped, instance, args, kwargs):
252 # We first check for the case where the decorator was applied
253 # to a class type.
254 #
255 # @decorator
256 # class mydecoratorclass(object):
257 # def __init__(self, arg=None):
258 # self.arg = arg
259 # def __call__(self, wrapped, instance, args, kwargs):
260 # return wrapped(*args, **kwargs)
261 #
262 # @mydecoratorclass(arg=1)
263 # def function():
264 # pass
265 #
266 # In this case an instance of the class is to be used as the
267 # decorator wrapper function. If args was empty at this point,
268 # then it means that there were optional keyword arguments
269 # supplied to be used when creating an instance of the class
270 # to be used as the wrapper function.
272 if instance is None and isclass(wrapped) and not args:
273 # We still need to be passed the target function to be
274 # wrapped as yet, so we need to return a further function
275 # to be able to capture it.
277 def _capture(target_wrapped):
278 # Now have the target function to be wrapped and need
279 # to create an instance of the class which is to act
280 # as the decorator wrapper function. Before we do that,
281 # we need to first check that use of the decorator
282 # hadn't been disabled by a simple boolean. If it was,
283 # the target function to be wrapped is returned instead.
285 _enabled = enabled
286 if type(_enabled) is bool:
287 if not _enabled:
288 return target_wrapped
289 _enabled = None
291 # Now create an instance of the class which is to act
292 # as the decorator wrapper function. Any arguments had
293 # to be supplied as keyword only arguments so that is
294 # all we pass when creating it.
296 target_wrapper = wrapped(**kwargs)
298 # Finally build the wrapper itself and return it.
300 return _build(target_wrapped, target_wrapper,
301 _enabled, adapter)
303 return _capture
305 # We should always have the target function to be wrapped at
306 # this point as the first (and only) value in args.
308 target_wrapped = args[0]
310 # Need to now check that use of the decorator hadn't been
311 # disabled by a simple boolean. If it was, then target
312 # function to be wrapped is returned instead.
314 _enabled = enabled
315 if type(_enabled) is bool:
316 if not _enabled:
317 return target_wrapped
318 _enabled = None
320 # We now need to build the wrapper, but there are a couple of
321 # different cases we need to consider.
323 if instance is None:
324 if isclass(wrapped):
325 # In this case the decorator was applied to a class
326 # type but optional keyword arguments were not supplied
327 # for initialising an instance of the class to be used
328 # as the decorator wrapper function.
329 #
330 # @decorator
331 # class mydecoratorclass(object):
332 # def __init__(self, arg=None):
333 # self.arg = arg
334 # def __call__(self, wrapped, instance,
335 # args, kwargs):
336 # return wrapped(*args, **kwargs)
337 #
338 # @mydecoratorclass
339 # def function():
340 # pass
341 #
342 # We still need to create an instance of the class to
343 # be used as the decorator wrapper function, but no
344 # arguments are pass.
346 target_wrapper = wrapped()
348 else:
349 # In this case the decorator was applied to a normal
350 # function, or possibly a static method of a class.
351 #
352 # @decorator
353 # def mydecoratorfuntion(wrapped, instance,
354 # args, kwargs):
355 # return wrapped(*args, **kwargs)
356 #
357 # @mydecoratorfunction
358 # def function():
359 # pass
360 #
361 # That normal function becomes the decorator wrapper
362 # function.
364 target_wrapper = wrapper
366 else:
367 if isclass(instance):
368 # In this case the decorator was applied to a class
369 # method.
370 #
371 # class myclass(object):
372 # @decorator
373 # @classmethod
374 # def decoratorclassmethod(cls, wrapped,
375 # instance, args, kwargs):
376 # return wrapped(*args, **kwargs)
377 #
378 # instance = myclass()
379 #
380 # @instance.decoratorclassmethod
381 # def function():
382 # pass
383 #
384 # This one is a bit strange because binding was actually
385 # performed on the wrapper created by our decorator
386 # factory. We need to apply that binding to the decorator
387 # wrapper function that the decorator factory
388 # was applied to.
390 target_wrapper = wrapper.__get__(None, instance)
392 else:
393 # In this case the decorator was applied to an instance
394 # method.
395 #
396 # class myclass(object):
397 # @decorator
398 # def decoratorclassmethod(self, wrapped,
399 # instance, args, kwargs):
400 # return wrapped(*args, **kwargs)
401 #
402 # instance = myclass()
403 #
404 # @instance.decoratorclassmethod
405 # def function():
406 # pass
407 #
408 # This one is a bit strange because binding was actually
409 # performed on the wrapper created by our decorator
410 # factory. We need to apply that binding to the decorator
411 # wrapper function that the decorator factory
412 # was applied to.
414 target_wrapper = wrapper.__get__(instance, type(instance))
416 # Finally build the wrapper itself and return it.
418 return _build(target_wrapped, target_wrapper, _enabled, adapter)
420 # We first return our magic function wrapper here so we can
421 # determine in what context the decorator factory was used. In
422 # other words, it is itself a universal decorator. The decorator
423 # function is used as the adapter so that linters see a signature
424 # corresponding to the decorator and not the wrapper it is being
425 # applied to.
427 return _build(wrapper, _wrapper, adapter=decorator)
429 else:
430 # The wrapper still has not been provided, so we are just
431 # collecting the optional keyword arguments. Return the
432 # decorator again wrapped in a partial using the collected
433 # arguments.
435 return partial(decorator, enabled=enabled, adapter=adapter,
436 proxy=proxy)
438# Decorator for implementing thread synchronization. It can be used as a
439# decorator, in which case the synchronization context is determined by
440# what type of function is wrapped, or it can also be used as a context
441# manager, where the user needs to supply the correct synchronization
442# context. It is also possible to supply an object which appears to be a
443# synchronization primitive of some sort, by virtue of having release()
444# and acquire() methods. In that case that will be used directly as the
445# synchronization primitive without creating a separate lock against the
446# derived or supplied context.
448def synchronized(wrapped):
449 # Determine if being passed an object which is a synchronization
450 # primitive. We can't check by type for Lock, RLock, Semaphore etc,
451 # as the means of creating them isn't the type. Therefore use the
452 # existence of acquire() and release() methods. This is more
453 # extensible anyway as it allows custom synchronization mechanisms.
455 if hasattr(wrapped, 'acquire') and hasattr(wrapped, 'release'):
456 # We remember what the original lock is and then return a new
457 # decorator which accesses and locks it. When returning the new
458 # decorator we wrap it with an object proxy so we can override
459 # the context manager methods in case it is being used to wrap
460 # synchronized statements with a 'with' statement.
462 lock = wrapped
464 @decorator
465 def _synchronized(wrapped, instance, args, kwargs):
466 # Execute the wrapped function while the original supplied
467 # lock is held.
469 with lock:
470 return wrapped(*args, **kwargs)
472 class _PartialDecorator(CallableObjectProxy):
474 def __enter__(self):
475 lock.acquire()
476 return lock
478 def __exit__(self, *args):
479 lock.release()
481 return _PartialDecorator(wrapped=_synchronized)
483 # Following only apply when the lock is being created automatically
484 # based on the context of what was supplied. In this case we supply
485 # a final decorator, but need to use FunctionWrapper directly as we
486 # want to derive from it to add context manager methods in case it is
487 # being used to wrap synchronized statements with a 'with' statement.
489 def _synchronized_lock(context):
490 # Attempt to retrieve the lock for the specific context.
492 lock = vars(context).get('_synchronized_lock', None)
494 if lock is None:
495 # There is no existing lock defined for the context we
496 # are dealing with so we need to create one. This needs
497 # to be done in a way to guarantee there is only one
498 # created, even if multiple threads try and create it at
499 # the same time. We can't always use the setdefault()
500 # method on the __dict__ for the context. This is the
501 # case where the context is a class, as __dict__ is
502 # actually a dictproxy. What we therefore do is use a
503 # meta lock on this wrapper itself, to control the
504 # creation and assignment of the lock attribute against
505 # the context.
507 with synchronized._synchronized_meta_lock:
508 # We need to check again for whether the lock we want
509 # exists in case two threads were trying to create it
510 # at the same time and were competing to create the
511 # meta lock.
513 lock = vars(context).get('_synchronized_lock', None)
515 if lock is None:
516 lock = RLock()
517 setattr(context, '_synchronized_lock', lock)
519 return lock
521 def _synchronized_wrapper(wrapped, instance, args, kwargs):
522 # Execute the wrapped function while the lock for the
523 # desired context is held. If instance is None then the
524 # wrapped function is used as the context.
526 with _synchronized_lock(instance if instance is not None else wrapped):
527 return wrapped(*args, **kwargs)
529 class _FinalDecorator(FunctionWrapper):
531 def __enter__(self):
532 self._self_lock = _synchronized_lock(self.__wrapped__)
533 self._self_lock.acquire()
534 return self._self_lock
536 def __exit__(self, *args):
537 self._self_lock.release()
539 return _FinalDecorator(wrapped=wrapped, wrapper=_synchronized_wrapper)
541synchronized._synchronized_meta_lock = Lock()