Decorator to explicitly mark functions that are exposed in a lib.
2608
def export(defn):
    """"""
    globals()[defn.__name__] = defn
    __all__.append(defn.__name__)
    return defn
819017
def deprecated(func):
    """"""

    @functools.wraps(func)
    def new_func(*args, **kwargs):
        if PY3K:
            func_code = func.__code__
        else:
            func_code = func.func_code

        warnings.warn_explicit(
            "Call to deprecated function {}.".format(func.__name__),
            category=DeprecationWarning,
            filename=func_code.co_filename,
            lineno=func_code.co_firstlineno + 1,
        )

        return func(*args, **kwargs)

    return new_func
1203598
def undecorate(func):
    """"""
    orig_call_wrapper = lambda x: x
    for call_wrapper, unwrap in SUPPORTED_DECORATOR.items():
        if isinstance(func, call_wrapper):
            func = unwrap(func)
            orig_call_wrapper = call_wrapper
            break
    return orig_call_wrapper, func
629609
def _decorate_namespace_function(bases: List[type], namespace: MutableMapping[str, Any], key: str) -> None:
    """"""
    # pylint: disable=too-many-branches
    # pylint: disable=too-many-locals

    value = namespace[key]
    assert inspect.isfunction(value) or isinstance(value, (staticmethod, classmethod))

    # Determine the function to be decorated
    if inspect.isfunction(value):
        func = value
    elif isinstance(value, (staticmethod, classmethod)):
        func = value.__func__
    else:
        raise NotImplementedError("Unexpected value for a function: {}".format(value))

    # Collect preconditions and postconditions of the function
    preconditions = []  # type: List[List[Contract]]
    snapshots = []  # type: List[Snapshot]
    postconditions = []  # type: List[Contract]

    contract_checker = icontract._checkers.find_checker(func=func)
    if contract_checker is not None:
        preconditions = contract_checker.__preconditions__  # type: ignore
        snapshots = contract_checker.__postcondition_snapshots__  # type: ignore
        postconditions = contract_checker.__postconditions__  # type: ignore

    # Collect the preconditions and postconditions from bases.
    #
    # Preconditions and postconditions of __init__ of base classes are deliberately ignored (and not collapsed) since
    # initialization is an operation specific to the concrete class and does not relate to the class hierarchy.
    if key not in ['__init__']:
        base_preconditions = []  # type: List[List[Contract]]
        base_snapshots = []  # type: List[Snapshot]
        base_postconditions = []  # type: List[Contract]

        bases_have_func = False
        for base in bases:
            if hasattr(base, key):
                bases_have_func = True

                # Check if there is a checker function in the base class
                base_func = getattr(base, key)
                base_contract_checker = icontract._checkers.find_checker(func=base_func)

                # Ignore functions which don't have preconditions or postconditions
                if base_contract_checker is not None:
                    base_preconditions.extend(base_contract_checker.__preconditions__)  # type: ignore
                    base_snapshots.extend(base_contract_checker.__postcondition_snapshots__)  # type: ignore
                    base_postconditions.extend(base_contract_checker.__postconditions__)  # type: ignore

        # Collapse preconditions and postconditions from the bases with the the function's own ones
        preconditions = _collapse_preconditions(
            base_preconditions=base_preconditions,
            bases_have_func=bases_have_func,
            preconditions=preconditions,
            func=func)

        snapshots = _collapse_snapshots(base_snapshots=base_snapshots, snapshots=snapshots)

        postconditions = _collapse_postconditions(
            base_postconditions=base_postconditions, postconditions=postconditions)

    if preconditions or postconditions:
        if contract_checker is None:
            contract_checker = icontract._checkers.decorate_with_checker(func=func)

            # Replace the function with the function decorated with contract checks
            if inspect.isfunction(value):
                namespace[key] = contract_checker
            elif isinstance(value, staticmethod):
                namespace[key] = staticmethod(contract_checker)

            elif isinstance(value, classmethod):
                namespace[key] = classmethod(contract_checker)

            else:
                raise NotImplementedError("Unexpected value for a function: {}".format(value))

        # Override the preconditions and postconditions
        contract_checker.__preconditions__ = preconditions  # type: ignore
        contract_checker.__postcondition_snapshots__ = snapshots  # type: ignore
        contract_checker.__postconditions__ = postconditions
176028
def _onDecorator(self, name, line, pos, absPosition):
        """"""
        # A class or a function must be on the top of the stack
        d = Decorator(name, line, pos, absPosition)
        if self.__lastDecorators is None:
            self.__lastDecorators = [d]
        else:
            self.__lastDecorators.append(d)
80573
def mark_experimental(fn):
    # type: (FunctionType) -> FunctionType
    
    @wraps(fn)
    def wrapper(*args, **kw):   # pylint: disable=missing-docstring
        from peltak.core import shell

        if shell.is_tty:
            warnings.warn("This command is has experimental status. The "
                          "interface is not yet stable and might change "
                          "without notice within with a patch version update. "
                          "Use at your own risk")
        return fn(*args, **kw)

    return wrapper
