toucan_data_sdk/utils/decorators.py
def domain(domain_name):
    """
    Allow to apply a function f(df: DataFrame) -> DataFrame) on dfs by specifying the key
    E.g instead of writing:
        def process_domain1(dfs):
            df = dfs['domain1']
            # actual process
            dfs['domain1'] = df
            return dfs

    You can write:
        @domain('domain1')
        def process_domain1(df):
            #actual process
            return df
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            dfs, *args = args
            if not isinstance(dfs, dict):
                raise TypeError(f'{dfs} is not a dict')
            df = dfs.pop(domain_name)
            df = func(df, *args, **kwargs)
            return {domain_name: df, **dfs}

        return wrapper

    return decorator
def domain(domain_name):
    """
    Allow to apply a function f(df: DataFrame) -> DataFrame) on dfs by specifying the key
    E.g instead of writing:
        def process_domain1(dfs):
            df = dfs['domain1']
            # actual process
            dfs['domain1'] = df
            return dfs

    You can write:
        @domain('domain1')
        def process_domain1(df):
            #actual process
            return df
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            dfs, *args = args
            if not isinstance(dfs, dict):
                raise TypeError(f'{dfs} is not a dict')
            df = dfs.pop(domain_name)
            df = func(df, *args, **kwargs)
            return {domain_name: df, **dfs}

        return wrapper

    return decorator
def domain ( domain_name ) : def decorator ( func ) : @ wraps ( func ) def wrapper ( * args , * * kwargs ) : dfs , * args = args if not isinstance ( dfs , dict ) : raise TypeError ( f'{dfs} is not a dict' ) df = dfs . pop ( domain_name ) df = func ( df , * args , * * kwargs ) return { domain_name : df , * * dfs } return wrapper return decorator
Allow to apply a function f(df: DataFrame) -> DataFrame) on dfs by specifying the key
    E.g instead of writing:
        def process_domain1(dfs):
            df = dfs['domain1']
            # actual process
            dfs['domain1'] = df
            return dfs

    You can write:
        @domain('domain1')
        def process_domain1(df):
            #actual process
            return df
Allow to apply a function f ( df : DataFrame ) - > DataFrame ) on dfs by specifying the key E . g instead of writing : def process_domain1 ( dfs ) : df = dfs [ domain1 ] # actual process dfs [ domain1 ] = df return dfs