数据库 `hscredit.database`

统一 SQL/NoSQL 连接池、参数化查询、流式读取、自动建表、分批写入、Redis/MongoDB CRUD、元数据导出与适配器扩展 API。使用流程和后端能力矩阵见数据库。

数据库门面

class hscredit.database.client.Database(database_type, *, pool_options=None, adapter_options=None, **connect_kwargs)[源代码]

基类：object

统一数据库连接与操作门面。

参数

database_typestr: 已注册数据库类型或别名。
pool_optionsmapping or PoolOptions, optional: 连接池配置。
adapter_optionsmapping, optional: 后端专有配置。
**connect_kwargs: 直接传递给后端驱动的连接参数。

属性

adapter: 当前数据库适配器实例。

参考样例

>>> with Database("mysql", host="127.0.0.1", database="risk") as db:
...     rows = db.query("SELECT 1", result="rows")

参数:

database_type (str)
pool_options (Mapping[str, Any] | None)
adapter_options (Mapping[str, Any] | None)
connect_kwargs (Any)

property closed: bool

数据库门面是否已经关闭。

返回:: 已调用 close() 时为 True，否则为 False。
返回类型:: bool

query(sql, params=None, result='dataframe')[源代码]

一次性执行查询并返回全部结果。

大结果集建议改用 stream_query() 或 read_query()，避免一次性占用过多内存。

参数:

sql (str) -- 由当前数据库执行的查询 SQL。
params (Any) -- 按底层驱动参数风格绑定的 SQL 参数，默认为 None。
result (str) -- 结果类型，支持 dataframe、records 和 rows。

返回:

dataframe 返回 DataFrame；records 返回记录字典列表；rows 返回原始行列表。

抛出:

StateError -- 数据库门面已经关闭。
ValidationError -- result 不是支持的结果类型。
DatabaseQueryError -- 数据库执行查询失败。

返回类型:

Any

参考样例

>>> frame = db.query("SELECT id FROM events WHERE id > %s", params=(10,))
>>> records = db.query("SELECT id FROM events", result="records")

execute(sql, params=None)[源代码]

执行单条 DDL 或 DML SQL。

参数:

sql (str) -- 要执行的 SQL。
params (Any) -- 按底层驱动参数风格绑定的 SQL 参数，默认为 None。

返回:

适配器报告的影响行数或原生执行结果。

抛出:

StateError -- 数据库门面已经关闭。
DatabaseQueryError -- SQL 执行失败。

返回类型:

Any

executemany(sql, values)[源代码]

使用多组绑定值批量执行同一条 SQL。

参数:

sql (str) -- 带驱动占位符的 SQL。
values (Any) -- 每行一组参数的可迭代对象。

返回:

适配器报告的累计影响行数或原生执行结果。

抛出:

StateError -- 数据库门面已经关闭。
DatabaseQueryError -- 批量执行失败。

返回类型:

Any

property native_client: Any: 返回 Redis 或 MongoDB 适配器持有的原生客户端。

read_one(resource, selector=None, **options)[源代码]

读取单个 Redis key 或 MongoDB 文档。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

Any

read_many(resource, selector=None, **options)[源代码]

批量读取 Redis keys 或 MongoDB 文档。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

Any

read(resource, selector=None, **options)[源代码]

根据输入形态自适应执行单条或批量读取。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

Any

write_one(resource, data, **options)[源代码]

写入单个 Redis key 或 MongoDB 文档。

参数:

resource (Any)
data (Any)
options (Any)

返回类型:

Any

write_many(resource, data=None, **options)[源代码]

批量写入 Redis key-value 或 MongoDB 文档。

参数:

resource (Any)
data (Any)
options (Any)

返回类型:

Any

write(resource, data=None, **options)[源代码]

根据适配器类型写入 SQL 表或 NoSQL 资源。

SQL 数据库使用 write(表名, 数据, ...)，默认按 mode="a" 进入 stream_write() 流水线；目标表不存在时根据首批数据自动建表。Redis 和 MongoDB 保持原有自适应单条/批量写入语义。

参数:

resource (Any) -- SQL 目标表限定名、Redis key/映射或 MongoDB collection。
data (Any) -- SQL DataFrame/记录迭代器、Redis 值或 MongoDB 文档。
options (Any) -- SQL 写入模式、批次、键字段和方言参数，或 NoSQL 后端选项。

返回:

SQL 返回 WriteResult；NoSQL 返回 NoSQLWriteResult。

抛出:

StateError -- 数据库门面已经关闭。
DatabaseCapabilityError -- 后端不能保证请求的写入语义。
DatabaseWriteError -- SQL 表准备或分批写入失败。

返回类型:

Any

参考样例

>>> result = db.write("risk.events", frame, key_columns="id")
>>> result = db.write("risk.events", frame, mode="r", key_columns=["id"])

delete_one(resource, selector=None, **options)[源代码]

删除单个 Redis key 或首个匹配 MongoDB 文档。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

Any

delete_many(resource, selector=None, **options)[源代码]

批量删除 Redis keys 或 MongoDB 文档。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

Any

delete(resource, selector=None, **options)[源代码]

根据输入形态与 many 选项自适应执行删除。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

Any

exists(resource, selector=None, **options)[源代码]

判断 Redis key 或 MongoDB 匹配文档是否存在。

参数:

resource (Any)
selector (Any)
options (Any)

返回类型:

bool

stream_query(sql, params=None, *, chunksize=50000, progress=False, retain=True, count_total=False, count_sql=None, total_rows=None, columns=None, json_fields=None, result='dataframe')[源代码]

打开可中断的分块查询流。

当 json_fields 不为空时，适配器把 JSON 路径提取下推到数据库，只传输 columns 和指定的 JSON 子字段，不返回原始大 JSON。JSON 字段定义格式为 {源字段: {输出字段: 路径或(路径, 默认值)}}。

