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.") |