in evalbench/databases/sqlite.py [0:0]
def _run_execute(query: str, eval_query: Optional[str] = None, rollback=False):
result: List = []
eval_result: List = []
error = None
try:
with self.engine.connect() as connection:
with connection.begin() as transaction:
result = self._execute_queries(connection, query)
if eval_query:
eval_result = self._execute_queries(connection, eval_query)
if batch_commands and len(batch_commands) > 0:
for command in batch_commands:
connection.execute(text(command))
if rollback:
transaction.rollback()
except Exception as e:
error = str(e)
if "database is locked" in error:
raise ResourceExhaustedError(
"SQLite Database is locked, retry later"
) from e
elif "disk I/O error" in error:
raise ResourceExhaustedError(
"Disk I/O error occurred, check storage"
) from e
return result, eval_result, error