SQL数据库#

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)[来源]#

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`()
`get_usable_table_names`()	获取可用表的名称。
`run`(command[, fetch, include_columns, ...])	执行SQL命令并返回表示结果的字符串。
`run_no_throw`(command[, fetch, ...])	执行一个SQL命令并返回表示结果的字符串。

Parameters:

engine (Engine)
schema (可选[str])
metadata (可选[MetaData])
ignore_tables (可选[列表[字符串]])
include_tables (可选[列表[字符串]])
sample_rows_in_table_info (int)
indexes_in_table_info (bool)
custom_table_info (可选[字典])
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]#

从数据库URI创建引擎。

Parameters:

engine (Engine)
schema (str | None)
metadata (MetaData | None)
ignore_tables (列表[字符串] | 无)
include_tables (列表[字符串] | 无)
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]#

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

Parameters:

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

Returns:

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

Return type:

SQLDatabase

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

Parameters:

catalog (str) – Databricks 数据库中的目录名称。
schema (str) – 目录中的模式名称。
host (可选[str]) – Databricks 工作区的主机名，不包括 ‘https://’ 部分。如果未提供，它会尝试从环境变量 ‘DATABRICKS_HOST’ 中获取。如果仍然不可用并且如果在 Databricks 笔记本中运行，则默认为当前工作区主机名。默认为 None。
api_token (Optional[str]) – 用于访问Databricks SQL仓库或集群的Databricks个人访问令牌。如果未提供，则尝试从‘DATABRICKS_TOKEN’中获取。如果仍然不可用并且在Databricks笔记本中运行，则为当前用户生成一个临时令牌。默认为None。
warehouse_id (可选[str]) – Databricks SQL 中的仓库 ID。如果提供，该方法将配置连接以使用此仓库。不能与 'cluster_id' 同时使用。默认为 None。
cluster_id (Optional[str]) – Databricks Runtime 中的集群 ID。如果提供了此参数，该方法将配置连接以使用此集群。不能与 'warehouse_id' 同时使用。如果在 Databricks 笔记本中运行且 'warehouse_id' 和 'cluster_id' 均为 None，则使用笔记本所附加的集群的 ID。默认为 None。
engine_args (Optional[dict]) – 连接Databricks时使用的参数。默认为None。
**kwargs (Any) – from_uri 方法的额外关键字参数。

Returns:

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

Return type:

SQLDatabase

Raises:

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

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

从URI构建一个SQLAlchemy引擎。

Parameters:

database_uri (str | URL)
engine_args (dict | None)
kwargs (Any)

Return type:

SQLDatabase

get_context() → Dict[str, Any][来源]#

返回您可能希望在代理提示中使用的数据库上下文。

Return type:: Dict[str, Any]

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

获取有关指定表的信息。

遵循最佳实践，如Rajkumar等人在2022年所指定的： (https://arxiv.org/abs/2204.00498)

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

Parameters:: table_names (列表[字符串] | 无)
Return type:: 字符串

get_table_info_no_throw(table_names: List[str] | None = None) → str[source]#

获取有关指定表的信息。

遵循最佳实践，如Rajkumar等人在2022年所指定的： (https://arxiv.org/abs/2204.00498)

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

Parameters:: table_names (列表[字符串] | 无)
Return type:: 字符串

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

自版本0.0.1起已弃用：请改用get_usable_table_names()。在langchain-community==1.0之前不会移除。

获取可用表的名称。

Return type:: 可迭代[str]

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

获取可用表的名称。

Return type:: 可迭代[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][来源]#

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

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

Parameters:

command (str | Executable)
fetch (Literal['all', 'one', 'cursor'])
include_columns (bool)
参数 (字典[字符串, 任意类型] | 无)
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]#

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

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

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

Parameters:

command (str)
fetch (Literal['all', 'one'])
include_columns (bool)
参数 (字典[字符串, 任意类型] | 无)
execution_options (Dict[str, Any] | None)

Return type:

str | 序列[字典[str, 任意]] | 结果[任意]

使用 SQLDatabase 的示例