Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def roundtrip(sql):
orig_ast = parse_sql(sql)
_remove_stmt_len_and_location(orig_ast)
serialized = RawStream()(Node(orig_ast))
try:
serialized_ast = parse_sql(serialized)
except: # noqa
raise RuntimeError("Could not reparse %r" % serialized)
_remove_stmt_len_and_location(serialized_ast)
assert orig_ast == serialized_ast, "%r != %r" % (sql, serialized)
indented = IndentedStream()(Node(orig_ast))
try:
indented_ast = parse_sql(indented)
except: # noqa
raise RuntimeError("Could not reparse %r" % indented)
_remove_stmt_len_and_location(indented_ast)
assert orig_ast == indented_ast, "%r != %r" % (sql, indented)
_remove_stmt_len_and_location(orig_ast)
serialized = RawStream()(Node(orig_ast))
try:
serialized_ast = parse_sql(serialized)
except: # noqa
raise RuntimeError("Could not reparse %r" % serialized)
_remove_stmt_len_and_location(serialized_ast)
assert orig_ast == serialized_ast, "%r != %r" % (sql, serialized)
indented = IndentedStream()(Node(orig_ast))
try:
indented_ast = parse_sql(indented)
except: # noqa
raise RuntimeError("Could not reparse %r" % indented)
_remove_stmt_len_and_location(indented_ast)
assert orig_ast == indented_ast, "%r != %r" % (sql, indented)
# Run ``pytest -s tests/`` to see the following output
print()
print(indented)
def roundtrip(sql):
orig_ast = parse_sql(sql)
_remove_stmt_len_and_location(orig_ast)
serialized = RawStream()(Node(orig_ast))
try:
serialized_ast = parse_sql(serialized)
except: # noqa
raise RuntimeError("Could not reparse %r" % serialized)
_remove_stmt_len_and_location(serialized_ast)
assert orig_ast == serialized_ast, "%r != %r" % (sql, serialized)
indented = IndentedStream()(Node(orig_ast))
try:
indented_ast = parse_sql(indented)
except: # noqa
raise RuntimeError("Could not reparse %r" % indented)
_remove_stmt_len_and_location(indented_ast)
assert orig_ast == indented_ast, "%r != %r" % (sql, indented)
# Run ``pytest -s tests/`` to see the following output
print()
print(indented)
def workhorse(args):
input = args.infile or sys.stdin
with input:
statement = input.read()
if args.parse_tree or args.plpgsql:
tree = parse_plpgsql(statement) if args.plpgsql else parse_sql(statement)
if args.no_location:
_remove_stmt_len_and_location(tree)
output = args.outfile or sys.stdout
with output:
json.dump(tree, output, sort_keys=True, indent=2)
output.write('\n')
else:
try:
prettified = prettify(
statement,
compact_lists_margin=args.compact_lists_margin,
split_string_literals_threshold=args.split_string_literals,
special_functions=args.special_functions,
comma_at_eoln=args.comma_at_eoln,
semicolon_after_last_statement=args.semicolon_after_last_statement)
except Error as e:
print()
raise SystemExit(e)