def rag_executor()

in nl2sql_library/app.py [0:0]

• 43 lines of code
• 4 McCabe index (conditional complexity)

def rag_executor():
    """
    Invokes the RAG Executor
    """
    question = request.json["question"]
    execute_sql = request.json["execute_sql"]

    logger.info("RAG SQL Generation engine for question : [{question}]")
    from nl2sql_lib_executors import NL2SQL_Executors

    try:
        nle = NL2SQL_Executors()
        res_id, sql = nle.rag_executor(question=question)
        # res_id, sql = nle.generate_query(question)
        sql_result = ""
        response_string = {
            "result_id": res_id,
            "generated_query": sql,
            "sql_result": sql_result,
            "error_msg": "",
        }
        log_sql(res_id, question, sql, "Rag Executor", execute_sql)
        if execute_sql:
            try:
                result = execute_bq_query(sql)
                sql_result = result2nl(question, result)
                response_string = {
                    "result_id": res_id,
                    "generated_query": sql,
                    "sql_result": sql_result,
                    "error_msg": "",
                }
            except RuntimeError:
                print("internal try catch")
                response_string = {
                    "result_id": res_id,
                    "generated_query": sql,
                    "sql_result": sql_result,
                    "error_msg": "",
                }
    except RuntimeError:
        logger.debug(f"RAG SQL generation unsuccessful : [{question}]")
        response_string = {
            "result_id": 0,
            "generated_query": "",
            "sql_result": "",
            "error_msg": "Error encountered in RAG executor",
        }

    return json.dumps(response_string)