SQLDatabase

class langchain_community.utilities.sql_database.SQLDatabase(engine: Engine, schema: str | None = None, metadata: MetaData | None = None, ignore_tables: List[str] | None = None, include_tables: List[str] | None = None, sample_rows_in_table_info: int = 3, indexes_in_table_info: bool = False, custom_table_info: dict | None = None, view_support: bool = False, max_string_length: int = 300, lazy_table_reflection: bool = False)

SQLAlchemy wrapper around a database.

Create engine from database URI.

Attributes

dialect	Return string representation of dialect to use.
table_info	Information about all tables in the database.

Methods

__init__(engine[, schema, metadata, ...])	Create engine from database URI.
from_cnosdb([url, user, password, tenant, ...])	Class method to create an SQLDatabase instance from a CnosDB connection.
from_databricks(catalog, schema[, host, ...])	Class method to create an SQLDatabase instance from a Databricks connection.
from_uri(database_uri[, engine_args])	Construct a SQLAlchemy engine from URI.
get_context()	Return db context that you may want in agent prompt.
get_table_info([table_names])	Get information about specified tables.
get_table_info_no_throw([table_names])	Get information about specified tables.
get_table_names()
get_usable_table_names()	Get names of tables available.
run(command[, fetch, include_columns, ...])	Execute a SQL command and return a string representing the results.
run_no_throw(command[, fetch, ...])	Execute a SQL command and return a string representing the results.

Parameters:

• engine (Engine)

• schema (Optional[str])

• metadata (Optional[MetaData])

• ignore_tables (Optional[List[str]])

• include_tables (Optional[List[str]])

• sample_rows_in_table_info (int)

• indexes_in_table_info (bool)

• custom_table_info (Optional[dict])

• view_support (bool)

• max_string_length (int)

• lazy_table_reflection (bool)

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

Create engine from database URI.

Parameters:

• engine (Engine)

• schema (str | None)

• metadata (MetaData | None)

• ignore_tables (List[str] | None)

• include_tables (List[str] | None)

• sample_rows_in_table_info (int)

• indexes_in_table_info (bool)

• custom_table_info (dict | None)

• view_support (bool)

• max_string_length (int)

• lazy_table_reflection (bool)

classmethod from_cnosdb(url: str = '127.0.0.1:8902', user: str = 'root', password: str = '', tenant: str = 'cnosdb', database: str = 'public') → SQLDatabase

Class method to create an SQLDatabase instance from a CnosDB connection. This method requires the ‘cnos-connector’ package. If not installed, it can be added using pip install cnos-connector.

Parameters:

• url (str) – The HTTP connection host name and port number of the CnosDB service, excluding “http://” or “https://”, with a default value of “127.0.0.1:8902”.

• user (str) – The username used to connect to the CnosDB service, with a default value of “root”.

• password (str) – The password of the user connecting to the CnosDB service, with a default value of “”.

• tenant (str) – The name of the tenant used to connect to the CnosDB service, with a default value of “cnosdb”.

• database (str) – The name of the database in the CnosDB tenant.

Returns:

An instance of SQLDatabase configured with the provided CnosDB connection details.

Return type:

SQLDatabase

classmethod from_databricks(catalog: str, schema: str, host: str | None = None, api_token: str | None = None, warehouse_id: str | None = None, cluster_id: str | None = None, engine_args: dict | None = None, **kwargs: Any) → SQLDatabase

Class method to create an SQLDatabase instance from a Databricks connection. This method requires the ‘databricks-sql-connector’ package. If not installed, it can be added using pip install databricks-sql-connector.

Parameters:

• catalog (str) – The catalog name in the Databricks database.

• schema (str) – The schema name in the catalog.

• host (Optional[str]) – The Databricks workspace hostname, excluding ‘https://’ part. If not provided, it attempts to fetch from the environment variable ‘DATABRICKS_HOST’. If still unavailable and if running in a Databricks notebook, it defaults to the current workspace hostname. Defaults to None.

• api_token (Optional[str]) – The Databricks personal access token for accessing the Databricks SQL warehouse or the cluster. If not provided, it attempts to fetch from ‘DATABRICKS_TOKEN’. If still unavailable and running in a Databricks notebook, a temporary token for the current user is generated. Defaults to None.

