in data_validation/__main__.py [0:0]
def get_aggregate_config(args, config_manager: ConfigManager):
"""Return list of formatted aggregation objects.
Args:
config_manager (ConfigManager): Validation config manager instance.
"""
aggregate_configs = [config_manager.build_config_count_aggregate()]
supported_data_types = [
"float64",
"float32",
"int8",
"int16",
"int32",
"int64",
"decimal",
"!float64",
"!float32",
"!int8",
"!int16",
"!int32",
"!int64",
"!decimal",
]
uuid_types = ["uuid", "!uuid"]
if args.wildcard_include_string_len:
supported_data_types.extend(["string", "!string"])
if args.wildcard_include_timestamp:
supported_data_types.extend(["timestamp", "!timestamp", "date", "!date"])
cast_to_bigint = True if args.cast_to_bigint else False
if args.count:
col_args = None if args.count == "*" else cli_tools.get_arg_list(args.count)
aggregate_configs += config_manager.build_config_column_aggregates(
"count", col_args, args.exclude_columns, None, cast_to_bigint=cast_to_bigint
)
if args.sum:
col_args = None if args.sum == "*" else cli_tools.get_arg_list(args.sum)
aggregate_configs += config_manager.build_config_column_aggregates(
"sum",
col_args,
args.exclude_columns,
supported_data_types,
cast_to_bigint=cast_to_bigint,
)
if args.avg:
col_args = None if args.avg == "*" else cli_tools.get_arg_list(args.avg)
aggregate_configs += config_manager.build_config_column_aggregates(
"avg",
col_args,
args.exclude_columns,
supported_data_types,
cast_to_bigint=cast_to_bigint,
)
if args.min:
col_args = None if args.min == "*" else cli_tools.get_arg_list(args.min)
aggregate_configs += config_manager.build_config_column_aggregates(
"min",
col_args,
args.exclude_columns,
supported_data_types + uuid_types,
cast_to_bigint=cast_to_bigint,
)
if args.max:
col_args = None if args.max == "*" else cli_tools.get_arg_list(args.max)
aggregate_configs += config_manager.build_config_column_aggregates(
"max",
col_args,
args.exclude_columns,
supported_data_types + uuid_types,
cast_to_bigint=cast_to_bigint,
)
if args.bit_xor:
col_args = None if args.bit_xor == "*" else cli_tools.get_arg_list(args.bit_xor)
aggregate_configs += config_manager.build_config_column_aggregates(
"bit_xor",
col_args,
args.exclude_columns,
supported_data_types,
cast_to_bigint=cast_to_bigint,
)
if args.std:
col_args = None if args.std == "*" else cli_tools.get_arg_list(args.std)
aggregate_configs += config_manager.build_config_column_aggregates(
"std",
col_args,
args.exclude_columns,
supported_data_types,
cast_to_bigint=cast_to_bigint,
)
return aggregate_configs