tools/athena/loadtoathena.py
def extract_agg_functions(
    dataset_ids: list, field_info: dict, formulas: dict, data_format: str
) -> dict:
    """
    returns the subset of aggregated functions for the selected datasets

    Example:
    ```
        {
            "bytesPerSecond" : "avg(bytes)",
            "1xx": "sum(case when statusCode / 100 = 1 then 1 else 0 end)",
        }
    ```
    """
    result = {}

    for col in dataset_ids:
        # skip if the column doesn't have any
        # aggregate functions
        if "agg" not in field_info.get(col, {}):
            continue

        for func_name in field_info[col]["agg"]:
            # initialize a flag here
            # this flag is used to validate if this function
            # has any dependent columns and checks if the
            # dependent columns are part of the selected dataset
            toadd_flag = True

            # skip if the function is not defined
            if func_name not in formulas["functions"]:
                continue

            query = formulas["functions"][func_name].get("q", None)
            # skip if no query is defined for the function
            if query is None:
                continue

            # check if the function has any dependent columns
            if "dependent_cols" in formulas["functions"].get(func_name):
                for idx, dep_col in enumerate(
                    formulas["functions"][func_name]["dependent_cols"]
                ):
                    # reset this flag to True for every function
                    toadd_flag = True
                    # check if the dependent columns are
                    # part of the selected dataset
                    if dep_col in dataset_ids:
                        if data_format == "STRUCTURED":
                            query = query.replace(
                                "_var" + str(idx) +
                                "_", field_info[dep_col]["name"]
                            )

                        if data_format == "JSON":
                            # if string, convert to varchar
                            query = query.replace(
                                "_var" + str(idx) + "_",
                                "CAST("
                                + field_info[dep_col]["name"]
                                + " as "
                                + "varchar"
                                if field_info[dep_col]["dtype"] == "string"
                                else field_info[dep_col]["dtype"] + ")",
                            )
                    else:
                        # this means one or more of the dependent columns
                        # are not part of the selected dataset
                        toadd_flag = False

            if toadd_flag:
                if data_format == "STRUCTURED":
                    result[func_name] = query.replace(
                        "_var_", field_info[col]["name"])

                if data_format == "JSON":
                    # if string, convert to varchar
                    result[func_name] = query.replace(
                        "_var_",
                        "CAST(" + field_info[col]["name"] + " as " + "varchar"
                        if field_info[col]["dtype"] == "string"
                        else field_info[col]["dtype"] + ")",
                    )

    # other aggregate functions
    # other functions is presumed to not have any dependent columns
    for func_name in field_info["others"]["agg"]:
        result[func_name] = formulas["functions"][func_name]["q"]

    return result

    returns the subset of aggregated functions for the selected datasets

    Example:
    ```
        {
            "bytesPerSecond" : "avg(bytes)",
            "1xx": "sum(case when statusCode / 100 = 1 then 1 else 0 end)",
        }
    ```
    
returns the subset of aggregated functions for the selected datasets
returns the subset of aggregated functions for the selected datasets
def extract_agg_functions(
    dataset_ids: list, field_info: dict, formulas: dict, data_format: str
) -> dict:
    result = {}
    for col in dataset_ids:
        if "agg" not in field_info.get(col, {}):
            continue
        for func_name in field_info[col]["agg"]:
            toadd_flag = True
            if func_name not in formulas["functions"]:
                continue
            query = formulas["functions"][func_name].get("q", None)
            if query is None:
                continue
            if "dependent_cols" in formulas["functions"].get(func_name):
                for idx, dep_col in enumerate(
                    formulas["functions"][func_name]["dependent_cols"]
                ):
                    toadd_flag = True
                    if dep_col in dataset_ids:
                        if data_format == "STRUCTURED":
                            query = query.replace(
                                "_var" + str(idx) +
                                "_", field_info[dep_col]["name"]
                            )
                        if data_format == "JSON":
                            query = query.replace(
                                "_var" + str(idx) + "_",
                                "CAST("
                                + field_info[dep_col]["name"]
                                + " as "
                                + "varchar"
                                if field_info[dep_col]["dtype"] == "string"
                                else field_info[dep_col]["dtype"] + ")",
                            )
                    else:
                        toadd_flag = False
            if toadd_flag:
                if data_format == "STRUCTURED":
                    result[func_name] = query.replace(
                        "_var_", field_info[col]["name"])
                if data_format == "JSON":
                    result[func_name] = query.replace(
                        "_var_",
                        "CAST(" + field_info[col]["name"] + " as " + "varchar"
                        if field_info[col]["dtype"] == "string"
                        else field_info[col]["dtype"] + ")",
                    )
    for func_name in field_info["others"]["agg"]:
        result[func_name] = formulas["functions"][func_name]["q"]
    return result
def extract_agg_functions ( dataset_ids : list , field_info : dict , formulas : dict , data_format : str ) -> dict : result = { } for col in dataset_ids : if "agg" not in field_info . get ( col , { } ) : continue for func_name in field_info [ col ] [ "agg" ] : toadd_flag = True if func_name not in formulas [ "functions" ] : continue query = formulas [ "functions" ] [ func_name ] . get ( "q" , None ) if query is None : continue if "dependent_cols" in formulas [ "functions" ] . get ( func_name ) : for idx , dep_col in enumerate ( formulas [ "functions" ] [ func_name ] [ "dependent_cols" ] ) : toadd_flag = True if dep_col in dataset_ids : if data_format == "STRUCTURED" : query = query . replace ( "_var" + str ( idx ) + "_" , field_info [ dep_col ] [ "name" ] ) if data_format == "JSON" : query = query . replace ( "_var" + str ( idx ) + "_" , "CAST(" + field_info [ dep_col ] [ "name" ] + " as " + "varchar" if field_info [ dep_col ] [ "dtype" ] == "string" else field_info [ dep_col ] [ "dtype" ] + ")" , ) else : toadd_flag = False if toadd_flag : if data_format == "STRUCTURED" : result [ func_name ] = query . replace ( "_var_" , field_info [ col ] [ "name" ] ) if data_format == "JSON" : result [ func_name ] = query . replace ( "_var_" , "CAST(" + field_info [ col ] [ "name" ] + " as " + "varchar" if field_info [ col ] [ "dtype" ] == "string" else field_info [ col ] [ "dtype" ] + ")" , ) for func_name in field_info [ "others" ] [ "agg" ] : result [ func_name ] = formulas [ "functions" ] [ func_name ] [ "q" ] return result
returns the subset of aggregated functions for the selected datasets
returns the subset of aggregated functions for the selected datasets
"""
    returns the subset of aggregated functions for the selected datasets

    Example:
    ```
        {
            "bytesPerSecond" : "avg(bytes)",
            "1xx": "sum(case when statusCode / 100 = 1 then 1 else 0 end)",
        }
    ```
    """ # skip if the column doesn't have any # aggregate functions # initialize a flag here # this flag is used to validate if this function # has any dependent columns and checks if the # dependent columns are part of the selected dataset # skip if the function is not defined # skip if no query is defined for the function # check if the function has any dependent columns # reset this flag to True for every function # check if the dependent columns are # part of the selected dataset # if string, convert to varchar # this means one or more of the dependent columns # are not part of the selected dataset # if string, convert to varchar # other aggregate functions # other functions is presumed to not have any dependent columns
dataset_ids
list
field_info
dict
formulas
dict
data_format
str
examples
None