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, 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] Get names of tables available.

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: 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, lazy_table_reflection: bool = False)[source]

Create engine from database URI.

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

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.

Parameters

  • database_uri (str) –

  • engine_args (Optional[dict]) –

  • 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: 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.

Parameters

table_names (Optional[List[str]]) –

Return type

str

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.

Parameters

table_names (Optional[List[str]]) –

Return type

str

get_table_names() → Iterable[str][source]

[Deprecated] Get names of tables available.

Notes

Deprecated since version 0.0.1: Use get_usable_table_name instead.

Return type

Iterable[str]

get_usable_table_names() → Iterable[str][source]

Get names of tables available.

Return type

Iterable[str]

run(command: Union[str, Executable], fetch: Literal['all', 'one', 'cursor'] = 'all', include_columns: bool = False, *, parameters: Optional[Dict[str, Any]] = None, execution_options: Optional[Dict[str, Any]] = None) → Union[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 (Union[str, Executable]) –

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

  • include_columns (bool) –

  • parameters (Optional[Dict[str, Any]]) –

  • execution_options (Optional[Dict[str, Any]]) –

Return type

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

run_no_throw(command: str, fetch: Literal['all', 'one'] = 'all', include_columns: bool = False, *, parameters: Optional[Dict[str, Any]] = None, execution_options: Optional[Dict[str, Any]] = None) → Union[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 (Optional[Dict[str, Any]]) –

  • execution_options (Optional[Dict[str, Any]]) –

Return type

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

Examples using SQLDatabase