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)

SQLAlchemy对数据库的封装。

从数据库URI创建引擎。

Attributes

dialect	返回要使用的方言的字符串表示。
table_info	数据库中所有表的信息。

Methods

__init__(engine[, schema, metadata, ...])	从数据库URI创建引擎。
from_cnosdb([url, user, password, tenant, ...])	用于从CnosDB连接创建SQLDatabase实例的类方法。 此方法需要'cnos-connector'包。如果未安装，可以使用`pip install cnos-connector`添加。
from_databricks(catalog, schema[, host, ...])	用于从Databricks连接创建SQLDatabase实例的类方法。 此方法需要'databricks-sql-connector'包。如果未安装， 可以使用`pip install databricks-sql-connector`进行添加。
from_uri(database_uri[, engine_args])	从URI构建一个SQLAlchemy引擎。
get_context()	返回可能在代理提示中需要的数据库上下文。
get_table_info([table_names])	获取指定表的信息。
get_table_info_no_throw([table_names])	获取指定表的信息。
get_table_names()	[Deprecated] 获取可用表的名称。
get_usable_table_names()	获取可用表的名称。
run(command[, fetch, include_columns, ...])	执行一个SQL命令并返回表示结果的字符串。
run_no_throw(command[, fetch, ...])	执行一个SQL命令并返回表示结果的字符串。

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)

从数据库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

用于从CnosDB连接创建SQLDatabase实例的类方法。 此方法需要’cnos-connector’包。如果未安装，可以使用`pip install cnos-connector`添加。

参数：

url（str）：CnosDB服务的HTTP连接主机名和端口号，不包括”http://”或”https://”，默认值为”127.0.0.1:8902”。 user（str）：用于连接到CnosDB服务的用户名，默认值为”root”。 password（str）：连接到CnosDB服务的用户的密码，默认值为””。 tenant（str）：用于连接到CnosDB服务的租户名称，默认值为”cnosdb”。 database（str）：CnosDB租户中数据库的名称。

返回：

SQLDatabase：使用提供的CnosDB连接详细信息配置的SQLDatabase实例。

Parameters

• url (str) –

• user (str) –

• password (str) –

• tenant (str) –

• database (str) –

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

用于从Databricks连接创建SQLDatabase实例的类方法。 此方法需要’databricks-sql-connector’包。如果未安装， 可以使用`pip install databricks-sql-connector`进行添加。

参数：

catalog（str）：Databricks数据库中的目录名称。 schema（str）：目录中的模式名称。 host（可选[str]）：Databricks工作区主机名，不包括

‘https://’部分。如果未提供，则尝试从 环境变量’DATABRICKS_HOST’中获取。如果仍然不可用且 在Databricks笔记本中运行，则默认为当前工作区 主机名。默认为None。

api_token（可选[str]）：用于

访问Databricks SQL仓库或集群的Databricks个人访问令牌。如果未提供， 尝试从’DATABRICKS_TOKEN’中获取。如果仍然不可用 并在Databricks笔记本中运行，则为当前用户生成临时令牌。 默认为None。

warehouse_id（可选[str]）：Databricks SQL中的仓库ID。如果

提供，则该方法配置连接以使用此仓库。 不能与’cluster_id’一起使用。默认为None。

cluster_id（可选[str]）：Databricks Runtime中的集群ID。如果

提供，则该方法配置连接以使用此集群。 不能与’warehouse_id’一起使用。如果在Databricks笔记本中运行 并且’warehouse_id’和’cluster_id’都为None，则使用笔记本所附加的 集群的ID。默认为None。

engine_args（可选[dict]）：连接时要使用的参数

Databricks。默认为None。

**kwargs（Any）：`from_uri`方法的其他关键字参数。

返回：

SQLDatabase：使用提供的Databricks连接详细信息配置的

SQLDatabase实例。

引发：

ValueError：如果未找到’databricks-sql-connector’，或者如果

提供了’warehouse_id’和’cluster_id’，或者如果既没有 提供’warehouse_id’也没有提供’cluster_id’且未在 Databricks笔记本中执行。

Parameters

• catalog (str) –

• schema (str) –

• host (Optional[str]) –

• api_token (Optional[str]) –

• warehouse_id (Optional[str]) –

• cluster_id (Optional[str]) –

• engine_args (Optional[dict]) –

• kwargs (Any) –

Return type

SQLDatabase

classmethod from_uri(database_uri: str, engine_args: Optional[dict] = None, **kwargs: Any) → SQLDatabase

从URI构建一个SQLAlchemy引擎。

Parameters

• database_uri (str) –

• engine_args (Optional[dict]) –

• kwargs (Any) –

Return type

SQLDatabase

get_context() → Dict[str, Any]

返回可能在代理提示中需要的数据库上下文。

Return type

Dict[str, Any]

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

获取指定表的信息。

遵循Rajkumar等人在2022年指定的最佳实践（https://arxiv.org/abs/2204.00498）。

如果`sample_rows_in_table_info`，则将指定数量的样本行附加到每个表描述中。 正如论文中所示，这可以提高性能。

Parameters

table_names (Optional[List[str]]) –

Return type

str

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

获取指定表的信息。

遵循Rajkumar等人在2022年指定的最佳实践（https://arxiv.org/abs/2204.00498）。

如果`sample_rows_in_table_info`，则将指定数量的样本行附加到每个表描述中。 正如论文中所示，这可以提高性能。

Parameters

table_names (Optional[List[str]]) –

Return type

str

get_table_names() → Iterable[str]

[Deprecated] 获取可用表的名称。

Notes

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

Return type

Iterable[str]

get_usable_table_names() → Iterable[str]

获取可用表的名称。

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

执行一个SQL命令并返回表示结果的字符串。

如果语句返回行，则返回结果的字符串。 如果语句没有返回行，则返回空字符串。

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

执行一个SQL命令并返回表示结果的字符串。

如果语句返回行，则返回结果的字符串。 如果语句未返回行，则返回空字符串。

如果语句抛出错误，则返回错误消息。

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

• !pip3 install rebuff openai -U

• CnosDB

• In order to build a selectable on SA’s Core API, you need a table definition.

• Uncomment the below to use LangSmith. Not required.

• Using LangSmith is recommended but not required. Uncomment below lines to use.

• os.environ[“OPENAI_API_KEY”] = getpass.getpass()

• sql_database.md