def cot_executor()

in nl2sql_library/app.py [0:0]

• 49 lines of code
• 5 McCabe index (conditional complexity)

def cot_executor():
    """
    Invokes the Chain of Thought executor
    """
    question = request.json["question"]
    execute_sql = request.json["execute_sql"]

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

    try:
        logger.info("CoT initialising the class")
        nle = NL2SQL_Executors()
        res_id, sql, df = nle.cot_executor(
            question=question, data_dict=data_dictionary_read
        )
        sql_result = ""
        response_string = {
            "result_id": res_id,
            "generated_query": sql,
            "sql_result": sql_result,
            "error_msg": "",
        }
        sql2 = "\t".join([line.strip() for line in sql])
        log_sql(res_id, question, str(sql2), "CoT 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,
                    "df": df.to_json(),
                    "error_msg": "",
                }
            except RuntimeError:
                print("internal try catch")
                response_string = {
                    "result_id": res_id,
                    "generated_query": sql,
                    "sql_result": sql_result,
                    "df": df.to_json(),
                    "error_msg": "",
                }
    except RuntimeError:
        logger.debug(f"CoT SQL generation unsuccessful : [{question}]")
        response_string = {
            "result_id": 0,
            "generated_query": "",
            "sql_result": "",
            "error_msg": "Error encountered in CoT executor",
        }

    return response_string