in data_validation/raw_query.py [0:0]
def print_raw_query_output(query_output: list, format: str = consts.FORMAT_TYPE_PYTHON):
"""Print a query resultset avoiding SQLAlchemy "... (nn characters truncated) ..." behaviour.
Args:
query_output (list): A set of rows from a SQLAlchemy query.
format (str): Output format:
FORMAT_TYPE_PYTHON: Legacy format matching previous behaviour, simple print the Python object.
FORMAT_TYPE_MINIMAL: Attempt to minimize additional noise for bare minimum output.
This is useful for a specific customer who are using DVT raw queries to
generate other DVT commands and don't want any escaping or extra parentheses
in the output.
"""
def row_to_tuple(row) -> tuple:
"""This prevents SQLAlchemy string truncation inside Row() objects by first converting them to a tuple."""
if isinstance(row, (Row, list)):
return tuple(row)
else:
return row
sanitized_output = [row_to_tuple(_) for _ in query_output or []]
if format == consts.FORMAT_TYPE_MINIMAL:
for row in _minimize_raw_query_output(sanitized_output):
print(row)
else:
print(sanitized_output)