Spaces:
Sleeping
Sleeping
File size: 1,697 Bytes
b3eb06a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
import sys
import shutil
import os
import uuid
import duckdb
from duckdb import ParserException, SyntaxException, BinderException, CatalogException
TMP_DIR = "tmp"
class WithDuckDBConnectionInTmpDir(object):
def __init__(self):
self.tmp_dir = TMP_DIR + str(uuid.uuid1())
os.makedirs(self.tmp_dir)
self.original_wd = os.getcwd()
def __enter__(self):
os.chdir(self.tmp_dir)
self.con = duckdb.connect()
self.con.execute("SET enable_external_access=False")
return self.con
def __exit__(self, *args):
self.con.close()
os.chdir(self.original_wd)
shutil.rmtree(self.tmp_dir)
def validate_query(query, schemas):
try:
with WithDuckDBConnectionInTmpDir() 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
else:
print(message)
if __name__ == '__main__':
if len(sys.argv) > 2:
validate_query(sys.argv[1], sys.argv[2])
else:
print("No query provided.") |