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
def cache ( # noqa: C901 requires = None , disabled = False , applied_on_method = False , check_param = True , limit = None ) : 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)
Avoid to recompute a function if its parameters and its source code doesnt have changed .