Python decorator 模块，decorate() 实例源码

def api_error_handle(func):
    def func_wrapper(f, *args, **kwargs):
        try:
            return f(*args, **kwargs)
        except RateLimitError as e:
            raise APIQuotaError(str(e))
        except TweepError as e:
            if e.api_code == TWITTER_AUTOMATED_REQUEST_ERROR:
                raise AutomatedRequestError
            elif e.api_code == TWITTER_OVER_CAPACITY_ERROR:
                raise OverCapacityError
            elif e.api_code in [TWITTER_CHARACTER_LIMIT_ERROR_1, TWITTER_CHARACTER_LIMIT_ERROR_2]:
                raise CharacterLimitError
            elif e.api_code == TWITTER_DAILY_STATUS_UPDATE_LIMIT_ERROR:
                raise DailyStatusUpdateError
            elif e.api_code == TWITTER_STATUS_DUPLICATE_ERROR:
                raise StatusDuplicateError
            else:
                raise

    return decorate(func, func_wrapper)

def api_version(created_ver, last_changed_ver):
    """Version check decorator. Currently only checks Bigger Than."""
    def api_min_version_decorator(function):      
        def wrapper(function, self, *args, **kwargs):
            if not self.version_check_mode == "none":
                if self.version_check_mode == "created":
                    version = created_ver
                else:
                    version = last_changed_ver
                major, minor, patch = parse_version_string(version)
                if major > self.mastodon_major:
                    raise MastodonVersionError("Version check failed (Need version " + version + ")")
                elif major == self.mastodon_major and minor > self.mastodon_minor:
                    raise MastodonVersionError("Version check failed (Need version " + version + ")")
                elif major == self.mastodon_major and minor == self.mastodon_minor and patch > self.mastodon_patch:
                    raise MastodonVersionError("Version check failed (Need version " + version + ")")
            return function(self, *args, **kwargs)
        function.__doc__ = function.__doc__ + "\n\n        *Added: Mastodon v" + created_ver + ", last changed: Mastodon v" + last_changed_ver + "*"
        return decorate(function, wrapper)
    return api_min_version_decorator

###
# Dict helper class.
# Defined at top level so it can be pickled.
###

def logging_scope(name, *wargs, **wkwargs):
    '''
    A decorator to add the decorated function as a new logging scope, with name `name`.
    All additional arguments are passed to `StatusLogger._scope_enter`. Current
    supported keyword arguments are "timed", in which case when the scope closes,
    the duration of the call is shown.
    '''
    def logging_scope(func, *args, **kwargs):
        logger._scope_enter(name, *wargs, **wkwargs)
        success = True
        try:
            f = func(*args, **kwargs)
            return f
        except Exception as e:
            success = False
            raise_with_traceback(e)
        finally:
            logger._scope_exit(success)
    return lambda func: decorate(func, logging_scope)

def leakcheck(fn):
    # Compare the memory (after GC) before and after the test to determine
    # if we leaked memory
    def wrapper(fn, *args, **kwargs):
        gc.collect()
        mem_before = MemoryUsage()
        rv = fn(*args, **kwargs)
        gc.collect()
        mem_after = MemoryUsage()
        assert mem_before >= mem_after
        return rv
    return decorator.decorate(fn, wrapper)

def node(comp, name=None, *args, **kw):
    def inner(f):
        if name is None:
            comp.add_node(f.__name__, f, *args, **kw)
        else:
            comp.add_node(name, f, *args, **kw)
        return decorator.decorate(f, _node)
    return inner

