Implementation:Run llama Llama index SQLDatabase

Overview

The SQLDatabase class provides a wrapper around SQLAlchemy's engine to interact with SQL databases. It offers methods for executing SQL commands, inserting data, and retrieving schema information. The implementation supports table filtering (include/exclude), sample rows for table descriptions, index information, view support, and custom table info overrides. It is based on the LangChain SQLDatabase utility.

Source File: llama-index-core/llama_index/core/utilities/sql_wrapper.py (248 lines)

Module: llama_index.core.utilities.sql_wrapper

Dependencies

Module	Import
sqlalchemy	MetaData, create_engine, insert, inspect, text
sqlalchemy.engine	Engine
sqlalchemy.exc	OperationalError, ProgrammingError

Class Definition

class SQLDatabase:
    """
    SQL Database.

    This class provides a wrapper around the SQLAlchemy engine to interact
    with a SQL database.
    """

Constructor

def __init__(
    self,
    engine: Engine,
    schema: Optional[str] = None,
    metadata: Optional[MetaData] = None,
    ignore_tables: Optional[List[str]] = None,
    include_tables: Optional[List[str]] = None,
    sample_rows_in_table_info: int = 3,
    indexes_in_table_info: bool = False,
    custom_table_info: Optional[dict] = None,
    view_support: bool = False,
    max_string_length: int = 300,
)

Parameter	Type	Default	Description
engine	Engine	required	SQLAlchemy engine instance
schema	Optional[str]	None	Database schema name
metadata	Optional[MetaData]	None	SQLAlchemy metadata instance; created if not provided
ignore_tables	Optional[List[str]]	None	Tables to exclude; mutually exclusive with include_tables
include_tables	Optional[List[str]]	None	Tables to include; mutually exclusive with ignore_tables
sample_rows_in_table_info	int	3	Number of sample rows to include in table info
indexes_in_table_info	bool	False	Whether to include index information in table info
custom_table_info	Optional[dict]	None	Dictionary of table names to custom info strings
view_support	bool	False	Whether to include database views alongside tables
max_string_length	int	300	Maximum string length for truncating column values in results

Constructor Validation

• Raises ValueError if both include_tables and ignore_tables are specified.
• Raises ValueError if specified include/ignore tables are not found in the database.
• Raises TypeError if sample_rows_in_table_info is not an integer.
• Raises TypeError if custom_table_info is not a dictionary.

Initialization Steps

1. Creates an inspector from the engine.
2. Discovers all tables (and optionally views) in the database.
3. Validates include/ignore table lists against discovered tables.
4. Computes the usable table set.
5. Filters custom table info to only include known tables.
6. Reflects metadata for usable tables.

Factory Method

from_uri

@classmethod
def from_uri(
    cls, database_uri: str, engine_args: Optional[dict] = None, **kwargs: Any
) -> "SQLDatabase"

Constructs a SQLDatabase from a database URI string. Creates a SQLAlchemy engine using create_engine() with optional engine arguments.

Properties

Property	Return Type	Description
engine	Engine	The underlying SQLAlchemy engine
metadata_obj	MetaData	The SQLAlchemy metadata object
dialect	str	The database dialect name (e.g., "sqlite", "postgresql")

Core Methods

get_usable_table_names

def get_usable_table_names(self) -> Iterable[str]

Returns a sorted list of available table names, respecting include_tables and ignore_tables filters.

get_table_columns

def get_table_columns(self, table_name: str) -> List[Any]

Returns column metadata for the specified table using the SQLAlchemy inspector.

get_single_table_info

def get_single_table_info(self, table_name: str) -> str

Returns a formatted string description of a single table including:

• Table name
• Column names and types (with comments if available)
• Table-level comment (if the dialect supports it; gracefully handles NotImplementedError)
• Foreign key relationships in the format: constrained_columns -> referred_table.referred_columns

Output format example:

Table 'users' has columns: id (INTEGER), name (VARCHAR(100)), email (VARCHAR(255)),
and foreign keys: ['department_id'] -> departments.['id'].

insert_into_table

def insert_into_table(self, table_name: str, data: dict) -> None

Inserts a single row into the specified table using SQLAlchemy's insert() construct. The data dictionary maps column names to values.

run_sql

def run_sql(self, command: str) -> Tuple[str, Dict]

Executes a raw SQL command and returns a tuple of:

• A string representation of the results
• A dictionary with keys "result" (list of truncated row tuples) and "col_keys" (list of column names)

Schema prefix handling: If a schema is configured, the method injects the schema name before FROM and JOIN clauses using string replacement.

Error handling: Catches ProgrammingError and OperationalError and re-raises as NotImplementedError with the original SQL and error message.

Result truncation: Each column value in the result is truncated using truncate_word() to respect the max_string_length setting.

truncate_word

def truncate_word(self, content: Any, *, length: int, suffix: str = "...") -> str

Truncates a string to the specified length, breaking at the last word boundary before the limit and appending the suffix. Returns non-string content unchanged. Returns content unchanged if already within the length limit.

Design Notes

• The class is intentionally not a pydantic model -- it wraps SQLAlchemy objects that are not serializable.
• The run_sql schema injection via string replacement (FROM -> FROM schema.) is a simple approach that works for common queries but may not handle all SQL patterns.
• The custom_table_info feature allows overriding auto-generated table descriptions with hand-crafted summaries, useful for complex schemas where auto-generated info may be confusing to an LLM.
• The from_uri factory is the most common construction path for end users.
• The module docstring notes this is "Based on langchain SQLDatabase" with a direct link to the original implementation.

See Also

• TokenCounter -- Another utility class in the same package