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 包装器。

从数据库 URI 创建引擎。

属性

dialect

返回要使用的方言的字符串表示。

table_info

数据库中所有表的详细信息。

方法

__init__(engine[, schema, metadata, ...])

从数据库 URI 创建引擎。

from_cnosdb([url, user, password, tenant, ...])

从CnosDB连接创建SQLDatabase实例的类方法。

from_databricks(catalog, schema[, host, ...])

从Databricks连接创建SQLDatabase实例的类方法。

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

自版本langchain-community==0.0.1以来已弃用： 使用 get_usable_table_names 代替。

get_usable_table_names()

获取可用表的名称。

run(command[, fetch, include_columns, ...])

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

run_no_throw(command[, fetch, ...])

执行SQL命令并返回代表结果的字符串。
参数

  • 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 创建引擎。

参数

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

从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租户中的数据库名。

返回

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

返回类型

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’包。如果未安装，则可以使用＜code＞pip install databricks-sql-connector＜/code＞进行安装。

参数

  • catalog （str）- Databricks数据库中的目录名称。

  • schema （str）- 目录中的架构名称。

  • host （Optional[str]) - Databricks工作区的域名，不包括‘＜a class="reference external" href="https://"＞https://＜/a＞’部分。如未提供，尝试从环境变量“DATABRICKS_HOST”获取。如果仍然不可用并且正在Databricks笔记本中运行，则默认为当前工作区的域名。默认为None。

  • api_token （Optional[str]) - 用于访问Databricks SQL仓库或集群的Databricks个人访问令牌。如果未提供，尝试从“DATABRICKS_TOKEN”获取。如果仍然不可用并且正在Databricks笔记本中运行，则生成当前用户的临时令牌。默认为None。

  • warehouse_id （Optional[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 (任何) – 为from_uri方法提供的额外关键字参数。

返回

配置了提供信息的一个SQLDatabase实例。

Databricks连接详细信息。

返回类型

SQLDatabase

抛出异常

ValueError – 如果找不到‘databricks-sql-connector’或同时提供了‘warehouse_id’和‘cluster_id’或者既没有‘warehouse_id’也没有‘cluster_id’且不在Databricks笔记本内部执行。

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

从URI构造SQLAlchemy引擎。

参数

  • database_uri (Union[str, URL]) –

  • engine_args (可选[dict]) –

  • kwargs (任何) –

返回类型

SQLDatabase

get_context() → Dict[str, Any][source]

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

返回类型

Dict[str, 任何]

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

获取指定表的信息。

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

如果设置了sample_rows_in_table_info，则每个表描述将附加指定数量的样本行。这可以像论文中展示的那样提高性能。

参数

table_names （可选 [ 字符串列表 ]） –

返回类型

字符串

get_table_info_no_throw(table_names: Optional[List[str]] = None) → 字符串[source]

获取指定表的信息。

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

如果设置了sample_rows_in_table_info，则每个表描述将附加指定数量的样本行。这可以像论文中展示的那样提高性能。

参数

table_names （可选 [ 字符串列表 ]） –

返回类型

字符串

get_table_names() → 可迭代对象[字符串][source]

自版本langchain-community==0.0.1以来已弃用： 使用 get_usable_table_names 代替。

获取可用表的名称。

返回类型

可迭代对象[字符串]

get_usable_table_names() → 可迭代对象[字符串][source]

获取可用表的名称。

返回类型

可迭代对象[字符串]

run(command: Union[str, Executable], fetch: Literal['all', 'one', 'cursor'], 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命令并返回代表结果的字符串。

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

参数

  • command (Union[str, Executable]) –

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

  • include_columns (bool) –

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

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

返回类型

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]

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

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

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

参数

  • command (str) –

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

  • include_columns (bool) –

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

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

返回类型

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

使用SQLDatabase的示例