参数:

sql (str) -- 原始查询 SQL。JSON 投影会把该查询包装为子查询。
params (Any) -- 原始 SQL 的绑定参数，默认为 None。
chunksize (int) -- 每次向 DB-API 流式游标请求的最大行数，默认 50000。
progress (bool) -- 是否显示读取进度，默认 False。未知总数时显示累计行数、速度和耗时。
retain (bool) -- 是否保留已经产生的分块，默认 True。设为 False 后不能合并历史数据。
count_total (bool) -- 是否为进度条自动执行 COUNT(1)，默认 False。
count_sql (str | None) -- 为进度条显式指定的统计 SQL；提供后会执行该 SQL。
total_rows (int | None) -- 已知总行数；提供后不执行统计 SQL。
columns (Sequence[str] | None) -- JSON 投影时原样保留的普通输出字段；不能包含 JSON 源字段。
json_fields (Mapping[str, Mapping[str, Any]] | None) -- JSON 源字段、输出字段、JSONPath 和可选默认值的嵌套映射。
result (str) -- 每个分块的结果类型，支持 dataframe、records 和 rows。

返回:

可迭代、可主动停止并可合并已读数据的查询流。

返回类型:

QueryStream

抛出:

ValidationError -- 分块、进度、结果类型或 JSON 投影参数无效。
DatabaseCapabilityError -- 当前适配器不支持 JSON 字段投影。
DatabaseQueryError -- 统计查询或打开流式查询失败。

参考样例

>>> stream = db.stream_query(
...     "SELECT id, huge_json FROM user_profile",
...     columns=["id"],
...     json_fields={
...         "huge_json": {
...             "city": ("$.address.city", "未知"),
...             "customer_id": "$.customer.id",
...         }
...     },
...     result="records",
... )
>>> for records in stream:
...     consume(records)

read_query(sql, params=None, *, chunksize=50000, progress=False, count_total=False, count_sql=None, total_rows=None, columns=None, json_fields=None, result='dataframe')[源代码]

消费完整查询流，并在中断后直接返回已经读取的数据。

参数语义与 stream_query() 一致，但本方法自动消费所有分块。发生 KeyboardInterrupt 时会关闭底层资源，并按照 result 返回当前已合并数据。

参数:

sql (str) -- 原始查询 SQL。
params (Any) -- 原始 SQL 的绑定参数，默认为 None。
chunksize (int) -- 每次请求的最大行数，默认 50000。
progress (bool) -- 是否显示进度，默认 False。
count_total (bool) -- 是否自动执行 COUNT(1) 获取进度条总数，默认 False。
count_sql (str | None) -- 为进度条显式指定的统计 SQL。
total_rows (int | None) -- 已知总行数。
columns (Sequence[str] | None) -- JSON 投影时原样保留的普通输出字段；不能包含 JSON 源字段。
json_fields (Mapping[str, Mapping[str, Any]] | None) -- JSON 子字段投影映射。
result (str) -- 返回类型，支持 dataframe、records 和 rows。

返回:

完整数据或中断前已读取的部分数据。

抛出:

ValidationError -- 查询或投影参数无效。
DatabaseQueryError -- 流式查询失败。

返回类型:

Any

参考样例

>>> frame = db.read_query("SELECT * FROM events", progress=True)
>>> rows = db.read_query("SELECT id FROM events", result="rows")

export_schema(targets=None, *, output=None, excel_params=None)[源代码]

读取数据库表结构并生成中文字段元数据宽表。

参数:

targets (Sequence[str] | None) -- 可选数据库或表目标，例如 risk、risk.events；默认扫描适配器可见范围。
output (Any | None) -- 可选 .xlsx 输出路径。提供时通过 dataframe2excel 导出。
excel_params (Mapping[str, Any] | None) -- 传递给 dataframe2excel 的附加参数。

返回:

中文列名的表和字段信息 DataFrame；数据库原始元数据值保持不变。

返回类型:

pandas.DataFrame

抛出:

ValidationError -- 目标、输出扩展名或 Excel 参数无效。
DatabaseMetadataError -- 元数据读取、目标匹配或 Excel 导出失败。

create_table(data, table_name, *, dialect_options=None)[源代码]

根据 DataFrame 字段和后端方言参数创建表。

参数:

data (DataFrame) -- 用于推断字段结构的非空列 DataFrame。
table_name (str) -- 表名或 数据库名.表名 等限定名。
dialect_options (Mapping[str, Any] | None) -- 后端专有建表参数，例如键、引擎、分区、字段类型和注释。

返回:

适配器返回的已执行 DDL 或原生结果。

抛出:

InputValidationError -- data 不是带字段的 DataFrame。
ValidationError -- 表名或方言参数无效。
DatabaseQueryError -- 建表 SQL 执行失败。

返回类型:

Any

stream_write(data, table_name, *, mode='a', batch_size=10000, key_columns=None, columns=None, dialect_options=None)[源代码]

把 DataFrame、DataFrame 分块或行记录迭代器流式写入目标表。

mode 默认为 a。目标表不存在时，四种模式都会根据首个有效分块和 dialect_options 创建表；已有表继续执行各模式原有的追加、覆盖、清空或重建语义。

参数:

data (Any) -- DataFrame、DataFrame 分块、记录字典或位置行的可迭代对象。
table_name (str) -- 目标表限定名。
mode (str) -- 写入模式。a 追加且主键重复不覆盖；r 追加且主键重复覆盖； o 保留表结构并清空重写；d 删除并按首批数据重建表后写入。
batch_size (int) -- 每个写入批次的最大行数，默认 10000。
key_columns (Sequence[str] | None) -- a 或 r 使用的显式键字段；未提供时由支持的适配器读取表元数据。
columns (Sequence[str] | None) -- 位置行字段名，或用于校验 DataFrame/记录字段顺序。
dialect_options (Mapping[str, Any] | None) -- 传递给目标适配器的建表和写入选项。