20067
def _check_decorator(fct):
        """"""
        def wrapper(self, *args, **kw):
            if self.is_enable():
                ret = fct(self, *args, **kw)
            else:
                ret = self.stats
            return ret
        return wrapper
457656
def decorate(func, caller):
    
    evaldict = dict(_call_=caller, _func_=func)
    fun = FunctionMaker.create(
        func, "return _call_(_func_, %(shortsignature)s)",
        evaldict, __wrapped__=func)
    if hasattr(func, '__qualname__'):
        fun.__qualname__ = func.__qualname__
    return fun
1366039
def _enable_cleanup(func):
    

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        self = args[0]
        result = func(*args, **kwargs)
        self.cleanup(self)
        return result

    return wrapper
292147
def patch_ui_functions(wrapper):
    ''''''
    NONE = object()
    import click

    saved = []

    for name, info in sorted(_ui_functions.items()):
        f = getattr(click, name, NONE)
        if f is NONE:
            continue

        new_f = wrapper(_copy_fn(f), info)

        argspec = getargspec(f)
        signature = inspect.formatargspec(*argspec) \
            .lstrip('(') \
            .rstrip(')')
        args = ', '.join(arg.split('=')[0].split(':')[0].strip()
                         for arg in signature.split(','))

        stub_f = eval('lambda {s}: {n}._real_click_fn({a})'
                      .format(n=f.__name__, s=signature, a=args))

        if PY2:
            saved.append((f, f.func_code))
            f.func_code = stub_f.func_code
        else:
            saved.append((f, f.__code__))
            f.__code__ = stub_f.__code__

        f._real_click_fn = new_f

    try:
        yield
    finally:
        for f, code in saved:
            if PY2:
                f.func_code = code
            else:
                f.__code__ = code

            del f._real_click_fn
637506
def representation_function_compiler(self, func_name):
        """"""

        def get_col_accessor(col):
            return ALCHEMY_TEMPLATES.col_accessor.safe_substitute(col=col)

        def get_col_evaluator(col):
            return ALCHEMY_TEMPLATES.col_evaluator.safe_substitute(col=col)

        col_evaluators = ", ".join([get_col_evaluator(n) for n in self.primary_keys])
        col_accessors = ", ".join([get_col_accessor(n) for n in self.primary_keys])

        return ALCHEMY_TEMPLATES.representor_function.safe_substitute(func_name=func_name,
                                                                      col_accessors=col_accessors,
                                                                      col_evaluators=col_evaluators,
                                                                      class_name=self.class_name)
481051
def api(func):
    """"""
    @wraps(func)
    def decorated_function(*args, **kwargs):
        try:
            value = func(*args, **kwargs)
        except BaseException as e:
            value = {
                "status": "failed",
                "error": "Unkonwn error!"
            }
            flask.current_app.logger.error("Unkonwn error: %s" % str(e))
            flask.current_app.logger.exception(e)

        status = 200
        mimetype = 'application/json'
        content = json.dumps(value)
        callback = flask.request.args.get('callback', False)
        if callback:
            content = str(callback) + '(' + content + ')'
            mimetype = 'application/javascript'

        return flask.current_app.response_class(content, status=status, mimetype=mimetype)
    return decorated_function
1334096
def with_log(func):
    
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        """Inject `log` argument into wrapped function.

        :param list args: Pass through all positional arguments.
        :param dict kwargs: Pass through all keyword arguments.
        """
        decorator_logger = logging.getLogger('@with_log')
        decorator_logger.debug('Entering %s() function call.', func.__name__)
        log = kwargs.get('log', logging.getLogger(func.__name__))
        try:
            ret = func(log=log, *args, **kwargs)
        finally:
            decorator_logger.debug('Leaving %s() function call.', func.__name__)
        return ret
    return wrapper
1018191
def log_method(log, level=logging.DEBUG):
    """"""

    def decorator(func):
        func_name = func.__name__

        @six.wraps(func)
        def wrapper(self, *args, **kwargs):
            if log.isEnabledFor(level):
                pretty_args = []
                if args:
                    pretty_args.extend(str(a) for a in args)
                if kwargs:
                    pretty_args.extend(
                        "%s=%s" % (k, v) for k, v in six.iteritems(kwargs))
                log.log(level, "%s(%s)", func_name, ", ".join(pretty_args))
            return func(self, *args, **kwargs)

        return wrapper

    return decorator
433249
def decorate_callable(self, func):
        
        def wrapper(*args, **kwargs):
            try:
                with self:
                    result = func(*args, **kwargs)
                return result
            except DBMutexError as e:
                if self.suppress_acquisition_exceptions:
                    LOG.error(e)
                else:
                    raise e
        functools.update_wrapper(wrapper, func)
        return wrapper
348633
def decorator(self, func):
        
        if inspect.isfunction(func):
            func._methodview = self
        elif inspect.ismethod(func):
            func.__func__._methodview = self
        else:
            raise AssertionError('Can only decorate function and methods, {} given'.format(func))
        return func
1028953
def function_trace(function_name):
    
    if newrelic:
        nr_transaction = newrelic.agent.current_transaction()
        with newrelic.agent.FunctionTrace(nr_transaction, function_name):
            yield
    else:
        yield