def _add_contract_to_setter(setter_fun, var_name, property_contract, property_name):

    # 0. check that we can import contracts
    try:
        # noinspection PyUnresolvedReferences
        from contracts import ContractNotRespected, contract
    except ImportError as e:
        raise Exception('Use of _add_contract_to_setter requires that PyContract library is installed. Check that you '
                        'can \'import contracts\'')

    # -- check if a contract already exists on the function
    if hasattr(setter_fun, '__contracts__'):
        msg = 'overridden setter for attribute ' + property_name + ' implemented by function ' \
              + str(setter_fun.__qualname__) + ' has a contract while there is a contract already defined ' \
              + 'for this property in the __init__ constructor. This will lead to double-contract in the final ' \
              + 'setter, please remove the one on the overridden setter.'
        warn(msg)

    # -- add the generated contract
    setter_fun_with_possible_contract = contract(setter_fun, **{var_name: property_contract})

    # the only thing we can't do is to replace the function's parameter name dynamically in the error messages
    # so we wrap the function again to catch the potential pycontracts error :(
    # old:
    # @functools.wraps(func) -> to make the wrapper function look like the wrapped function
    # def wrapper(self, *args, **kwargs):
    # new:
    # we now use 'decorate' to have a wrapper that has the same signature, see below
    def _contracts_parser_interceptor(func, self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except ContractNotRespected as er:
            er.error = er.error.replace('\'val\'', '\'' + property_name + '\'')
            raise er

    # f = _contracts_parser_interceptor(f)
    setter_fun_with_possible_contract = decorate(setter_fun_with_possible_contract, _contracts_parser_interceptor)
    return setter_fun_with_possible_contract

def _add_validators_to_setter(setter_fun, var_name, validators, property_name):

    # 0. check that we can import validate
    # note: this is useless now but maybe one day validate will be another project ?
    try:
        # noinspection PyUnresolvedReferences
        from valid8 import decorate_with_validators
    except ImportError:
        raise Exception('Use of _add_contract_to_setter requires that validate library is installed. Check that you can'
                        ' \'import validate\'')

    # -- check if a contract already exists on the function
    if hasattr(setter_fun, '__validators__'):
        msg = 'overridden setter for attribute ' + property_name + ' implemented by function ' \
              + str(setter_fun.__qualname__) + ' has validators while there are validators already defined ' \
              'for this property in the __init__ constructor. This will lead to double-contract in the final ' \
              'setter, please remove the one on the overridden setter.'
        warn(msg)

    # -- add the generated contract
    setter_fun_with_validation = decorate_with_validators(setter_fun, **{var_name: validators})

    # # the only thing we can't do is to replace the function's parameter name dynamically in the validation error
    # #  messages so we wrap the function again to catch the potential pycontracts error :(
    # # old:
    # # @functools.wraps(func) -> to make the wrapper function look like the wrapped function
    # # def wrapper(self, *args, **kwargs):
    # # new:
    # # we now use 'decorate' to have a wrapper that has the same signature, see below
    # def _contracts_parser_interceptor(func, self, *args, **kwargs):
    #     try:
    #         return func(self, *args, **kwargs)
    #     except ContractNotRespected as e:
    #         e.error = e.error.replace('\'val\'', '\'' + property_name + '\'')
    #         raise e

    # f = _contracts_parser_interceptor(f)
    # setter_fun_with_possible_contract = decorate(setter_fun_with_possible_contract, _contracts_parser_interceptor)
    return setter_fun_with_validation

def autoprops_override_decorate(func: Callable, attribute:str = None, is_getter:bool = True) -> Callable:
    """
    Used to decorate a function as an overridden getter or setter, without using the @getter_override or
    @setter_override annotations. If the overridden setter has no @contract, the contract will still be
     dynamically added. Note: this should be executed BEFORE @autoprops or autoprops_decorate().

    :param func: the function on which to execute. Note that it won't be wrapped but simply annotated.
    :param attribute: the attribute name. If None, the function name will be used
    :param is_getter: True for a getter override, False for a setter override.
    :return:
    """

    # Simply annotate the fact that this is a function
    attr_name = attribute or func.__name__
    if is_getter:
        if hasattr(func, __GETTER_OVERRIDE_ANNOTATION):
            raise DuplicateOverrideError('Getter is overridden twice for attribute name : ' + attr_name)
        else:
            func.__getter_override__ = attr_name
    else:
        if hasattr(func, __SETTER_OVERRIDE_ANNOTATION):
            raise DuplicateOverrideError('Setter is overridden twice for attribute name : ' + attr_name)
        else:
            func.__setter_override__ = attr_name

    return func

def deprecated(*optional_message):
    """This is a decorator which can be used to mark functions
    as deprecated. It will result in a warning being emitted
    when the function is used.

    Parameters
    ----------
    *optional_message : str
        an optional user level hint which should indicate which feature to use otherwise.

    """
    def _deprecated(func, *args, **kw):
        caller_stack = stack()[1:]
        while len(caller_stack) > 0:
            frame = caller_stack.pop(0)
            filename = frame[1]
            # skip callee frames if they are other decorators or this file(func)
            if 'decorator' in filename or __file__ in filename:
                continue
            else: break
        lineno = frame[2]
        # avoid cyclic references!
        del caller_stack, frame

        user_msg = 'Call to deprecated function "%s". Called from %s line %i. %s' \
                   % (func.__name__, filename, lineno, msg)

        warnings.warn_explicit(
            user_msg,
            category=Chainsaw_DeprecationWarning,
            filename=filename,
            lineno=lineno
        )
        return func(*args, **kw)

    # add deprecation notice to func docstring:


    if len(optional_message) == 1 and callable(optional_message[0]):
        # this is the function itself, decorate!
        msg = ""
        return decorate(optional_message[0], _deprecated)
    else:
        # actually got a message (or empty parenthesis)
        msg = optional_message[0] if len(optional_message) > 0 else ""
        return decorator(_deprecated)