in data_validation/__main__.py [0:0]
def _get_calculated_config(args, config_manager: ConfigManager) -> List[dict]:
"""Return list of formatted calculated objects.
Args:
config_manager(ConfigManager): Validation config manager instance.
"""
calculated_configs = []
fields = []
if config_manager.hash:
col_list = (
None
if config_manager.hash == "*"
else cli_tools.get_arg_list(config_manager.hash)
)
fields = config_manager.build_dependent_aliases(
"hash", col_list, args.exclude_columns
)
elif config_manager.concat:
col_list = (
None
if config_manager.concat == "*"
else cli_tools.get_arg_list(config_manager.concat)
)
fields = config_manager.build_dependent_aliases(
"concat", col_list, args.exclude_columns
)
if len(fields) > 0:
max_depth = max([x["depth"] for x in fields])
else:
max_depth = 0
for field in fields:
calculated_configs.append(
config_manager.build_config_calculated_fields(
field["source_reference"],
field["target_reference"],
field["calc_type"],
field["name"],
field["depth"],
custom_params=field.get("calc_params"),
)
)
if config_manager.hash:
config_manager.append_comparison_fields(
config_manager.build_config_comparison_fields(
["hash__all"], depth=max_depth
)
)
elif config_manager.concat:
config_manager.append_comparison_fields(
config_manager.build_config_comparison_fields(
["concat__all"], depth=max_depth
)
)
return calculated_configs