Implementation:Explodinggradients Ragas Text2SQL DB Utils

Field	Value
source	Explodinggradients_Ragas|https://github.com/explodinggradients/ragas
domains	Examples, Text2SQL
last_updated	2026-02-10 00:00 GMT

Overview

A SQLite database utility module providing the SQLiteDB class and convenience functions for executing SQL queries and retrieving schema information in Text-to-SQL evaluation workflows.

Description

The db_utils.py module provides a simple, evaluation-focused interface for interacting with SQLite databases. The central SQLiteDB class manages database connections and provides methods for executing SELECT queries with automatic SQL normalization (quote fixing, whitespace cleanup, date function replacement, case-insensitive comparisons) and retrieving schema information. A security check restricts execution to SELECT statements only. The module also exposes two convenience functions -- execute_sql and get_database_schema -- that handle connection lifecycle automatically. By default, the module connects to the BookSQL dataset at BookSQL-files/BookSQL/accounting.sqlite. All query results are returned as pandas DataFrames. The module includes a CLI for interactive query testing via --query, --schema, and --tables flags.

Usage

# Execute a query from the command line
python -m ragas_examples.text2sql.db_utils --query "SELECT COUNT(*) FROM master_txn_table"

# Show database schema
python -m ragas_examples.text2sql.db_utils --schema

# List all tables
python -m ragas_examples.text2sql.db_utils --tables

Code Reference

Field	Value
Source Location	examples/ragas_examples/text2sql/db_utils.py
File Size	301 lines

Class: SQLiteDB

class SQLiteDB:
    def __init__(self, db_path: Optional[str] = None)
    def connect(self) -> Tuple[bool, str]
    def disconnect(self) -> None
    def execute_query(self, sql: str, replace_current_date: bool = True,
                      case_insensitive: bool = True) -> Tuple[bool, Union[pd.DataFrame, str]]
    def get_schema_info(self) -> Tuple[bool, Union[pd.DataFrame, str]]
    def get_table_names(self) -> Tuple[bool, Union[list, str]]

Convenience Functions

def execute_sql(sql: str, db_path: Optional[str] = None,
                replace_current_date: bool = True,
                case_insensitive: bool = True) -> Tuple[bool, Union[pd.DataFrame, str]]

def get_database_schema(db_path: Optional[str] = None) -> Tuple[bool, Union[pd.DataFrame, str]]

I/O Contract

Function/Method	Input	Output
SQLiteDB.__init__	db_path: Optional[str] (defaults to BookSQL path)	Initializes connection state
SQLiteDB.connect	None	Tuple[bool, str] -- (success, message)
SQLiteDB.execute_query	sql: str, normalization flags	Tuple[bool, Union[DataFrame, str]] -- (success, result or error)
SQLiteDB.get_schema_info	None	Tuple[bool, Union[DataFrame, str]] -- DataFrame with name, type, sql columns
SQLiteDB.get_table_names	None	Tuple[bool, Union[list, str]] -- list of table names or error
execute_sql	sql: str, db_path, flags	Tuple[bool, Union[DataFrame, str]] -- auto-managed connection
get_database_schema	db_path: Optional[str]	Tuple[bool, Union[DataFrame, str]] -- auto-managed connection

Usage Examples

from ragas_examples.text2sql.db_utils import SQLiteDB, execute_sql

# Using the convenience function
success, result = execute_sql("SELECT COUNT(*) FROM master_txn_table")
if success:
    print(f"Row count: {result.iloc[0, 0]}")

# Using the class directly
db = SQLiteDB("path/to/database.sqlite")
success, msg = db.connect()
if success:
    ok, schema = db.get_schema_info()
    if ok:
        print(schema.to_string())
    db.disconnect()

Related Pages

• Explodinggradients_Ragas_Text2SQL_Data_Utils -- Data preparation utilities that use this module for validation
• Explodinggradients_Ragas_Text2SQL_Validate_Dataset -- Validation pipeline that imports execute_sql from this module