• warehouse_id (Optional[str]) – The warehouse ID in the Databricks SQL. If provided, the method configures the connection to use this warehouse. Cannot be used with ‘cluster_id’. Defaults to None.

• cluster_id (Optional[str]) – The cluster ID in the Databricks Runtime. If provided, the method configures the connection to use this cluster. Cannot be used with ‘warehouse_id’. If running in a Databricks notebook and both ‘warehouse_id’ and ‘cluster_id’ are None, it uses the ID of the cluster the notebook is attached to. Defaults to None.

• engine_args (Optional[dict]) – The arguments to be used when connecting Databricks. Defaults to None.

• **kwargs (Any) – Additional keyword arguments for the from_uri method.

Returns:

An instance of SQLDatabase configured with the provided

Databricks connection details.

Return type:

SQLDatabase

Raises:

ValueError – If ‘databricks-sql-connector’ is not found, or if both ‘warehouse_id’ and ‘cluster_id’ are provided, or if neither ‘warehouse_id’ nor ‘cluster_id’ are provided and it’s not executing inside a Databricks notebook.

classmethod from_uri(database_uri: str | URL, engine_args: dict | None = None, **kwargs: Any) → SQLDatabase

Construct a SQLAlchemy engine from URI.

Parameters:

• database_uri (str | URL)

• engine_args (dict | None)

• kwargs (Any)

Return type:

SQLDatabase

get_context() → Dict[str, Any]

Return db context that you may want in agent prompt.

Return type:

Dict[str, Any]

get_table_info(table_names: List[str] | None = None) → str

Get information about specified tables.

Follows best practices as specified in: Rajkumar et al, 2022 (https://arxiv.org/abs/2204.00498)

If sample_rows_in_table_info, the specified number of sample rows will be appended to each table description. This can increase performance as demonstrated in the paper.

Parameters:

table_names (List[str] | None)

Return type:

str

get_table_info_no_throw(table_names: List[str] | None = None) → str

Get information about specified tables.

Follows best practices as specified in: Rajkumar et al, 2022 (https://arxiv.org/abs/2204.00498)

If sample_rows_in_table_info, the specified number of sample rows will be appended to each table description. This can increase performance as demonstrated in the paper.

Parameters:

table_names (List[str] | None)

Return type:

str

get_table_names() → Iterable[str]

Deprecated since version langchain-community==0.0.1: Use get_usable_table_names() instead. It will not be removed until langchain-community==1.0.

Get names of tables available.

Return type:

Iterable[str]

get_usable_table_names() → Iterable[str]

Get names of tables available.

Return type:

Iterable[str]

run(command: str | Executable, fetch: Literal['all', 'one', 'cursor'] = 'all', include_columns: bool = False, *, parameters: Dict[str, Any] | None = None, execution_options: Dict[str, Any] | None = None) → str | Sequence[Dict[str, Any]] | Result[Any]

Execute a SQL command and return a string representing the results.

If the statement returns rows, a string of the results is returned. If the statement returns no rows, an empty string is returned.

Parameters:

• command (str | Executable)

• fetch (Literal['all', 'one', 'cursor'])

• include_columns (bool)

• parameters (Dict[str, Any] | None)

• execution_options (Dict[str, Any] | None)

Return type:

str | Sequence[Dict[str, Any]] | Result[Any]

run_no_throw(command: str, fetch: Literal['all', 'one'] = 'all', include_columns: bool = False, *, parameters: Dict[str, Any] | None = None, execution_options: Dict[str, Any] | None = None) → str | Sequence[Dict[str, Any]] | Result[Any]

Execute a SQL command and return a string representing the results.

If the statement returns rows, a string of the results is returned. If the statement returns no rows, an empty string is returned.

If the statement throws an error, the error message is returned.

Parameters:

• command (str)

• fetch (Literal['all', 'one'])

• include_columns (bool)

• parameters (Dict[str, Any] | None)

• execution_options (Dict[str, Any] | None)

Return type:

str | Sequence[Dict[str, Any]] | Result[Any]

Examples using SQLDatabase

• Build a Question/Answering system over SQL data

• CnosDB

• How to better prompt when doing SQL question-answering

• How to deal with large databases when doing SQL question-answering

• How to do query validation as part of SQL question-answering

• How to do question answering over CSVs

• Rebuff

• SQLDatabase Toolkit