Function exec_on_db

evaluation/test-suite-sql-eval/exec_eval.py:160–168

(
    sqlite_path: str, query: str, process_id: str = "", timeout: int = TIMEOUT
)

async def exec_on_db(
    sqlite_path: str, query: str, process_id: str = "", timeout: int = TIMEOUT
) -> Tuple[str, Any]:
    try:
        return await asyncio.wait_for(exec_on_db_(sqlite_path, query), timeout)
    except asyncio.TimeoutError:
        return ('exception', TimeoutError)
    except Exception as e:
        return ("exception", e)


# postprocess the model predictions to avoid execution errors