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)[source]#

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)[source]#

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[source]#

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

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[source]#

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][source]#

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[source]#

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[source]#

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][source]#

Deprecated since version langchain-community==0.0.1: Use get_usable_table_names() instead.

Get names of tables available.

Return type:: Iterable[str]

get_usable_table_names() → Iterable[str][source]#

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][source]#

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][source]#

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