in data_validation/cli_tools.py [0:0]
def get_filters(filter_value: str) -> List[Dict]:
"""Returns filters for source and target from --filters argument.
A filter is the condition that is used in a SQL WHERE clause.
If only one filter is specified, it applies to both source and target
For a doc on regular expression for filters see docs/internal/filters_regex.md
"""
filters = util.split_not_in_quotes(filter_value, ":")
if len(filters) not in (1, 2):
raise argparse.ArgumentTypeError("Unable to parse filter arguments.")
filters = [_.strip() for _ in filters]
if len(filters) == 1:
if not filters[0]:
raise argparse.ArgumentTypeError("Empty string not allowed in filter")
filter_dict = {
"type": "custom",
"source": filters[0],
"target": filters[0],
}
elif len(filters) == 2:
if not filters[0] or not filters[1]:
raise argparse.ArgumentTypeError("Empty string not allowed in filter")
filter_dict = {
"type": "custom",
"source": filters[0],
"target": filters[1],
}
filter_config = [
filter_dict,
]
return filter_config