Spaces:
Running
Running
| import sys | |
| import duckdb | |
| from duckdb import ParserException, SyntaxException, BinderException | |
| def validate_query(query, schemas): | |
| try: | |
| with duckdb.connect(":memory:", config={"enable_external_access": False}) as duckdb_conn: | |
| # register schemas | |
| for schema in schemas.split(";"): | |
| duckdb_conn.execute(schema) | |
| cursor = duckdb_conn.cursor() | |
| cursor.execute(query) | |
| except ParserException as e: | |
| raise e | |
| except SyntaxException as e: | |
| raise e | |
| except BinderException as e: | |
| raise e | |
| except Exception as e: | |
| message = str(e) | |
| if "but it exists" in message and "extension" in message: | |
| print(message) | |
| elif message.startswith("Catalog Error: Table with name"): | |
| raise e | |
| elif "Catalog Error: Table Function with name" in message: | |
| raise e | |
| elif "Catalog Error: Copy Function" in message: | |
| raise e | |
| if __name__ == '__main__': | |
| if len(sys.argv) > 2: | |
| validate_query(sys.argv[1], sys.argv[2]) | |
| else: | |
| print("No query provided.") |