返回:

完成状态、接收/插入/更新/跳过行数及已提交批次数。

返回类型:

WriteResult

抛出:

InputValidationError -- 数据为空、批次字段不一致或缺少键字段。
ValidationError -- 模式、批次大小、表名或方言参数无效。
DatabaseCapabilityError -- 目标数据库或表不支持指定模式。
DatabaseWriteError -- 准备、批次写入或收尾失败；异常的 result 保留部分统计。

atomic_replace_table(data, table_name, *, batch_size=10000, columns=None, keep_backup=True)[源代码]

显式选择暂存完整写入后的一次原子整表替换，不改变stream_write。

首版仅支持Oracle MySQL 8.0.13+/8.4中已存在的InnoDB基础表；拒绝入向/出向外键、触发器和无法证明完整元数据可见性的连接。适用于受控维护连接，不自动改变权限，也不建议为日常业务账号扩大权限。

CREATE TABLE LIKE保留列属性/索引及MySQL支持复制的定义；CHECK约束名称由服务器重新生成，不保证元数据名称不变。拒绝DATA DIRECTORY/INDEX DIRECTORY表选项。暂存和备份均为同schema的随机托管名。分批数据仅提交到暂存表，最终在写锁内一次RENAME交换。构建期间对旧表的并发写入不会合并到新数据，调用方应安排维护窗口；默认保留旧表备份。 MySQL原子DDL不是包含全部DML的外部事务。需要CREATE/INSERT/ALTER/DROP/ LOCK TABLES及可验证的元数据可见性；暂不展开角色授权证明。专用会话启用严格sql_mode并禁用ALLOW_INVALID_DATES；每条单语句批量 INSERT后核查警告，无警告才提交。每批不自动拆语句，超出服务器包大小时明确失败，调用方需减小batch_size。lock_wait_timeout固定为30秒/次元数据锁申请，并非整个操作总超时；不修改全局配置。

rows_staged/batches_staged表示确认提交到暂存的数量；rows_inserted和 rows_published只表示目标发布后行数。target_published为False/True/None，分别表示未发布/已确认发布/请求后结果未知，未知时禁止自动重试。 batches_committed在此入口表示已确认目标发布次数（0或1），不是暂存批数。清理失败记录在details，不掩盖主异常；成功发布但清理失败仍返回已发布收据。

参数:

data (Any)
table_name (str)
batch_size (int)
columns (Sequence[str] | None)
keep_backup (bool)

返回类型:

WriteResult

close()[源代码]

关闭连接池或原生客户端；重复调用不产生副作用。

关闭后所有查询和写入方法都会抛出 StateError。

返回类型:: None

类外快捷操作

数据库与 NoSQL 的类外快捷操作。

快捷函数接受以下任一 source，并把实际操作委托给 Database 的同名方法：

含 db_type 的连接配置映射；
已创建的 Database 实例；
PyMySQL、python-oracledb、Impyla、PyODPS 等原生 DB-API 连接。

配置创建的连接池在操作结束后自动关闭；传入的 Database 和原生连接均视为借用，不由快捷函数关闭。配置创建的流式查询在查询流完成、停止或关闭时释放连接池。

参考样例

>>> from hscredit.database import read_query
>>> config = {"db_type": "mysql", "host": "127.0.0.1", "user": "risk", "password": "***"}
>>> frame = read_query(config, "SELECT id FROM events")

hscredit.database.shortcuts.query(source, sql, params=None, result='dataframe', *, db_type=None)[源代码]

快捷执行一次性 SQL 查询。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
sql (str) -- 查询 SQL。
params (Any) -- 驱动绑定参数。
result (str) -- dataframe、records 或 rows。
db_type (str | None) -- 原生连接无法自动识别时使用的数据库类型。

返回:

对应格式的完整查询结果。

返回类型:

Any

hscredit.database.shortcuts.execute(source, sql, params=None, *, db_type=None)[源代码]

快捷执行单条 DDL 或 DML SQL。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
sql (str) -- 要执行的 SQL。
params (Any) -- 驱动绑定参数。
db_type (str | None) -- 可选数据库类型。

返回:

适配器报告的影响行数或原生结果。

返回类型:

Any

hscredit.database.shortcuts.executemany(source, sql, values, *, db_type=None)[源代码]

快捷批量执行同一条 SQL。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
sql (str) -- 带驱动占位符的 SQL。
values (Any) -- 多组绑定值。
db_type (str | None) -- 可选数据库类型。

返回:

累计影响行数或原生结果。

返回类型:

Any

hscredit.database.shortcuts.stream_query(source, sql, params=None, *, db_type=None, **options)[源代码]

快捷打开流式 SQL 查询；拥有的数据源随查询流关闭。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
sql (str) -- 查询 SQL。
params (Any) -- 驱动绑定参数。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 传递给 Database.stream_query() 的分块、进度和 JSON 投影参数。

返回:

可中断并可合并已读数据的 QueryStream。

返回类型:

Any

hscredit.database.shortcuts.read_query(source, sql, params=None, *, db_type=None, **options)[源代码]

快捷消费流式 SQL 查询并返回合并结果。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
sql (str) -- 查询 SQL。
params (Any) -- 驱动绑定参数。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 传递给 Database.read_query() 的选项。

返回:

DataFrame、记录字典列表或原始行列表。

返回类型:

Any

hscredit.database.shortcuts.export_schema(source, targets=None, *, db_type=None, **options)[源代码]

