toucan_data_sdk/utils/decorators.py
def cache(  # noqa: C901
    requires=None,
    disabled=False,
    applied_on_method=False,
    check_param=True,
    limit=None
):
    """ Avoid to recompute a function if its parameters and its source code doesnt have changed.

        Args:
            requires: list of dependencies (functions or function names)
            disabled (bool): disable the cache mecanism for this function (useful if you
                                 only want to use the dependency mecanism)
            applied_on_method (bool): ignore the first argument (useful to ignore "self")
            check_param (True, False or a str): the name of the parameter to check.
                                                    False to not check any of them.
                                                    True (default) to check all of them.
            limit (int or None): number of cache entries to keep (no limit by default)
    """
    if not requires:
        requires = []
    elif isinstance(requires, collections.Callable):
        requires = [requires]

    if not isinstance(check_param, (bool, str)):
        raise TypeError("'check_param' must be a str (name of the param to check) or a bool")
    if limit is not None and not isinstance(limit, int):
        raise TypeError("'limit' must be an int (number of cache entries to keep) or None")

    # We keep data in the function attributes so that this data
    # is not erased between two calls:
    if not hasattr(cache, 'funcs_references'):
        cache.funcs_references = {}  # dict of {function_name -> function_object (or None)}
    if not hasattr(cache, 'dependencies'):
        cache.dependencies = {}  # dict of {function_name -> [list of function names]}
    if not hasattr(cache, 'memories'):
        cache.memories = {}  # dict of {thread_id -> joblib.Memory object}

    def decorator(func):
        """ This code is executed when the augment module is read (when decorator is applied).
            Here we populate cache.funcs_references and cache.dependencies to use them later. """
        cache.funcs_references[func.__name__] = get_orig_function(func)
        dependencies_names = []
        for requirement in requires:
            if isinstance(requirement, collections.Callable):
                req_name = requirement.__name__
                cache.funcs_references[req_name] = get_orig_function(requirement)
            elif requirement not in cache.funcs_references:
                req_name = requirement
                cache.funcs_references[req_name] = None
            dependencies_names.append(req_name)

        cache.dependencies[func.__name__] = dependencies_names

        @wraps(func)
        def wrapper(*args, **kwargs):
            """ This code is executed when a decorated function is actually executed.
                It uses the previously built dependency tree (see above). """
            current_memory = cache.memories.get(current_thread().name)
            if disabled is True or current_memory is None:
                return func(*args, **kwargs)

            # if cache is enabled, we compute the md5 hash of the concatenated source codes
            # of all the dependencies.
            concatenated_source_code = ''
            dependencies = resolve_dependencies(func.__name__, cache.dependencies)
            for func_name in dependencies:
                function = cache.funcs_references[func_name]
                if function is None:
                    raise Exception(f"Can't get source code of function '{func_name}'")
                source_code = get_func_sourcecode(function)
                concatenated_source_code += source_code
            md5_hash = md5(str.encode(concatenated_source_code)).hexdigest()

            # Add extra parameters so that joblib checks they didnt have changed:
            tmp_extra_kwargs = {
                '__func_dependencies_hash__': md5_hash,
                '__original_func_name__': func.__name__,
            }

            if check_param is True:
                kwargs.update(tmp_extra_kwargs)

                if applied_on_method:
                    self_arg, args = args[0], args[1:]

                @wraps(func)
                def f(*args, **kwargs):
                    # delete the extra parameters that the underlying function doesnt expect:
                    for k in tmp_extra_kwargs.keys():
                        del kwargs[k]

                    if applied_on_method:
                        args = (self_arg,) + args
                    return func(*args, **kwargs)

                f = current_memory.cache(f)
                result = f(*args, **kwargs)
            else:
                if isinstance(check_param, str):
                    check_only_param_value = get_param_value_from_func_call(
                        param_name=check_param,
                        func=func,
                        call_args=args,
                        call_kwargs=kwargs,
                    )
                    tmp_extra_kwargs['__check_only__'] = check_only_param_value

                @wraps(func)
                def f(**tmp_extra_kwargs):
                    return func(*args, **kwargs)

                f = current_memory.cache(f)
                result = f(**tmp_extra_kwargs)

            if limit is not None:
                clean_cachedir_old_entries(f.store_backend, func.__name__, limit)

            return result

        return wrapper

    return decorator
