validate_sql.py

import sys
import duckdb
from duckdb import ParserException, SyntaxException, BinderException, CatalogException


def validate_query(query, schemas):
    valid = True
    msg = ""
    try:
        print("Running query: ", query)
        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:
        msg = str(e)
        valid = False
    except SyntaxException as e:
        msg = str(e)
        valid = False
    except BinderException as e:
        msg = str(e)
        valid = False
    except CatalogException as e:
        msg = str(e)
        if "but it exists" in msg and "extension" in msg:
            valid = True
            msg = ""
        else:
            valid = False
    except Exception as e:
        msg = str(e)
        valid = True
    return valid, msg


if __name__ == "__main__":
    if len(sys.argv) > 2:
        validate_query(sys.argv[1], sys.argv[2])
    else:
        print("No query provided.")