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()

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

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

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[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 | 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