def cache(  # noqa: C901
    requires=None,
    disabled=False,
    applied_on_method=False,
    check_param=True,
    limit=None
):
    """ Avoid to recompute a function if its parameters and its source code doesnt have changed.

        Args:
            requires: list of dependencies (functions or function names)
            disabled (bool): disable the cache mecanism for this function (useful if you
                                 only want to use the dependency mecanism)
            applied_on_method (bool): ignore the first argument (useful to ignore "self")
            check_param (True, False or a str): the name of the parameter to check.
                                                    False to not check any of them.
                                                    True (default) to check all of them.
            limit (int or None): number of cache entries to keep (no limit by default)
    """
    if not requires:
        requires = []
    elif isinstance(requires, collections.Callable):
        requires = [requires]

    if not isinstance(check_param, (bool, str)):
        raise TypeError("'check_param' must be a str (name of the param to check) or a bool")
    if limit is not None and not isinstance(limit, int):
        raise TypeError("'limit' must be an int (number of cache entries to keep) or None")

    # We keep data in the function attributes so that this data
    # is not erased between two calls:
    if not hasattr(cache, 'funcs_references'):
        cache.funcs_references = {}  # dict of {function_name -> function_object (or None)}
    if not hasattr(cache, 'dependencies'):
        cache.dependencies = {}  # dict of {function_name -> [list of function names]}
    if not hasattr(cache, 'memories'):
        cache.memories = {}  # dict of {thread_id -> joblib.Memory object}

    def decorator(func):
        """ This code is executed when the augment module is read (when decorator is applied).
            Here we populate cache.funcs_references and cache.dependencies to use them later. """
        cache.funcs_references[func.__name__] = get_orig_function(func)
        dependencies_names = []
        for requirement in requires:
            if isinstance(requirement, collections.Callable):
                req_name = requirement.__name__
                cache.funcs_references[req_name] = get_orig_function(requirement)
            elif requirement not in cache.funcs_references:
                req_name = requirement
                cache.funcs_references[req_name] = None
            dependencies_names.append(req_name)

        cache.dependencies[func.__name__] = dependencies_names

        @wraps(func)
        def wrapper(*args, **kwargs):
            """ This code is executed when a decorated function is actually executed.
                It uses the previously built dependency tree (see above). """
            current_memory = cache.memories.get(current_thread().name)
            if disabled is True or current_memory is None:
                return func(*args, **kwargs)

            # if cache is enabled, we compute the md5 hash of the concatenated source codes
            # of all the dependencies.
            concatenated_source_code = ''
            dependencies = resolve_dependencies(func.__name__, cache.dependencies)
            for func_name in dependencies:
                function = cache.funcs_references[func_name]
                if function is None:
                    raise Exception(f"Can't get source code of function '{func_name}'")
                source_code = get_func_sourcecode(function)
                concatenated_source_code += source_code
            md5_hash = md5(str.encode(concatenated_source_code)).hexdigest()

            # Add extra parameters so that joblib checks they didnt have changed:
            tmp_extra_kwargs = {
                '__func_dependencies_hash__': md5_hash,
                '__original_func_name__': func.__name__,
            }

            if check_param is True:
                kwargs.update(tmp_extra_kwargs)

                if applied_on_method:
                    self_arg, args = args[0], args[1:]

                @wraps(func)
                def f(*args, **kwargs):
                    # delete the extra parameters that the underlying function doesnt expect:
                    for k in tmp_extra_kwargs.keys():
                        del kwargs[k]

                    if applied_on_method:
                        args = (self_arg,) + args
                    return func(*args, **kwargs)

                f = current_memory.cache(f)
                result = f(*args, **kwargs)
            else:
                if isinstance(check_param, str):
                    check_only_param_value = get_param_value_from_func_call(
                        param_name=check_param,
                        func=func,
                        call_args=args,
                        call_kwargs=kwargs,
                    )
                    tmp_extra_kwargs['__check_only__'] = check_only_param_value

                @wraps(func)
                def f(**tmp_extra_kwargs):
                    return func(*args, **kwargs)

                f = current_memory.cache(f)
                result = f(**tmp_extra_kwargs)

            if limit is not None:
                clean_cachedir_old_entries(f.store_backend, func.__name__, limit)

            return result

        return wrapper

    return decorator
Avoid to recompute a function if its parameters and its source code doesnt have changed.

        Args:
            requires: list of dependencies (functions or function names)
            disabled (bool): disable the cache mecanism for this function (useful if you
                                 only want to use the dependency mecanism)
            applied_on_method (bool): ignore the first argument (useful to ignore "self")
            check_param (True, False or a str): the name of the parameter to check.
                                                    False to not check any of them.
                                                    True (default) to check all of them.
            limit (int or None): number of cache entries to keep (no limit by default)