I have a list that contains SQL code that can be executed in an external Trino CLI. So for instance, my nested list would look like:
sql = []
sql = [['test 1', 'SELECT * FROM a.testtable1'],['test 2', 'SELECT * FROM a.testtable1']]
This simple loop detects if there's a syntax error:
sql_results = []
for l in sql:
sql_code = l[1]
try:
cur.execute(sql_code
rows = cur.fetchall()
except Exception as e:
status = str(e)
status = 'OK' if len(rows) > 1
sql_results.append([l[0],sql_code,status])
It works good, but sometimes the queries take too long and kill the process. Knowing that if one query lasts more than 3 seconds in its execution, then it means the syntax is OK (and I'm only interested in checking the syntax, and NOT in the result of the query) I'd like to add a time validation. Something like: If the SQL execution lasts more than 3 seconds, then kill it and status = 'OK'
I tried this using time:
import time
sql_results = []
for l in sql:
sql_code = l[1]
try:
timeout = time.time() + 2
cur.execute(sql_code)
rows = cur.fetchall()
except Exception as e:
status = str(e)
status = 'OK' if len(rows) > 1 or time.time() > timeout
sql_results.append([l[0],sql_code,status])
But it does not do much, and I keep getting the occasional timeouts. Any idea?
