我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用functools.update_wrapper()。
def decorator(target): """A signature-matching decorator factory.""" def decorate(fn): if not inspect.isfunction(fn): raise Exception("not a decoratable function") spec = compat.inspect_getfullargspec(fn) names = tuple(spec[0]) + spec[1:3] + (fn.__name__,) targ_name, fn_name = _unique_symbols(names, 'target', 'fn') metadata = dict(target=targ_name, fn=fn_name) metadata.update(format_argspec_plus(spec, grouped=False)) metadata['name'] = fn.__name__ code = """\ def %(name)s(%(args)s): return %(target)s(%(fn)s, %(apply_kw)s) """ % metadata decorated = _exec_code_in_env(code, {targ_name: target, fn_name: fn}, fn.__name__) decorated.__defaults__ = getattr(fn, 'im_func', fn).__defaults__ decorated.__wrapped__ = fn return update_wrapper(decorated, fn) return update_wrapper(decorate, target)
def memoized_instancemethod(fn): """Decorate a method memoize its return value. Best applied to no-arg methods: memoization is not sensitive to argument values, and will always return the same value even when called with different arguments. """ def oneshot(self, *args, **kw): result = fn(self, *args, **kw) memo = lambda *a, **kw: result memo.__name__ = fn.__name__ memo.__doc__ = fn.__doc__ self.__dict__[fn.__name__] = memo return result return update_wrapper(oneshot, fn)
def application(cls, f): """Decorate a function as responder that accepts the request as first argument. This works like the :func:`responder` decorator but the function is passed the request object as first argument and the request object will be closed automatically:: @Request.application def my_wsgi_app(request): return Response('Hello World!') :param f: the WSGI callable to decorate :return: a new WSGI callable """ #: return a callable that wraps the -2nd argument with the request #: and calls the function with all the arguments up to that one and #: the request. The return value is then called with the latest #: two arguments. This makes it possible to use this decorator for #: both methods and standalone WSGI functions. def application(*args): request = cls(args[-2]) with request: return f(*args[:-2] + (request,))(*args[-2:]) return update_wrapper(application, f)
def _retry_on_deadlock(f): """Decorator to retry a DB API call if Deadlock was received.""" @functools.wraps(f) def wrapped(*args, **kwargs): while True: try: return f(*args, **kwargs) except db_exc.DBDeadlock: LOG.warning(_LW("Deadlock detected when running " "'%(func_name)s': Retrying..."), dict(func_name=f.__name__)) # Retry! time.sleep(0.5) continue functools.update_wrapper(wrapped, f) return wrapped
def get_urls(self): from django.conf.urls import url def wrap(view): def wrapper(*args, **kwargs): return self.admin_site.admin_view(view)(*args, **kwargs) wrapper.model_admin = self return update_wrapper(wrapper, view) info = self.model._meta.app_label, self.model._meta.model_name urlpatterns = [ url(r'^$', wrap(self.changelist_view), name='%s_%s_changelist' % info), url(r'^add/$', wrap(self.add_view), name='%s_%s_add' % info), url(r'^(.+)/history/$', wrap(self.history_view), name='%s_%s_history' % info), url(r'^(.+)/delete/$', wrap(self.delete_view), name='%s_%s_delete' % info), url(r'^(.+)/change/$', wrap(self.change_view), name='%s_%s_change' % info), # For backwards compatibility (was the change url before 1.9) url(r'^(.+)/$', wrap(RedirectView.as_view( pattern_name='%s:%s_%s_change' % ((self.admin_site.name,) + info) ))), ] return urlpatterns
def exhaust_stream(f): """Helper decorator for methods that exhausts the stream on return.""" def wrapper(self, stream, *args, **kwargs): try: return f(self, stream, *args, **kwargs) finally: exhaust = getattr(stream, 'exhaust', None) if exhaust is not None: exhaust() else: while 1: chunk = stream.read(1024 * 64) if not chunk: break return update_wrapper(wrapper, f)
def get_urls(self): urls = super(SentNotificationAdmin, self).get_urls() opts = self.model._meta # pylint: disable=W0212 def wrap(view): """ Copied from super class """ def wrapper(*args, **kwargs): """ Copied from super class """ return self.admin_site.admin_view(view)(*args, **kwargs) return update_wrapper(wrapper, view) info = opts.app_label, opts.model_name return [ url(r'^(.+)/resend/$', wrap(self.resend_view), name='%s_%s_resend' % info), ] + urls
def do(self, job_func, *args, **kwargs): """Specifies the job_func that should be called every time the job runs. Any additional arguments are passed on to job_func when the job runs. """ self.job_func = functools.partial(job_func, *args, **kwargs) try: functools.update_wrapper(self.job_func, job_func) except AttributeError: # job_funcs already wrapped by functools.partial won't have # __name__, __module__ or __doc__ and the update_wrapper() # call will fail. pass self._schedule_next_run() return self
def get_api_urls(self): def wrap(view): def wrapper(*args, **kwargs): return self.admin_site.admin_view(view)(*args, **kwargs) wrapper.model_admin = self return update_wrapper(wrapper, view) info = self.model._meta.app_label, self.model._meta.model_name urlpatterns = [ url(r'^$', wrap(self.admin_site.admin_view(self.dispatch)), name='%s_%s_api_list' % info), url(r'^(?P<pk>[-\d]+)/$', wrap(self.admin_site.admin_view(self.dispatch)), name='%s_%s_api_detail' % info), url(r'^meta/$', wrap(self.admin_site.admin_view(self.api_meta_view)), name='%s_%s_api_meta' % info), ] return urlpatterns
def get_urls(self): """ Add our own ``move`` view. """ def wrap(view): def wrapper(*args, **kwargs): return self.admin_site.admin_view(view)(*args, **kwargs) wrapper.model_admin = self return update_wrapper(wrapper, view) info = self.model._meta.app_label, self.model._meta.model_name return [ url(r'^(.+)/move/$', wrap(self.move_view), name='%s_%s_move' % info), ] + super(TreeAdmin, self).get_urls()
def webhook(f): @csrf_exempt def new_f(request, *args, **kwargs): data = json.loads(request.body) or {} with Context.for_request(request, data) as context: return f(request, context, data, *args, **kwargs) return update_wrapper(new_f, f)
def with_context(f): def new_f(request, *args, **kwargs): with Context.for_request(request) as context: return f(request, context, *args, **kwargs) return update_wrapper(new_f, f)
def allow_frame(f): def new_f(request, *args, **kwargs): resp = f(request, *args, **kwargs) # put something here so that sentry does not overwrite it # with deny. resp['X-Frame-Options'] = 'allow' return resp return update_wrapper(new_f, f)
def cors(f): def new_f(request, *args, **kwargs): origin = request.META.get('HTTP_ORIGIN') resp = f(request, *args, **kwargs) resp['Access-Control-Allow-Origin'] = origin resp['Access-Control-Request-Method'] = 'GET, HEAD, OPTIONS' resp['Access-Control-Allow-Headers'] = 'X-Requested-With' resp['Access-Control-Allow-Credentials'] = 'true' resp['Access-Control-Max-Age'] = '1728000' return resp return update_wrapper(new_f, f)
def __call__(self, fn): def _preparer_wrapper(test_class_instance, **kwargs): self.live_test = not isinstance(test_class_instance, ReplayableTest) self.test_class_instance = test_class_instance if self.live_test or test_class_instance.in_recording: resource_name = self.random_name if not self.live_test and isinstance(self, RecordingProcessor): test_class_instance.recording_processors.append(self) else: resource_name = self.moniker with self.override_disable_recording(): parameter_update = self.create_resource( resource_name, **kwargs ) test_class_instance.addCleanup( lambda: self.remove_resource_with_record_override(resource_name, **kwargs) ) if parameter_update: kwargs.update(parameter_update) if not is_preparer_func(fn): # the next function is the actual test function. the kwargs need to be trimmed so # that parameters which are not required will not be passed to it. args, _, kw, _ = inspect.getargspec(fn) # pylint: disable=deprecated-method if kw is None: args = set(args) for key in [k for k in kwargs if k not in args]: del kwargs[key] fn(test_class_instance, **kwargs) setattr(_preparer_wrapper, '__is_preparer', True) functools.update_wrapper(_preparer_wrapper, fn) return _preparer_wrapper
def __call__(self, fn): import_deps = self.import_deps spec = compat.inspect_getfullargspec(fn) spec_zero = list(spec[0]) hasself = spec_zero[0] in ('self', 'cls') for i in range(len(import_deps)): spec[0][i + (1 if hasself else 0)] = "import_deps[%r]" % i inner_spec = format_argspec_plus(spec, grouped=False) for impname in import_deps: del spec_zero[1 if hasself else 0] spec[0][:] = spec_zero outer_spec = format_argspec_plus(spec, grouped=False) code = 'lambda %(args)s: fn(%(apply_kw)s)' % { "args": outer_spec['args'], "apply_kw": inner_spec['apply_kw'] } decorated = eval(code, locals()) decorated.__defaults__ = getattr(fn, 'im_func', fn).__defaults__ return update_wrapper(decorated, fn)
def _wrap_w_kw(self, fn): def wrap(*arg, **kw): return fn(*arg) return update_wrapper(wrap, fn)
def responder(f): """Marks a function as responder. Decorate a function with it and it will automatically call the return value as WSGI application. Example:: @responder def application(environ, start_response): return Response('Hello World!') """ return update_wrapper(lambda *a: f(*a)(*a[-2:]), f)
def middleware(self, func): """Like `make_middleware` but for decorating functions. Example usage:: @manager.middleware def application(environ, start_response): ... The difference to `make_middleware` is that the function passed will have all the arguments copied from the inner application (name, docstring, module). """ return update_wrapper(self.make_middleware(func), func)
def native_string_result(func): def wrapper(*args, **kwargs): return func(*args, **kwargs).encode('utf-8') return functools.update_wrapper(wrapper, func)
def emits_module_deprecation_warning(f): def new_f(self, *args, **kwargs): with catch_warnings() as log: f(self, *args, **kwargs) self.assert_true(log, 'expected deprecation warning') for entry in log: self.assert_in('Modules are deprecated', str(entry['message'])) return update_wrapper(new_f, f)
def setupmethod(f): """Wraps a method so that it performs a check in debug mode if the first request was already handled. """ def wrapper_func(self, *args, **kwargs): if self.debug and self._got_first_request: raise AssertionError('A setup function was called after the ' 'first request was handled. This usually indicates a bug ' 'in the application where a module was not imported ' 'and decorators or other functionality was called too late.\n' 'To fix this make sure to import all your view modules, ' 'database models and everything related at a central place ' 'before the application starts serving requests.') return f(self, *args, **kwargs) return update_wrapper(wrapper_func, f)
def record_once(self, func): """Works like :meth:`record` but wraps the function in another function that will ensure the function is only called once. If the blueprint is registered a second time on the application, the function passed is not called. """ def wrapper(state): if state.first_registration: func(state) return self.record(update_wrapper(wrapper, func))
def timing(func): def wrapper(*args, **argd): start_time = time.time() ret = func(*args, **argd) print 'function: %s used %.2f seconds' % (func.__name__, time.time()-start_time) return ret functools.update_wrapper(wrapper, func) return wrapper
def cached(func): cache = {} def template(*args): #: template is wrapper; func is wrapped key = (func, )+args try: ret = cache[key] except KeyError: ret = func(*args) cache[key] = ret else: pass return ret functools.update_wrapper(template, func) return template
def crossdomain(origin=None, methods=None, headers=None, max_age=21600, attach_to_all=True, automatic_options=True): if methods is not None: methods = ', '.join(sorted(x.upper() for x in methods)) if headers is not None and not isinstance(headers, list): headers = ', '.join(x.upper() for x in headers) if not isinstance(origin, list): origin = ', '.join(origin) if isinstance(max_age, timedelta): max_age = max_age.total_seconds() def get_methods(): if methods is not None: return methods options_resp = current_app.make_default_options_response() return options_resp.headers['allow'] def decorator(f): def wrapped_function(*args, **kwargs): if automatic_options and request.method == 'OPTIONS': resp = current_app.make_default_options_response() else: resp = make_response(f(*args, **kwargs)) if not attach_to_all and request.method != 'OPTIONS': return resp h = resp.headers h['Access-Control-Allow-Origin'] = origin h['Access-Control-Allow-Methods'] = get_methods() h['Access-Control-Max-Age'] = str(max_age) h['Access-Control-Allow-Credentials'] = 'true' h['Access-Control-Allow-Headers'] = "Origin, X-Requested-With, Content-Type, Accept, Authorization" if headers is not None: h['Access-Control-Allow-Headers'] = headers return resp f.provide_automatic_options = False return update_wrapper(wrapped_function, f) return decorator
def ratelimit(limit, per, send_x_headers=True, scope_func=lambda: request.remote_addr, key_func=lambda: request.endpoint, path=lambda: request.path): """ Decorator for limiting the access to a route. Returns the function if within the limit, otherwise TooManyRequests error """ def decorator(f): @wraps(f) def rate_limited(*args, **kwargs): try: key = 'rate-limit/%s/%s/' % (key_func(), scope_func()) rlimit = RateLimit(key, limit, per, send_x_headers) g._view_rate_limit = rlimit #if over_limit is not None and rlimit.over_limit: if rlimit.over_limit: raise TooManyRequests return f(*args, **kwargs) except Exception as e: return error.format_exception(e, target=path(), action=f.__name__) return update_wrapper(rate_limited, f) return decorator
def pass_context(f): """Marks a callback as wanting to receive the current context object as first argument. """ def new_func(*args, **kwargs): return f(get_current_context(), *args, **kwargs) return update_wrapper(new_func, f)
