validate_sql.py

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