快捷导出数据库表和字段元数据。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
targets (Any) -- 数据库或 数据库.表 目标。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 输出路径和 excel_params。

返回:

中文字段元数据 DataFrame。

返回类型:

Any

hscredit.database.shortcuts.create_table(source, data, table_name, *, db_type=None, **options)[源代码]

快捷根据 DataFrame 创建表。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
data (Any) -- 用于推断表结构的 DataFrame。
table_name (str) -- 目标表限定名。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 传递给 Database.create_table() 的方言参数。

返回:

已执行 DDL 或适配器原生结果。

返回类型:

Any

hscredit.database.shortcuts.stream_write(source, data, table_name, *, db_type=None, **options)[源代码]

快捷流式写入 DataFrame 或记录迭代器。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
data (Any) -- DataFrame、分块或记录迭代器。
table_name (str) -- 目标表限定名。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 写入模式、批次、键字段和方言参数。

返回:

WriteResult 写入统计。

返回类型:

Any

hscredit.database.shortcuts.read_one(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷读取单个 Redis key 或 MongoDB 文档。

参数:

source (Any) -- 数据库配置或 Database 实例。
resource (Any) -- Redis key 或 MongoDB collection。
selector (Any) -- MongoDB 查询条件。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 后端读取选项。

返回:

单值或单个文档。

返回类型:

Any

hscredit.database.shortcuts.read_many(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷批量读取 Redis keys 或 MongoDB 文档。

参数与 read_one() 一致，返回后端批量读取结果。

参数:

source (Any)
resource (Any)
selector (Any)
db_type (str | None)
options (Any)

返回类型:

Any

hscredit.database.shortcuts.read(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷自适应执行单条或批量 NoSQL 读取。

参数与 read_one() 一致，具体单条/批量语义由适配器根据输入和选项确定。

参数:

source (Any)
resource (Any)
selector (Any)
db_type (str | None)
options (Any)

返回类型:

Any

hscredit.database.shortcuts.write_one(source, resource, data, *, db_type=None, **options)[源代码]

快捷写入单个 Redis key 或 MongoDB 文档。

参数:

source (Any) -- 数据库配置或 Database 实例。
resource (Any) -- Redis key 或 MongoDB collection。
data (Any) -- 要写入的值或文档。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 后端写入选项。

返回:

后端写入结果。

返回类型:

Any

hscredit.database.shortcuts.write_many(source, resource, data=None, *, db_type=None, **options)[源代码]

快捷批量写入 Redis key-value 或 MongoDB 文档。

参数与 write_one() 一致，data 为批量输入。

参数:

source (Any)
resource (Any)
data (Any)
db_type (str | None)
options (Any)

返回类型:

Any

hscredit.database.shortcuts.write(source, resource, data=None, *, db_type=None, **options)[源代码]

快捷写入 SQL 表，或自适应执行单条/批量 NoSQL 写入。

SQL 用法为 write(source, 表名, 数据, ...)，默认 mode="a"；目标表不存在时根据首批数据自动创建。NoSQL 参数与 write_one() 一致，具体单条/批量语义由适配器确定。

参数:

source (Any) -- 数据库配置、Database 实例或原生 DB-API 连接。
resource (Any) -- SQL 目标表限定名、Redis key/映射或 MongoDB collection。
data (Any) -- SQL DataFrame/记录迭代器、Redis 值或 MongoDB 文档。
db_type (str | None) -- 可选数据库类型。
options (Any) -- SQL 写入模式、批次、键字段和方言参数，或 NoSQL 后端选项。

返回:

SQL WriteResult 或 NoSQLWriteResult。

返回类型:

Any

hscredit.database.shortcuts.delete_one(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷删除单个 Redis key 或 MongoDB 文档。

参数:

source (Any) -- 数据库配置或 Database 实例。
resource (Any) -- Redis key 或 MongoDB collection。
selector (Any) -- MongoDB 查询条件。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 后端删除选项。

返回:

后端删除结果。

返回类型:

Any

hscredit.database.shortcuts.delete_many(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷批量删除 Redis keys 或 MongoDB 文档。

参数与 delete_one() 一致，返回后端批量删除结果。

参数:

source (Any)
resource (Any)
selector (Any)
db_type (str | None)
options (Any)

返回类型:

Any

hscredit.database.shortcuts.delete(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷自适应执行单条或批量 NoSQL 删除。

参数与 delete_one() 一致，具体单条/批量语义由适配器确定。

参数:

source (Any)
resource (Any)
selector (Any)
db_type (str | None)
options (Any)

返回类型:

Any

hscredit.database.shortcuts.exists(source, resource, selector=None, *, db_type=None, **options)[源代码]

快捷判断 Redis key 或 MongoDB 文档是否存在。

参数:

source (Any) -- 数据库配置或 Database 实例。
resource (Any) -- Redis key 或 MongoDB collection。
selector (Any) -- MongoDB 查询条件。
db_type (str | None) -- 可选数据库类型。
options (Any) -- 后端查询选项。

返回:

是否存在。

返回类型:

bool

流式查询

class hscredit.database.stream.QueryStream(resource, *, chunksize, retain=True, total_rows=None, progress=False, result='dataframe', defaults=None, sql=None, params=None)[源代码]

基类：object

可中断并可合并已读数据的分块查询迭代器。

QueryStream 由 Database.stream_query 创建。迭代期间只持有当前数据库资源；retain=True 时额外保留已经返回的标准化 DataFrame 分块，用于停止或中断后合并结果。

参数

resource: 实现 fetchmany(size)、close() 和 columns 的适配器查询资源。
chunksizeint: 每次请求的最大行数。
retainbool, default=True: 是否保留已读取分块。
total_rowsint, optional: 进度条总行数。
progressbool, default=False: 是否显示 tqdm 进度条。
result{"dataframe", "records", "rows"}, default="dataframe": 每次迭代和最终合并使用的结果类型。
defaultsmapping, optional: JSON 投影字段在数据库返回 NULL 时使用的默认值。
sqlstr, optional: 当前流实际执行的 SQL，用于失败异常上下文。
paramsAny, optional: SQL 绑定参数，仅作为失败异常属性保存。

属性

stateStreamState: 当前运行、完成、中断、失败或关闭状态。
rows_readint: 已经读取并返回的累计行数。
interrupt_reasonstr, optional: 主动停止、中断或失败原因。

参考样例

>>> stream = db.stream_query("SELECT id FROM events", chunksize=1000)
>>> first = next(stream)
>>> stream.stop("仅抽取首批")
>>> partial = stream.to_dataframe()

参数:

resource (Any)
chunksize (int)
retain (bool)
total_rows (int | None)
progress (bool)
result (str)
defaults (Mapping[str, Any] | None)
sql (str | None)
params (Any)

stop(reason='用户主动停止')[源代码]

安全停止读取并关闭底层查询资源。

参数:: reason (str) -- 写入流状态和最终 DataFrame 属性的停止原因。
返回:: None。已经读取的数据可继续通过 to_result() 获取。
返回类型:: None

close()[源代码]

关闭查询资源但不把状态标记为主动中断。

尚在运行时状态会变为 closed；重复调用不会重复关闭游标或连接。

返回类型:: None

to_dataframe()[源代码]

把已保留分块合并为 DataFrame，并附加读取状态属性。

返回:: 连续索引的合并 DataFrame。attrs 包含 completed、rows_read、 total_rows、state、interrupted_at 和 interrupt_reason。
返回类型:: pandas.DataFrame
抛出:: StateError -- 创建流时设置了 retain=False。

to_records()[源代码]

把已保留分块合并为记录字典列表。

返回:: 每行一个字段名到原始值映射的列表。
抛出:: StateError -- 创建流时设置了 retain=False。
返回类型:: List[Mapping[str, Any]]

to_rows()[源代码]

把已保留分块合并为原始行元组列表。

返回:: 字段顺序与查询结果一致的行元组列表。
抛出:: StateError -- 创建流时设置了 retain=False。
返回类型:: List[Any]

to_result()[源代码]

按创建流时的 result 配置返回合并结果。

返回:: DataFrame、list[dict] 或行元组列表。
抛出:: StateError -- 创建流时设置了 retain=False。
返回类型:: Any

to_envelope()[源代码]

返回包含完整性状态的结果信封，records/rows也不会丢失中断信息。

返回类型:: Mapping[str, Any]

hscredit.database.types.RESULT_TYPES = frozenset({'dataframe', 'records', 'rows'}): Database.query、stream_query 和 read_query 共用的结果类型。

JSON 投影扩展

class hscredit.database.adapters.base.BaseDatabaseAdapter(*, connect_kwargs, pool_options, adapter_options=None)[源代码]

基类：object

数据库适配器公共基类。

参数

connect_kwargsdict: 传递给底层数据库驱动的连接参数。
pool_optionsPoolOptions: 已校验的连接池配置。
adapter_optionsdict: 仅由适配器解释的方言或原生通道参数。

属性

database_typestr: 注册表使用的数据库类型名称。
capabilitiesDatabaseCapabilities: 适配器默认保证的事务、读取、写入与元数据能力。

参考样例

自定义后端至少实现查询、流资源和所需写入方法；支持 JSON 字段投影时覆盖 json_extract_expression()，公共 SQL 包装由 build_json_projection_sql() 完成。

参数:

connect_kwargs (Mapping[str, Any])
pool_options (PoolOptions)
adapter_options (Mapping[str, Any] | None)

database_type = 'base'

identifier_quote = '"'

is_nosql = False

capabilities = DatabaseCapabilities(transactions=True, streaming_read=True, native_bulk_write=False, metadata_export=True, write_modes=frozenset({'d', 'o'}), atomic_replace=False)

property closed: bool: 适配器是否已关闭。

ensure_open()[源代码]

确保适配器仍可使用。

返回类型:: None

capabilities_for_table(table_name, table_metadata=None)[源代码]

返回目标表可保证的能力。

参数:

table_name (str)
table_metadata (Mapping[str, Any] | None)

返回类型:

DatabaseCapabilities

require_write_mode(table_name, mode, table_metadata=None)[源代码]

校验目标表是否支持指定写入模式。

参数:

table_name (str)
mode (str)
table_metadata (Mapping[str, Any] | None)

返回类型:

DatabaseCapabilities

build_count_sql(sql)[源代码]

生成通用子查询计数 SQL。

参数:: sql (str)
返回类型:: str

quote_identifier(identifier)[源代码]

按当前数据库方言引用单个标识符。

参数:: identifier (str)
返回类型:: str

quote_qualified_name(name)[源代码]

逐段引用数据库对象限定名。

参数:: name (str)
返回类型:: str

json_extract_expression(column_sql, path)[源代码]

生成从单个 JSON 字段提取路径值的 SQL 表达式。

适配器只负责返回表达式，不添加输出别名。path 已经由公共层完成安全校验。

参数:

column_sql (str) -- 已按当前方言引用的 JSON 源字段表达式。
path (str) -- 以 $ 开头的 JSONPath。

返回:

返回标量或嵌套 JSON 原始值的 SQL 表达式。

抛出:

DatabaseCapabilityError -- 适配器未实现 JSON 字段投影。

返回类型:

str

build_json_projection_sql(sql, *, columns=None, json_fields=None)[源代码]

把原查询包装为仅返回普通字段和指定 JSON 子字段的 SQL。

参数:

sql (str) -- 输出中必须包含 columns 和所有 JSON 源字段的原始查询。
columns (Sequence[str] | None) -- 原样保留的普通字段名序列；不能包含 json_fields 的源字段。
json_fields (Mapping[str, Mapping[str, Any]] | None) -- {源字段: {输出字段: 路径或(路径, 默认值)}} 映射。默认值由 QueryStream 在分块返回前处理，不写入 SQL。

返回:

仅选择目标字段的后端方言 SQL；没有投影时返回原 SQL。

抛出:

ValidationError -- 字段名、JSONPath、默认值简写或输出名重复无效。
DatabaseCapabilityError -- 当前适配器没有 JSON 路径提取实现。

返回类型:

str

create_table(data, table_name, *, dialect_options=None)[源代码]

创建目标表。

参数:

data (Any)
table_name (str)
dialect_options (Mapping[str, Any] | None)

返回类型:

Any

ensure_table(data, table_name, *, dialect_options=None, exists=None)[源代码]

目标表不存在时根据首批数据创建表，并容忍并发建表竞态。

参数:

data (Any)
table_name (str)
dialect_options (Mapping[str, Any] | None)
exists (bool | None)

返回类型:

Any

is_table_already_exists_error(exc)[源代码]

判断建表异常是否明确表示目标表已存在。

参数:: exc (BaseException)
返回类型:: bool

table_exists(table_name)[源代码]

只读判断目标表是否存在；无法通用判断时返回 None。

参数:: table_name (str)
返回类型:: bool | None

validate_write(table_name, mode, first_batch, *, key_columns=None, dialect_options=None, table_exists=None)[源代码]

在任何建表、清空或删除动作之前校验写入能力。

参数:

table_name (str)
mode (str)
first_batch (Any)
key_columns (Sequence[str] | None)
dialect_options (Mapping[str, Any] | None)
table_exists (bool | None)

返回类型:

None

resolve_key_columns(table_name, key_columns, first_batch, *, dialect_options=None)[源代码]

解析显式或数据库元数据中的主键字段。

参数:

table_name (str)
key_columns (Sequence[str] | None)
first_batch (Any)
dialect_options (Mapping[str, Any] | None)

返回类型:

Sequence[str] | None

prepare_write(table_name, mode, first_batch, *, key_columns=None, dialect_options=None)[源代码]

在首个批次写入前校验并准备目标表。

参数:

table_name (str)
mode (str)
first_batch (Any)
key_columns (Sequence[str] | None)
dialect_options (Mapping[str, Any] | None)

返回类型:

None

write_batch(table_name, batch, mode, batch_index, *, key_columns=None, dialect_options=None)[源代码]

写入一个已经校验的 DataFrame 批次。

参数:

table_name (str)
batch (Any)
mode (str)
batch_index (int)
key_columns (Sequence[str] | None)
dialect_options (Mapping[str, Any] | None)

返回类型:

Any

finish_write(table_name, mode, result, *, dialect_options=None)[源代码]

完成适配器专有的写入收尾。

参数:

table_name (str)
mode (str)
result (Any)
dialect_options (Mapping[str, Any] | None)

返回类型:

None

atomic_replace_session(table_name)[源代码]

显式暂存/原子替换会话；默认后端不承诺该能力。

参数:: table_name (str)

query(sql, params=None, result='dataframe')[源代码]

执行查询。

参数:

sql (str)
params (Any)
result (str)

返回类型:

Any

close()[源代码]

关闭适配器持有的资源。

返回类型:: None

适配器注册

hscredit.database.registry.register_adapter(name, adapter_class, *, aliases=(), replace=False)[源代码]

注册自定义数据库适配器及其别名。

参数:

name (str) -- 新适配器的规范名称。
adapter_class (Type[BaseDatabaseAdapter]) -- BaseDatabaseAdapter 的实现类。
aliases (Iterable[str]) -- 可选别名集合。
replace (bool) -- 是否允许替换同名适配器和冲突别名，默认 False。

返回:

None。

抛出:

ValidationError -- 名称、适配器类型或别名冲突无效。

返回类型:

None

参考样例

>>> register_adapter("custom_db", CustomDatabaseAdapter, aliases=["custom"])
>>> Database("custom", host="127.0.0.1")

hscredit.database.registry.get_adapter_class(name)[源代码]

按数据库类型或别名获取适配器类。

内置适配器在第一次获取时才导入，从而保持所有数据库驱动均为可选依赖。

参数:: name (str) -- 数据库类型或别名。
返回:: 已加载的适配器类。
抛出:: ValidationError -- 数据库类型未注册或导入入口无效。
返回类型:: Type[BaseDatabaseAdapter]

hscredit.database.registry.available_adapters()[源代码]

返回当前已注册的规范数据库类型。

返回:: 按名称排序且不包含别名的数据库类型元组。
返回类型:: tuple[str, ...]

公共类型

class hscredit.database.types.PoolOptions(mincached=0, maxcached=0, maxshared=0, maxconnections=0, blocking=False, maxusage=None, setsession=None, ping=1)[源代码]

基类：object

DBUtils 兼容的连接池配置。

参数

mincached、maxcached、maxshared、maxconnections 与 DBUtils PooledDB 含义一致，值为 0 时沿用 DBUtils 的“不限制”语义。

参数:

mincached (int)
maxcached (int)
maxshared (int)
maxconnections (int)
blocking (bool)
maxusage (int | None)
setsession (Tuple[str, ...] | None)
ping (int)

mincached: int = 0

maxcached: int = 0

maxshared: int = 0

maxconnections: int = 0

blocking: bool = False

maxusage: int | None = None

setsession: Tuple[str, ...] | None = None

ping: int = 1

classmethod from_mapping(value=None)[源代码]

从映射创建并校验连接池配置。

参数:: value (Mapping[str, Any] | None) -- PoolOptions、参数映射或 None。
返回:: 已校验的不可变连接池配置。
抛出:: ValidationError -- 包含未知参数或参数范围无效。
返回类型:: PoolOptions

to_dbutils_kwargs()[源代码]

转换为 DBUtils PooledDB 关键字参数。

返回:: 可直接传给 PooledDB 的新字典。
返回类型:: dict
抛出:: ValidationError -- 当前配置组合无效。

class hscredit.database.types.RedisPoolOptions(max_connections=None, blocking=False, timeout=None)[源代码]

基类：object

Redis 原生连接池配置。

参数:

max_connections (int | None)
blocking (bool)
timeout (float | None)

max_connections: int | None = None

blocking: bool = False

timeout: float | None = None

classmethod from_mapping(value=None)[源代码]

从映射创建 Redis 连接池配置。

参数:: value (Mapping[str, Any] | None)
返回类型:: RedisPoolOptions

to_redis_kwargs()[源代码]

转换为 redis-py 连接池参数。

返回类型:: Dict[str, Any]

class hscredit.database.types.MongoPoolOptions(min_pool_size=0, max_pool_size=100, max_connecting=2, wait_queue_timeout_ms=None, max_idle_time_ms=None)[源代码]

基类：object

PyMongo MongoClient 原生连接池配置。

参数:

min_pool_size (int)
max_pool_size (int)
max_connecting (int)
wait_queue_timeout_ms (int | None)
max_idle_time_ms (int | None)

min_pool_size: int = 0

max_pool_size: int = 100

max_connecting: int = 2

wait_queue_timeout_ms: int | None = None

max_idle_time_ms: int | None = None

classmethod from_mapping(value=None)[源代码]

从映射创建 MongoDB 连接池配置。

参数:: value (Mapping[str, Any] | None)
返回类型:: MongoPoolOptions

to_mongo_kwargs()[源代码]

转换为 MongoClient 驼峰连接池参数。

返回类型:: Dict[str, Any]

class hscredit.database.types.DatabaseCapabilities(transactions=True, streaming_read=True, native_bulk_write=False, metadata_export=True, write_modes=<factory>, atomic_replace=False)[源代码]

基类：object

数据库或目标表可保证的能力。

参数

transactionsbool: 是否支持事务提交和回滚。
streaming_readbool: 是否支持流式读取。
native_bulk_writebool: 是否具有后端原生批量写入通道。
metadata_exportbool: 是否支持表结构扫描。
write_modesfrozenset[str]: 可保证的 a/r/o/d 写入模式集合。
atomic_replacebool: 是否提供显式暂存替换协议；True仍须运行时验证版本、表结构及可见性前提。

参数:

transactions (bool)
streaming_read (bool)
native_bulk_write (bool)
metadata_export (bool)
write_modes (FrozenSet[str])
atomic_replace (bool)

transactions: bool = True

streaming_read: bool = True

native_bulk_write: bool = False

metadata_export: bool = True

write_modes: FrozenSet[str]

atomic_replace: bool = False

class hscredit.database.types.WriteResult(mode, completed, rows_received=0, rows_inserted=None, rows_updated=None, rows_skipped=None, batches_committed=0, failed_batch=None, consistency=None, details=<factory>, staging_table=None, backup_table=None, rows_staged=0, batches_staged=0, target_published=None, rows_published=None, phase=None)[源代码]

基类：object

流式写入结果。

参数

mode{"a", "r", "o", "d"}: 本次写入模式。
completedbool: 是否完成全部批次和适配器收尾。
rows_received、rows_inserted、rows_updated、rows_skippedint, optional: 输入及后端报告的行数统计。
batches_committedint: 已成功提交的批次数。
failed_batchint, optional: 失败批次编号；准备阶段为 0，收尾阶段为已提交批次数加 1。
consistencystr, optional: 后端最终一致性说明。
detailsdict: 适配器附加的原始统计信息。
rows_staged、batches_stagedint: atomic_replace_table确认提交至暂存表的历史数量，不表示目标已经更新。
target_publishedbool, optional: 原子替换入口的未发布/已确认发布/未知状态；普通stream_write不使用此字段。
rows_publishedint, optional: 确认发布至目标的行数；发布结果未知时为None。
phasestr, optional: preparing/staging/publishing/published/publication_unknown等操作阶段。

参数:

mode (str)
completed (bool)
rows_received (int)
rows_inserted (int | None)
rows_updated (int | None)
rows_skipped (int | None)
batches_committed (int)
failed_batch (int | None)
consistency (str | None)
details (Dict[str, Any])
staging_table (str | None)
backup_table (str | None)
rows_staged (int)
batches_staged (int)
target_published (bool | None)
rows_published (int | None)
phase (str | None)

mode: str

completed: bool

rows_received: int = 0

rows_inserted: int | None = None

rows_updated: int | None = None

rows_skipped: int | None = None

batches_committed: int = 0

failed_batch: int | None = None

consistency: str | None = None

details: Dict[str, Any]

staging_table: str | None = None

backup_table: str | None = None

rows_staged: int = 0

batches_staged: int = 0

target_published: bool | None = None

rows_published: int | None = None

phase: str | None = None

class hscredit.database.types.NoSQLWriteResult(operation, acknowledged, affected_count=None, matched_count=None, modified_count=None, identifiers=<factory>, details=<factory>)[源代码]

基类：object

Redis 与 MongoDB 共用的写入、更新和删除结果。

参数:

operation (str)
acknowledged (bool)
affected_count (int | None)
matched_count (int | None)
modified_count (int | None)
identifiers (Tuple[Any, ...])
details (Mapping[str, Any])

operation: str

acknowledged: bool

affected_count: int | None = None

matched_count: int | None = None

modified_count: int | None = None

identifiers: Tuple[Any, ...]

details: Mapping[str, Any]

class hscredit.database.types.StreamState(value)[源代码]

基类：str, Enum

流式查询生命周期状态。

running 表示仍可读取；completed 表示自然耗尽；interrupted 表示主动停止或键盘中断；failed 表示读取失败；closed 表示在耗尽前关闭。

RUNNING = 'running'

COMPLETED = 'completed'

INTERRUPTED = 'interrupted'

FAILED = 'failed'

CLOSED = 'closed'

class hscredit.database.metadata.QualifiedTarget(raw, parts)[源代码]

基类：object

未改写大小写的数据库对象限定名。

参数

rawstr: 用户提供并去除首尾空白后的原始目标。
partstuple[str, ...]: 以点拆分的数据库、模式和表名片段。

参数:

raw (str)
parts (Tuple[str, ...])

raw: str

parts: Tuple[str, ...]

classmethod parse(value)[源代码]

解析以点分隔的数据库、模式和表目标。

参数:: value (str) -- 数据库、数据库.表 或多级限定名。
返回:: 保持原始大小写的限定目标。
抛出:: ValidationError -- 目标为空或包含空片段。
返回类型:: QualifiedTarget

class hscredit.database.metadata.MetadataInspection(rows=<factory>, errors=<factory>)[源代码]

基类：object

适配器元数据扫描结果。

参数

rowsiterable of mapping: 使用内部英文字段键保存的表/字段记录。
errorslist: 部分数据库或表扫描失败时保留的原始错误信息。

参数:

rows (Iterable[Mapping[str, Any]])
errors (List[Any])

rows: Iterable[Mapping[str, Any]]

errors: List[Any]

异常

数据库模块异常。

所有异常均兼容 hscredit 统一异常体系，并为调用方保留 SQL、绑定参数和底层驱动异常链。

exception hscredit.database.exceptions.DatabaseError(message, *, sql=None, params=None, driver_error=None, cleanup_error=None)[源代码]

基类：HSCreditError

数据库模块基础异常，并提供可见且可编程读取的执行上下文。

参数

messagestr: 面向用户的中文错误概要。
sqlstr, optional: 实际交给数据库执行的 SQL。
paramsAny, optional: SQL 绑定参数；只作为属性保存，不进入异常文本。
driver_errorBaseException, optional: 底层数据库驱动返回的原始异常。
cleanup_errorBaseException, optional: 主操作失败后发生的资源清理异常。

属性

sqlstr, optional: 当前异常链中最接近底层的 SQL。
paramsAny, optional: 对应 SQL 的绑定参数。
driver_errorBaseException, optional: 当前异常链中最底层的数据库驱动异常。
cleanup_errorBaseException, optional: 关闭游标、连接或查询流时发生的次要异常。

参考样例

>>> try:
...     db.execute("DELETE FROM events WHERE id=%s", params=(1,))
... except DatabaseError as exc:
...     print(exc)          # 显示 SQL 和数据库原始错误，但不显示绑定参数
...     print(exc.params)   # 确有排查需要时显式查看参数

参数:

message (str)
sql (str | None)
params (Any)
driver_error (BaseException | None)
cleanup_error (BaseException | None)

property sql: str | None: 当前异常链中实际执行的 SQL。

property params: Any: SQL 绑定参数；为避免敏感信息泄露，不进入异常文本。

property driver_error: BaseException | None: 当前异常链中最底层的数据库驱动异常。

property cleanup_error: BaseException | None: 主操作失败后发生的资源清理异常。

exception hscredit.database.exceptions.DatabaseConnectionError(message, *, sql=None, params=None, driver_error=None, cleanup_error=None)[源代码]

基类：DatabaseError

数据库连接或连接池操作失败。

参数:

message (str)
sql (str | None)
params (Any)
driver_error (BaseException | None)
cleanup_error (BaseException | None)

exception hscredit.database.exceptions.DatabaseQueryError(message, *, sql=None, params=None, driver_error=None, cleanup_error=None)[源代码]

基类：DatabaseError

数据库查询或 SQL 执行失败。

参数:

message (str)
sql (str | None)
params (Any)
driver_error (BaseException | None)
cleanup_error (BaseException | None)

exception hscredit.database.exceptions.DatabaseWriteError(message, *, result=None, sql=None, params=None, driver_error=None, cleanup_error=None)[源代码]

基类：DatabaseError

数据库写入失败，并可携带部分写入结果。

除 DatabaseError 的 SQL 上下文属性外，result 保存失败前已经提交的批次和行数统计。

参数:

message (str)
result (WriteResult | None)
sql (str | None)
params (Any)
driver_error (BaseException | None)
cleanup_error (BaseException | None)

exception hscredit.database.exceptions.DatabaseMetadataError(message, *, sql=None, params=None, driver_error=None, cleanup_error=None)[源代码]

基类：DatabaseError

数据库元数据读取或导出失败。

参数:

message (str)
sql (str | None)
params (Any)
driver_error (BaseException | None)
cleanup_error (BaseException | None)

exception hscredit.database.exceptions.DatabaseCapabilityError(message, *, sql=None, params=None, driver_error=None, cleanup_error=None)[源代码]

基类：DatabaseError

数据库或目标表不支持所请求的能力。

参数:

message (str)
sql (str | None)
params (Any)
driver_error (BaseException | None)
cleanup_error (BaseException | None)

数据库 hscredit.database

数据库门面

类外快捷操作

流式查询

JSON 投影扩展

适配器注册

公共类型

异常

数据库 `hscredit.database`