langchain_community.utilities.sql_database.SQLDatabase¶

class langchain_community.utilities.sql_database.SQLDatabase(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)[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_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 names of tables available.

get_usable_table_names()

Get names of tables available.

run(command[, fetch])

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.
__init__(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)[source]¶

Create engine from database URI.

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

classmethod from_databricks(catalog: str, schema: str, host: Optional[str] = None, api_token: Optional[str] = None, warehouse_id: Optional[str] = None, cluster_id: Optional[str] = None, engine_args: Optional[dict] = None, **kwargs: Any) → SQLDatabase[source]¶

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, engine_args: Optional[dict] = None, **kwargs: Any) → SQLDatabase[source]¶

Construct a SQLAlchemy engine from URI.

get_table_info(table_names: Optional[List[str]] = 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.

get_table_info_no_throw(table_names: Optional[List[str]] = 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.

get_table_names() → Iterable[str][source]¶

Get names of tables available.

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

Get names of tables available.

run(command: str, fetch: Union[Literal['all'], Literal['one']] = 'all') → str[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.

run_no_throw(command: str, fetch: Union[Literal['all'], Literal['one']] = 'all') → str[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.

Examples using SQLDatabase¶