核心查询函数
chdb.query
使用 chDB 引擎执行 SQL 查询。
这是主要的查询函数,使用嵌入式 ClickHouse 引擎执行 SQL 语句。支持多种输出格式,并且可用于临时数据库 或基于文件的数据库。
语法
chdb.query(sql, output_format='CSV', path='', udf_path='')参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
sql |
str | required | 要执行的 SQL 查询字符串 |
output_format |
str | "CSV" |
结果的输出格式。支持的格式: • "CSV" - 逗号分隔值• "JSON" - JSON 格式• "Arrow" - Apache Arrow 格式• "Parquet" - Parquet 格式• "DataFrame" - Pandas DataFrame• "ArrowTable" - PyArrow Table• "Debug" - 启用详细日志 |
path |
str | "" |
数据库文件路径。默认为临时的非持久化数据库 (等同于 ":memory:") 。传入文件路径以持久化到磁盘 |
udf_path |
str | "" |
基于旧版子进程的 UDF 目录路径。原生 Python UDF (@func / create_function) 无需此参数 |
返回值
以指定的格式返回查询结果:
| 返回类型 | 条件 |
|---|---|
str |
用于 CSV、JSON 等文本格式 |
pd.DataFrame |
当 output_format 为 "DataFrame" 或 "dataframe" 时 |
pa.Table |
当 output_format 为 "ArrowTable" 或 "arrowtable" 时 |
| chdb result object | 用于其他格式 |
引发异常
| 异常 | 条件 |
|---|---|
ChdbError |
如果 SQL 查询执行失败 |
ImportError |
如果 DataFrame/Arrow 格式所需的依赖项缺失 |
示例
>>> # Basic CSV query
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello">>> # Query with DataFrame output
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
id msg
0 1 hello>>> # Query with file-based database
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")chdb.sql
使用 chDB 引擎执行 SQL 查询。
这是主要的查询函数,使用嵌入式 ClickHouse 引擎执行 SQL 语句。支持多种输出格式,并且可用于临时 或基于文件的数据库。
语法
chdb.sql(sql, output_format='CSV', path='', udf_path='')参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
sql |
str | required | 要执行的 SQL 查询字符串 |
output_format |
str | "CSV" |
结果的输出格式。支持的格式: • "CSV" - 逗号分隔值• "JSON" - JSON 格式• "Arrow" - Apache Arrow 格式• "Parquet" - Parquet 格式• "DataFrame" - Pandas DataFrame• "ArrowTable" - PyArrow Table• "Debug" - 启用详细日志 |
path |
str | "" |
数据库文件路径。默认使用临时的非持久化数据库 (等同于 ":memory:") 。传入文件路径以持久保存到磁盘 |
udf_path |
str | "" |
旧版基于子进程的 UDF 目录路径。原生 Python UDF (@func / create_function) 无需使用 |
返回值
按指定格式返回查询结果:
| 返回类型 | 条件 |
|---|---|
str |
适用于 CSV、JSON 等文本格式 |
pd.DataFrame |
当 output_format 为 "DataFrame" 或 "dataframe" 时 |
pa.Table |
当 output_format 为 "ArrowTable" 或 "arrowtable" 时 |
| chdb result object | 适用于其他格式 |
引发异常
| 异常 | 条件 |
|---|---|
ChdbError |
如果 SQL 查询执行失败 |
ImportError |
如果缺少 DataFrame/Arrow 格式所需的依赖项 |
示例
>>> # Basic CSV query
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello">>> # Query with DataFrame output
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
id msg
0 1 hello>>> # Query with file-based database
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")chdb.to_arrowTable
将查询结果转换为 PyArrow 表。
将 chDB 的查询结果转换为 PyArrow 表,以高效处理列式数据。 如果结果为空,则返回空表。
语法
chdb.to_arrowTable(res)参数
| 参数 | 描述 |
|---|---|
res |
包含二进制 Arrow 数据的 chDB 查询结果对象 |
返回值
| 返回类型 | 描述 |
|---|---|
pa.Table |
包含查询结果的 PyArrow 表 |
引发的异常
| 错误类型 | 描述 |
|---|---|
ImportError |
如果未安装 pyarrow 或 pandas |
示例
>>> result = chdb.query("SELECT 1 as id, 'hello' as msg", "Arrow")
>>> table = chdb.to_arrowTable(result)
>>> print(table.to_pandas())
id msg
0 1 hellochdb.to_df
将查询结果转换为 pandas DataFrame。
先将 chDB 的查询结果转换为 PyArrow Table,再使用多线程将其转换为 pandas,以提升性能。
语法
chdb.to_df(r)参数
| 参数 | 描述 |
|---|---|
r |
包含二进制 Arrow 数据的 chDB 查询结果对象 |
返回值
| 返回类型 | 描述 |
|---|---|
pd.DataFrame |
包含查询结果的 pandas DataFrame |
引发异常
| 异常 | 条件 |
|---|---|
ImportError |
如果未安装 pyarrow 或 pandas |
示例
>>> result = chdb.query("SELECT 1 as id, 'hello' as msg", "Arrow")
>>> df = chdb.to_df(result)
>>> print(df)
id msg
0 1 hello连接与会话管理
可使用以下会话函数:
chdb.connect
创建到 chDB 后台 server 的连接。
此函数会建立一个到 chDB (ClickHouse) database engine 的 Connection。 每次调用都会返回独立的连接,并且可同时打开任意数量连接到同一数据库路径的连接。
chdb.connect(connection_string: str = ':memory:') → Connection参数:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
connection_string |
str | ":memory:" |
数据库连接字符串。参见下方格式。 |
基础版格式
| 格式 | 说明 |
|---|---|
":memory:" |
临时数据库 (默认) |
"test.db" |
相对路径数据库文件 |
"file:test.db" |
与相对路径相同 |
"/path/to/test.db" |
绝对路径数据库文件 |
"file:/path/to/test.db" |
与绝对路径相同 |
带查询参数
| 格式 | 说明 |
|---|---|
"file:test.db?param1=value1¶m2=value2" |
带参数的相对路径 |
"file::memory:?verbose&log-level=test" |
带参数的临时数据库 |
"///path/to/test.db?param1=value1¶m2=value2" |
带参数的绝对路径 |
查询参数处理
查询参数会作为启动参数传递给 ClickHouse 引擎。 特殊参数处理如下:
| 特殊参数 | 转换为 | 说明 |
|---|---|---|
mode=ro |
--readonly=1 |
只读模式 |
verbose |
(flag) | 启用详细日志 |
log-level=test |
(setting) | 设置日志级别 |
如需完整参数列表,请参见 clickhouse local --help --verbose
返回值
| 返回类型 | 说明 |
|---|---|
Connection |
数据库连接对象,支持: • 使用 Connection.cursor() 创建游标• 使用 Connection.query() 直接执行查询• 使用 Connection.send_query() 执行流式查询• 支持上下文管理器协议,可自动清理 |
抛出异常
| 异常 | 条件 |
|---|---|
RuntimeError |
连接数据库失败时 |
示例
>>> # Temporary database
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # File-based database
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # With parameters
>>> conn = connect("data.db?mode=ro") # Read-only mode
>>> conn = connect(":memory:?verbose&log-level=debug") # Debug logging
>>>
>>> # Using context manager for automatic cleanup
>>> with connect("data.db") as conn:
... result = conn.query("SELECT 1")
... print(result)
>>> # Connection automatically closed另请参阅
Connection- 数据库连接类Cursor- 用于 DB-API 2.0 操作的数据库游标
异常处理
class chdb.ChdbError
基类:Exception
与 chDB 相关错误的基础异常类。
当 chDB 查询执行失败或发生 错误时,会引发此异常。它继承自标准 Python 的 Exception 类, 并提供来自底层 ClickHouse 引擎的错误信息。
class chdb.session.Session
基类:object
Session 会保留查询状态。
如果 path 为 None,会话会使用进程范围的临时 (:memory:) 数据库,因此所有未指定 path 的会话 都能看到彼此的表;仅当最后一个此类会话 或 connection 关闭时,其临时目录才会被删除。
你也可以传入一个 path,在该 path 创建数据库,数据将保存在那里。
你也可以使用连接字符串传入该 path 和其他参数。
class chdb.session.Session(path=None)示例
| Connection String | 描述 |
|---|---|
":memory:" |
临时数据库 |
"test.db" |
相对路径 |
"file:test.db" |
同上 |
"/path/to/test.db" |
绝对路径 |
"file:/path/to/test.db" |
同上 |
"file:test.db?param1=value1¶m2=value2" |
带查询参数的相对路径 |
"file::memory:?verbose&log-level=test" |
带查询参数的临时数据库 |
"///path/to/test.db?param1=value1¶m2=value2" |
带查询参数的绝对路径 |
cleanup
在处理异常时清理会话资源。
此方法会尝试关闭会话,同时抑制清理过程中可能发生的任何异常。在异常处理场景中,或者当你需要确保无论会话处于何种状态都能执行清理时,它尤其有用。
语法
cleanup()示例
>>> session = Session("test.db")
>>> try:
... session.query("INVALID SQL")
... finally:
... session.cleanup() # 无论是否发生错误,均可安全清理另请参阅
close()- 用于显式关闭会话,并传播错误
close
关闭会话并清理资源。
此方法会关闭底层连接并重置全局会话状态。 调用此方法后,会话将失效,无法再用于 后续查询。
语法
close()示例
>>> session = Session("test.db")
>>> session.query("SELECT 1")
>>> session.close() # 显式关闭会话query
执行 SQL 查询并返回结果。
此方法会在会话的数据库中执行 SQL 查询,并以指定格式返回结果。该方法支持多种输出格式,并在多次查询之间保持会话状态。
语法
query(sql, fmt='CSV', udf_path='')参数
| Parameter | Type | Default | Description |
|---|---|---|---|
sql |
str | required | 要执行的 SQL 查询字符串 |
fmt |
str | "CSV" |
结果的输出格式。可用格式: • "CSV" - 逗号分隔值• "JSON" - JSON 格式• "TabSeparated" - 制表符分隔值• "Pretty" - 美化表格格式• "JSONCompact" - 紧凑 JSON 格式• "Arrow" - Apache Arrow 格式• "Parquet" - Parquet 格式 |
udf_path |
str | "" |
旧版基于子进程的 UDF 目录路径。原生 Python UDF (@func / create_function) 无需此项。如果未指定,则使用会话初始化时的 UDF 路径 |
返回值
以指定格式返回查询结果。
具体返回类型取决于 fmt 参数:
- String 格式 (CSV、JSON 等) 返回
str - 二进制格式 (Arrow、Parquet) 返回
bytes
引发异常
| 异常 | Condition |
|---|---|
RuntimeError |
如果会话已关闭或无效 |
ValueError |
如果 SQL 查询格式有误 |
示例
>>> session = Session("test.db")
>>>
>>> # Basic query with default CSV format
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1>>> # Query with JSON format
>>> result = session.query("SELECT 1 as number", fmt="JSON")
>>> print(result)
{"number": "1"}>>> # Complex query with table creation
>>> session.query("CREATE TABLE test (id INT, name String) ENGINE = Memory")
>>> session.query("INSERT INTO test VALUES (1, 'Alice'), (2, 'Bob')")
>>> result = session.query("SELECT * FROM test ORDER BY id")
>>> print(result)
id,name
1,Alice
2,Bob参见
send_query()- 用于以流式方式执行查询sql- 该方法的别名
send_query
执行 SQL 查询,并返回一个流式结果迭代器。
此方法会针对会话的数据库执行 SQL 查询,并返回一个流式结果对象,让你能够遍历结果,而无需一次将全部内容加载到内存中。这对于大型结果集尤其有用。
语法
send_query(sql, fmt='CSV') → StreamingResult参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
sql |
str | required | 要执行的 SQL 查询字符串 |
fmt |
str | "CSV" |
结果的输出格式。可用格式: • "CSV" - 逗号分隔值• "JSON" - JSON 格式• "TabSeparated" - 制表符分隔值• "JSONCompact" - 紧凑 JSON 格式• "Arrow" - Apache Arrow 格式• "Parquet" - Parquet 格式 |
返回值
| 返回类型 | 描述 |
|---|---|
StreamingResult |
一个流式结果迭代器,会以增量方式返回查询结果。该迭代器可用于 for 循环,也可转换为其他数据结构 |
抛出异常
| 异常 | 条件 |
|---|---|
RuntimeError |
如果会话已关闭或无效 |
ValueError |
如果 SQL 查询格式有误 |
示例
>>> session = Session("test.db")
>>> session.query("CREATE TABLE big_table (id INT, data String) ENGINE = MergeTree() order by id")
>>>
>>> # 插入大型数据集
>>> for i in range(1000):
... session.query(f"INSERT INTO big_table VALUES ({i}, 'data_{i}')")
>>>
>>> # 流式传输结果以避免内存问题
>>> streaming_result = session.send_query("SELECT * FROM big_table ORDER BY id")
>>> for chunk in streaming_result:
... print(f"Processing chunk: {len(chunk)} bytes")
... # 处理数据块,无需将整个结果集加载到内存中>>> # Using with context manager
>>> with session.send_query("SELECT COUNT(*) FROM big_table") as stream:
... for result in stream:
... print(f"Count result: {result}")另请参见
query()- 用于执行非流式查询chdb.state.sqlitelike.StreamingResult- 流式结果迭代器
sql
执行 SQL 查询并返回结果。
此方法会在会话的数据库中执行 SQL 查询,并以指定格式返回结果。该方法支持多种输出格式,并在多次查询之间保持会话状态。
语法
sql(sql, fmt='CSV', udf_path='')参数
| Parameter | Type | Default | Description |
|---|---|---|---|
sql |
str | required | 要执行的 SQL 查询字符串 |
fmt |
str | "CSV" |
结果的输出格式。可用格式: • "CSV" - 逗号分隔值• "JSON" - JSON 格式• "TabSeparated" - 制表符分隔值• "Pretty" - 格式化表格输出• "JSONCompact" - 紧凑 JSON 格式• "Arrow" - Apache Arrow 格式• "Parquet" - Parquet 格式 |
udf_path |
str | "" |
旧版基于子进程的 UDF 目录路径。原生 Python UDF (@func / create_function) 不需要此路径。如果未指定,则使用会话初始化时的 UDF 路径 |
返回值
以指定格式返回查询结果。
具体返回类型取决于 fmt 参数:
- String 格式 (CSV、JSON 等) 返回 str
- 二进制格式 (Arrow、Parquet) 返回 bytes
引发:
| 异常 | Condition |
|---|---|
RuntimeError |
如果会话已关闭或无效 |
ValueError |
如果 SQL 查询格式错误 |
示例
>>> session = Session("test.db")
>>>
>>> # Basic query with default CSV format
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1>>> # Query with JSON format
>>> result = session.query("SELECT 1 as number", fmt="JSON")
>>> print(result)
{"number": "1"}>>> # Complex query with table creation
>>> session.query("CREATE TABLE test (id INT, name String) ENGINE = MergeTree() order by id")
>>> session.query("INSERT INTO test VALUES (1, 'Alice'), (2, 'Bob')")
>>> result = session.query("SELECT * FROM test ORDER BY id")
>>> print(result)
id,name
1,Alice
2,Bob另请参见
send_query()- 用于流式执行查询sql- 该方法的别名
状态管理
chdb.state.connect
创建一个到 chDB 后台服务器的 Connection。
此函数会建立与 chDB (ClickHouse) 数据库引擎的连接。 每次调用都会返回独立的连接,并且可同时打开任意数量指向同一数据库路径的连接。
语法
chdb.state.connect(connection_string: str = ':memory:') → Connection参数
| Parameter | Type | Default | Description |
|---|---|---|---|
connection_string(str, optional) |
str | ":memory:" |
数据库连接字符串。请参见下方格式。 |
基础格式
支持的连接字符串格式:
| Format | Description |
|---|---|
":memory:" |
临时数据库 (默认) |
"test.db" |
相对路径数据库文件 |
"file:test.db" |
与相对路径相同 |
"/path/to/test.db" |
绝对路径数据库文件 |
"file:/path/to/test.db" |
与绝对路径相同 |
带查询参数
| Format | Description |
|---|---|
"file:test.db?param1=value1¶m2=value2" |
带参数的相对路径 |
"file::memory:?verbose&log-level=test" |
带参数的临时数据库 |
"///path/to/test.db?param1=value1¶m2=value2" |
带参数的绝对路径 |
查询参数处理
查询参数会作为启动参数传递给 ClickHouse 引擎。 特殊参数处理如下:
| Special Parameter | Becomes | Description |
|---|---|---|
mode=ro |
--readonly=1 |
只读模式 |
verbose |
(flag) | 启用详细日志 |
log-level=test |
(setting) | 设置日志级别 |
完整参数列表请参见 clickhouse local --help --verbose
返回值
| Return Type | Description |
|---|---|
Connection |
数据库连接对象,支持: • 使用 Connection.cursor() 创建游标• 使用 Connection.query() 直接执行查询• 使用 Connection.send_query() 执行流式查询• 支持上下文管理器协议以自动清理 |
引发
| Exception | Condition |
|---|---|
RuntimeError |
连接数据库失败时 |
示例
>>> # Temporary database
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # File-based database
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # With parameters
>>> conn = connect("data.db?mode=ro") # Read-only mode
>>> conn = connect(":memory:?verbose&log-level=debug") # Debug logging
>>>
>>> # Using context manager for automatic cleanup
>>> with connect("data.db") as conn:
... result = conn.query("SELECT 1")
... print(result)
>>> # Connection automatically closed另请参见
Connection- 数据库连接类Cursor- 用于 DB-API 2.0 操作的数据库游标
class chdb.state.sqlitelike.Connection
基类:object
语法
class chdb.state.sqlitelike.Connection(connection_string: str)close
关闭连接并清理资源。
此方法会关闭数据库连接,并清理所有相关资源, 包括活动游标。调用此方法后, 该连接将失效,无法再用于后续操作。
语法
close() → None示例
>>> conn = connect("test.db")
>>> # 使用连接执行查询
>>> conn.query("CREATE TABLE test (id INT) ENGINE = Memory")
>>> # 完成后关闭连接
>>> conn.close()>>> # 使用上下文管理器(自动清理)
>>> with connect("test.db") as conn:
... conn.query("SELECT 1")
... # 连接自动关闭cursor
创建一个用于执行查询的 Cursor 对象。
此方法会创建一个数据库游标,提供执行查询和获取结果所需的标准 DB-API 2.0 接口。 该游标可对查询执行 和结果获取进行更细粒度的控制。
语法
cursor() → Cursor返回值
| 返回类型 | 描述 |
|---|---|
Cursor |
用于数据库操作的游标对象 |
示例
>>> conn = connect(":memory:")
>>> cursor = conn.cursor()
>>> cursor.execute("CREATE TABLE test (id INT, name String) ENGINE = Memory")
>>> cursor.execute("INSERT INTO test VALUES (1, 'Alice')")
>>> cursor.execute("SELECT * FROM test")
>>> rows = cursor.fetchall()
>>> print(rows)
((1, 'Alice'),)另请参见
Cursor- 数据库游标的实现
query
执行 SQL 查询并返回完整结果。
此方法会同步执行 SQL 查询,并返回完整的 结果集。它支持多种输出格式,并会自动进行 格式特定的后处理。
语法
query(query: str, format: str = 'CSV') → Any参数:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
query |
str | required | 要执行的 SQL 查询字符串 |
format |
str | "CSV" |
结果的输出格式。支持的格式: • "CSV" - 逗号分隔值 (字符串) • "JSON" - JSON 格式 (字符串) • "Arrow" - Apache Arrow 格式 (字节) • "Dataframe" - Pandas DataFrame (需要 pandas) • "Arrowtable" - PyArrow Table (需要 pyarrow) |
返回值
| 返回类型 | 说明 |
|---|---|
str |
用于字符串格式 (CSV、JSON) |
bytes |
用于 Arrow 格式 |
pandas.DataFrame |
用于 dataframe 格式 |
pyarrow.Table |
用于 arrowtable 格式 |
引发的异常
| 异常 | 条件 |
|---|---|
RuntimeError |
如果查询执行失败 |
ImportError |
如果未安装该格式所需的依赖包 |
示例
>>> conn = connect(":memory:")
>>>
>>> # 基础 CSV 查询
>>> result = conn.query("SELECT 1 as num, 'hello' as text")
>>> print(result)
num,text
1,hello>>> # DataFrame 格式
>>> df = conn.query("SELECT number FROM numbers(5)", "dataframe")
>>> print(df)
number
0 0
1 1
2 2
3 3
4 4另请参见
send_query()- 用于以流式方式执行查询
send_query
执行 SQL 查询并返回一个流式结果迭代器。
此方法会执行 SQL 查询,并返回一个 StreamingResult 对象, 让你能够遍历结果,而不必一次性将所有内容 加载到内存中。这非常适合处理大型结果集。
语法
send_query(query: str, format: str = 'CSV') → StreamingResult参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
query |
str | 必填 | 要执行的 SQL 查询字符串 |
format |
str | "CSV" |
结果的输出格式。支持的格式: • "CSV" - 逗号分隔值• "JSON" - JSON 格式• "Arrow" - Apache Arrow 格式 (启用 record_batch() 方法) • "dataframe" - Pandas DataFrame 分块• "arrowtable" - PyArrow Table 分块 |
返回
| 返回类型 | 描述 |
|---|---|
StreamingResult |
用于处理查询结果的流式迭代器,支持: • 迭代器协议 (for 循环) • 上下文管理器协议 (with 语句) • 使用 fetch() 方法手动拉取 • PyArrow RecordBatch 流式传输 (仅限 Arrow 格式) |
引发
| 异常 | 条件 |
|---|---|
RuntimeError |
查询执行失败时 |
ImportError |
未安装该格式所需的包时 |
示例
>>> conn = connect(":memory:")
>>>
>>> # 基础流式传输
>>> stream = conn.send_query("SELECT number FROM numbers(1000)")
>>> for 数据块 in stream:
... print(f"正在处理数据块:{len(chunk)} 字节")>>> # 使用上下文管理器进行清理
>>> with conn.send_query("SELECT * FROM large_table") as stream:
... 数据块 = stream.fetch()
... while chunk:
... process_data(chunk)
... chunk = stream.fetch()>>> # Arrow 格式与 RecordBatch 流式传输
>>> stream = conn.send_query("SELECT * FROM data", "Arrow")
>>> reader = stream.record_batch(rows_per_batch=10000)
>>> for batch in reader:
... print(f"Batch shape: {batch.num_rows} x {batch.num_columns}")另请参阅
query()- 用于执行非流式查询StreamingResult- 流式结果迭代器
class chdb.state.sqlitelike.StreamingResult
基类:object
用于处理大型查询结果的流式结果迭代器。
此类提供迭代器接口,可在不将整个结果集加载到内存中的情况下流式处理查询结果。它支持多种输出格式,并提供用于手动拉取结果和流式传输 PyArrow RecordBatch 的方法。
class chdb.state.sqlitelike.StreamingResultfetch
拉取下一批流式结果。
此方法会从流式查询结果中拉取下一批可用数据。返回数据的格式取决于发起流式查询时指定的格式。
语法
fetch() → Any返回值
| 返回类型 | 描述 |
|---|---|
str |
用于文本格式 (CSV、JSON) |
bytes |
用于二进制格式 (Arrow、Parquet) |
None |
当结果流已耗尽时 |
示例
>>> stream = conn.send_query("SELECT * FROM large_table")
>>> chunk = stream.fetch()
>>> while chunk is not None:
... process_data(chunk)
... chunk = stream.fetch()cancel
取消流式查询并清理资源。
此方法会取消任何正在进行的流式查询,并释放相关 资源。当你希望在流耗尽之前停止处理结果时, 应调用它。
语法
cancel() → None示例
>>> stream = conn.send_query("SELECT * FROM very_large_table")
>>> for i, 数据块 in enumerate(stream):
... if i >= 10: # 仅处理前 10 个 数据块
... stream.cancel()
... break
... process_data(chunk)close
关闭流式结果并清理资源。
cancel() 的别名。用于关闭流式结果迭代器,
并释放相关资源。
语法
close() → Nonerecord_batch
创建一个 PyArrow RecordBatchReader,以高效进行批次处理。
该方法会创建一个 PyArrow RecordBatchReader,使您能够高效地 迭代 Arrow 格式的查询结果。这是在使用 PyArrow 时处理大型 结果集的最高效方式。
语法
record_batch(rows_per_batch: int = 1000000) → pa.RecordBatchReader参数
| Parameter | Type | Default | Description |
|---|---|---|---|
rows_per_batch |
int | 1000000 |
每个批次的行数 |
返回值
| Return Type | Description |
|---|---|
pa.RecordBatchReader |
用于遍历各批次的 PyArrow RecordBatchReader |
示例
>>> stream = conn.send_query("SELECT * FROM data", format="Arrow")
>>> reader = stream.record_batch(rows_per_batch=10000)
>>> for batch in reader:
... print(f"Processing batch: {batch.num_rows} rows")
... df = batch.to_pandas()
... process_dataframe(df)迭代器协议
StreamingResult 支持 Python 的迭代器协议,因此可以 直接用于 for 循环:
>>> stream = conn.send_query("SELECT number FROM numbers(1000000)")
>>> for 数据块 in stream:
... print(f"Chunk size: {len(chunk)} bytes")上下文管理器协议
StreamingResult 支持上下文管理器协议,可自动清理资源:
>>> with conn.send_query("SELECT * FROM data") as stream:
... for chunk in stream:
... process(chunk)
>>> # Stream 已自动关闭类 chdb.state.sqlitelike.Cursor
基类:object
class chdb.state.sqlitelike.Cursor(connection)close
关闭游标并释放资源。
此方法会关闭游标并释放所有相关资源。 调用此方法后,游标将失效,且无法再 用于后续操作。
语法
close() → None示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchone()
>>> cursor.close() # 清理游标资源column_names
返回最近一次执行的查询的列名列表。
此方法返回最近一次执行的 SELECT 查询的列名。返回的名称顺序与它们在结果集中的顺序 一致。
语法
column_names() → list返回值
| 返回类型 | 说明 |
|---|---|
list |
由列名字符串组成的列表;如果尚未执行任何查询,或查询未返回任何列,则为空列表 |
示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name, email FROM users LIMIT 1")
>>> print(cursor.column_names())
['id', 'name', 'email']另请参阅
column_types()- 获取列类型信息description- DB-API 2.0 列描述
column_types
返回上一次执行的查询中的列类型列表。
此方法返回最近一次执行的 SELECT 查询的 ClickHouse 列类型名称。返回的类型顺序与其在结果集中的出现顺序一致。
语法
column_types() → list返回值
| 返回类型 | 说明 |
|---|---|
list |
由 ClickHouse 类型名称字符串组成的列表;如果尚未执行任何查询,或查询未返回任何列,则返回空列表 |
示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT toInt32(1), toString('hello')")
>>> print(cursor.column_types())
['Int32', 'String']另请参阅
column_names()- 获取列名信息description- DB-API 2.0 列说明
commit
提交所有待处理的事务。
此方法会提交所有待处理的数据库事务。在 ClickHouse 中, 大多数操作都会自动提交,但提供此方法是为了兼容 DB-API 2.0。
语法
commit() → None示例
>>> cursor = conn.cursor()
>>> cursor.execute("INSERT INTO test VALUES (1, 'data')")
>>> cursor.commit()property description : list
按照 DB-API 2.0 规范返回列描述。
此属性会返回一个包含 7 项 Tuple 的列表,用于描述最近一次执行的 SELECT 查询结果集中的每一列。每个 Tuple 包含: (name, type_code, display_size, internal_size, precision, scale, null_ok)
目前,仅提供 name 和 type_code,其他字段均设为 None。
返回值
| 返回类型 | 描述 |
|---|---|
list |
由描述各列的 7 项 Tuple 组成的列表;如果尚未执行任何 SELECT 查询,则为空列表 |
示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name FROM users LIMIT 1")
>>> for desc in cursor.description:
... print(f"Column: {desc[0]}, Type: {desc[1]}")
Column: id, Type: Int32
Column: name, Type: String另请参见
column_names()- 仅获取列名column_types()- 仅获取列类型
execute
执行 SQL 查询,并为后续拉取结果做好准备。
此方法执行 SQL 查询,并将结果准备为可通过 fetch 方法获取。它会处理结果数据的解析,以及 ClickHouse 数据类型的自动转换。
语法
execute(query: str) → None参数:
| Parameter | Type | Description |
|---|---|---|
query |
str | 要执行的 SQL 查询字符串 |
引发
| Exception | Condition |
|---|---|
Exception |
如果查询执行失败,或结果解析失败 |
示例
>>> cursor = conn.cursor()
>>>
>>> # 执行 DDL
>>> cursor.execute("CREATE TABLE test (id INT, name String) ENGINE = Memory")
>>>
>>> # 执行 DML
>>> cursor.execute("INSERT INTO test VALUES (1, 'Alice')")
>>>
>>> # 执行 SELECT 并获取结果
>>> cursor.execute("SELECT * FROM test")
>>> rows = cursor.fetchall()
>>> print(rows)
((1, 'Alice'),)另请参见
fetchone()- 获取单行fetchmany()- 获取多行fetchall()- 获取所有剩余行
fetchall
从查询结果中拉取所有剩余行。
此方法从当前游标位置开始, 拉取当前查询结果集中所有剩余行。它返回一个由行 Tuple 组成的 Tuple,并应用适当的 Python 类型转换。
语法
fetchall() → tuple返回值:
| Return Type | Description |
|---|---|
tuple |
包含结果集中所有剩余行的 Tuple。若没有可用的行,则返回空 Tuple |
示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name FROM users")
>>> all_users = cursor.fetchall()
>>> for user_id, user_name in all_users:
... print(f"User {user_id}: {user_name}")另请参阅
fetchone()- 拉取单行数据fetchmany()- 分批拉取多行数据
fetchmany
从查询结果中拉取多行。
此方法会从当前查询结果集中最多拉取 ‘size’ 行。它返回一个由行 Tuple 组成的 Tuple,其中每一行都包含经过适当 Python 类型转换的列值。
语法
fetchmany(size: int = 1) → tuple参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
size |
int | 1 |
最多可拉取的行数 |
返回值
| 返回类型 | 描述 |
|---|---|
tuple |
包含最多 'size' 个行元组的 tuple。如果结果集已耗尽,返回的行数可能会更少 |
示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT * FROM large_table")
>>>
>>> # 批量处理结果
>>> while True:
... batch = cursor.fetchmany(100) # 每次拉取 100 行
... if not batch:
... break
... process_batch(batch)另请参阅
fetchone()- 拉取单行fetchall()- 拉取所有剩余行
fetchone
从查询结果中拉取下一行。
此方法会从当前查询结果集 中拉取下一条可用记录。它会返回一个包含各列值的 Tuple, 并进行适当的 Python 类型转换。
语法
fetchone() → tuple | None返回:
| 返回类型 | 描述 |
|---|---|
Optional[tuple] |
下一行,以由各列值组成的 Tuple 返回;如果没有更多行,则为 None |
示例
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name FROM users")
>>> row = cursor.fetchone()
>>> while row is not None:
... user_id, user_name = row
... print(f"User {user_id}: {user_name}")
... row = cursor.fetchone()另请参阅
fetchmany()- 拉取多行fetchall()- 拉取所有剩余行
chdb.state.sqlitelike
将查询结果转换为 PyArrow Table。
此函数会将 chdb 的查询结果转换为 PyArrow Table 格式, 从而实现高效的列式数据访问,并可与其他数据处理库 互操作。
语法
chdb.state.sqlitelike.to_arrowTable(res)参数:
| Parameter | Type | Description |
|---|---|---|
res |
- | 来自 chdb 的查询结果对象,包含 Arrow 格式 格式的数据 |
返回值
| Return Type | Description |
|---|---|
pyarrow.Table |
包含查询结果的 PyArrow Table 对象 |
引发异常
| Exception | Condition |
|---|---|
ImportError |
如果未安装 pyarrow 或 pandas 包 |
示例
>>> import chdb
>>> result = chdb.query("SELECT 1 as num, 'hello' as text", "Arrow")
>>> table = to_arrowTable(result)
>>> print(table.schema)
num: int64
text: string
>>> print(table.to_pandas())
num text
0 1 hellochdb.state.sqlitelike.to_df
将查询结果转换为 Pandas DataFrame。
此函数会先将 chdb 查询结果转换为 PyArrow Table,然后再转换为 DataFrame,从而能够通过 Pandas API 方便地进行数据分析。
语法
chdb.state.sqlitelike.to_df(r)参数:
| 参数 | Type | 说明 |
|---|---|---|
r |
- | 来自 chdb 的查询结果对象,包含 Arrow 格式 数据 |
返回:
| 返回类型 | 说明 |
|---|---|
pandas.DataFrame |
包含查询结果的 DataFrame,并具有相应的列名和数据类型 |
引发
| Exception | 条件 |
|---|---|
ImportError |
如果未安装 pyarrow 或 pandas 包 |
另请参阅
to_arrowTable()- 用于转换为 PyArrow Table 格式
示例
>>> import chdb
>>> result = chdb.query("SELECT 1 as num, 'hello' as text", "Arrow")
>>> df = to_df(result)
>>> print(df)
num text
0 1 hello
>>> print(df.dtypes)
num int64
text object
dtype: objectDataFrame 集成
类 chdb.dataframe.Table
基类:
class chdb.dataframe.Table(*args: Any, **kwargs: Any)数据库 API (DBAPI) 2.0 接口
chDB 提供兼容 Python DB-API 2.0 的数据库连接接口,使你能够将 chDB 与需要标准数据库接口的工具和框架配合使用。
chDB DB-API 2.0 接口包括:
- 连接:使用 connection string 管理数据库连接
- 游标:执行查询并获取结果
- 类型系统:符合 DB-API 2.0 规范的类型常量和转换器
- 错误处理:标准的数据库异常层次结构
- 线程安全:1 级线程安全 (线程可共享模块,但不可共享连接)
核心函数
Database API (DBAPI) 2.0 接口提供以下核心函数:
chdb.dbapi.connect
初始化新的数据库连接。
语法
chdb.dbapi.connect(*args, **kwargs)参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
path |
str | None |
数据库文件路径。临时非持久化数据库时为 None |
抛出
| 异常 | 条件 |
|---|---|
err.Error |
如果无法建立连接 |
chdb.dbapi.get_client_info()
获取客户端版本信息。
以字符串形式返回 chDB 客户端版本号,以兼容 MySQLdb。
语法
chdb.dbapi.get_client_info()返回值
| 返回类型 | 描述 |
|---|---|
str |
采用 'major.minor.patch' 格式的版本字符串 |
类型构造器
chdb.dbapi.Binary(x)
将 x 以二进制类型返回。
此函数会将输入转换为 bytes 类型,以便根据 DB-API 2.0 规范用于二进制 database 字段。
语法
chdb.dbapi.Binary(x)参数
| 参数 | 类型 | 描述 |
|---|---|---|
x |
- | 要转换为二进制格式的输入数据 |
返回值
| 返回类型 | 描述 |
|---|---|
bytes |
转换为字节后的输入数据 |
连接类
类 chdb.dbapi.connections.Connection(path=None)
基类:object
符合 DB-API 2.0 规范的 chDB 数据库连接。
此类提供标准的 DB-API 接口,用于连接到 chDB 数据库并与之交互。它同时支持临时数据库和基于文件的数据库。
该连接管理底层的 chDB engine,并提供用于 执行查询、管理事务 (对 ClickHouse 而言是空操作) 以及创建游标的方法。
class chdb.dbapi.connections.Connection(path=None)参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
path |
str | None |
数据库文件路径。如果为 None (默认值) ,则使用临时的非持久化数据库 (等同于 ':memory:') 。传入类似 'database.db' 的文件路径,即可将数据持久化到磁盘。 |
变量
| 变量 | 类型 | 描述 |
|---|---|---|
encoding |
str | 查询的字符编码,默认为 'utf8' |
open |
bool | 如果连接处于打开状态,则为 True;如果已关闭,则为 False |
示例
>>> # Temporary database
>>> conn = Connection()
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchall()
>>> conn.close()>>> # File-based database
>>> conn = Connection('mydata.db')
>>> with conn.cursor() as cur:
... cur.execute("CREATE TABLE users (id INT, name STRING) ENGINE = MergeTree() order by id")
... cur.execute("INSERT INTO users VALUES (1, 'Alice')")
>>> conn.close()>>> # Context manager usage
>>> with Connection() as cur:
... cur.execute("SELECT version()")
... version = cur.fetchone()close
关闭数据库连接。
关闭底层的 chDB 连接,并将该连接标记为已关闭。 此后对该连接执行的操作将引发 Error。
语法
close()抛出
| 异常 | 条件 |
|---|---|
err.Error |
如果连接已关闭 |
commit
提交当前事务。
语法
commit()cursor
创建一个用于执行查询的新游标。
语法
cursor(cursor=None)参数
| 参数 | 类型 | 描述 |
|---|---|---|
cursor |
- | 已忽略,仅为兼容性而保留 |
返回值
| 返回类型 | 描述 |
|---|---|
Cursor |
此 连接 的新游标对象 |
引发
| 异常 | 条件 |
|---|---|
err.Error |
如果 连接 已关闭 |
示例
>>> conn = Connection()
>>> cur = conn.cursor()
>>> cur.execute("SELECT 1")
>>> result = cur.fetchone()escape
对值进行转义,以便安全地插入 SQL 查询中。
语法
escape(obj, mapping=None)参数
| 参数 | 类型 | 描述 |
|---|---|---|
obj |
- | 要转义的值 (字符串、字节、数值等) |
mapping |
- | 用于转义的可选字符映射 |
返回值
| 返回类型 | 描述 |
|---|---|
| - | 经过转义后的输入内容,可用于 SQL 查询 |
示例
>>> conn = Connection()
>>> safe_value = conn.escape("O'Reilly")
>>> query = f"SELECT * FROM users WHERE name = {safe_value}"escape_string
对字符串值进行转义,以用于 SQL 查询。
语法
escape_string(s)参数
| 参数 | 类型 | 描述 |
|---|---|---|
s |
str | 要转义的字符串 |
返回值
| 返回类型 | 描述 |
|---|---|
str |
适合安全包含在 SQL 中的转义后字符串 |
property open
检查连接是否已打开。
返回值
| 返回类型 | 描述 |
|---|---|
bool |
连接已打开时为 True,已关闭时为 False |
query
直接执行 SQL 查询并返回原始结果。
此方法会绕过游标接口,直接执行查询。 对于标准的 DB-API 用法,建议优先使用 cursor() 方法。
语法
query(sql, fmt='CSV')参数:
| Parameter | Type | Default | Description |
|---|---|---|---|
sql |
str or bytes | required | 要执行的 SQL 查询 |
fmt |
str | "CSV" |
输出格式。支持的格式包括 "CSV"、"JSON"、"Arrow"、"Parquet" 等。 |
返回值
| Return Type | Description |
|---|---|
| - | 指定格式的查询结果 |
引发异常
| Exception | Condition |
|---|---|
err.InterfaceError |
如果连接已关闭或查询失败 |
示例
>>> conn = Connection()
>>> result = conn.query("SELECT 1, 'hello'", "CSV")
>>> print(result)
"1,hello\n"property resp
获取最后一次查询的响应。
返回值
| 返回类型 | 描述 |
|---|---|
| - | 最后一次调用 query() 时返回的原始响应 |
rollback
回滚当前事务。
语法
rollback()Cursor 类
class chdb.dbapi.cursors.Cursor
基类:object
用于执行查询并获取结果的 DB-API 2.0 游标。
该游标提供用于执行 SQL 语句、管理查询结果以及浏览结果集的方法。它支持参数绑定、批量操作, 并遵循 DB-API 2.0 规范。
不要直接创建 Cursor 实例。请改用 Connection.cursor()。
class chdb.dbapi.cursors.Cursor(connection)| Variable | 类型 | 描述 |
|---|---|---|
description |
tuple | 上一次查询结果的列元数据 |
rowcount |
int | 上一次查询影响的行数 (如果未知则为 -1) |
arraysize |
int | 一次默认拉取的行数 (默认值:1) |
lastrowid |
- | 最后一条插入行的 ID (如适用) |
max_stmt_length |
int | executemany() 的最大语句长度 (默认值:1024000) |
示例
>>> conn = Connection()
>>> cur = conn.cursor()
>>> cur.execute("SELECT 1 as id, 'test' as name")
>>> result = cur.fetchone()
>>> print(result) # (1, 'test')
>>> cur.close()callproc
执行存储过程 (占位用实现) 。
语法
callproc(procname, args=())参数
| 参数 | Type | 描述 |
|---|---|---|
procname |
str | 要执行的存储过程名称 |
args |
sequence | 传递给存储过程的参数 |
返回值
| Return Type | 描述 |
|---|---|
sequence |
原始 args 参数 (未修改) |
close
关闭游标并释放相关资源。
关闭后,游标将无法再使用,任何操作都会抛出异常。 关闭游标会读取完所有剩余数据,并释放底层游标。
语法
close()execute
执行 SQL 查询,并可选择绑定参数。
此方法执行单条 SQL 语句,并可选择进行参数替换。 它支持多种参数占位符形式,使用更灵活。
语法
execute(query, args=None)参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
query |
str | required | 要执行的 SQL 查询 |
args |
tuple/list/dict | None |
要绑定到占位符的参数 |
返回值
| 返回类型 | 描述 |
|---|---|
int |
受影响的行数 (未知时为 -1) |
参数风格
| 风格 | 示例 |
|---|---|
| Question mark style | "SELECT * FROM users WHERE id = ?" |
| Named style | "SELECT * FROM users WHERE name = %(name)s" |
| Format style | "SELECT * FROM users WHERE age = %s" (旧版) |
示例
>>> # 问号参数
>>> cur.execute("SELECT * FROM users WHERE id = ? AND age > ?", (123, 18))
>>>
>>> # 命名参数
>>> cur.execute("SELECT * FROM users WHERE name = %(name)s", {'name': 'Alice'})
>>>
>>> # 无参数
>>> cur.execute("SELECT COUNT(*) FROM users")抛出
| Exception | 条件 |
|---|---|
ProgrammingError |
如果游标已关闭或查询格式不正确 |
InterfaceError |
如果执行期间发生数据库错误 |
executemany(query, args)
使用不同的参数集多次执行查询。
该方法可高效地对同一 SQL 查询执行多次, 每次使用不同的参数值。它尤其适用于批量 INSERT 操作。
语法
executemany(query, args)参数
| 参数 | 类型 | 说明 |
|---|---|---|
query |
str | 要多次执行的 SQL 查询 |
args |
sequence | 每次执行对应的参数元组/字典/列表序列 |
返回值
| 返回类型 | 说明 |
|---|---|
int |
所有执行中受影响的行总数 |
示例
>>> # 使用问号参数进行批量插入
>>> users_data = [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]
>>> cur.executemany("INSERT INTO users VALUES (?, ?)", users_data)
>>>
>>> # 使用命名参数进行批量插入
>>> users_data = [
... {'id': 1, 'name': 'Alice'},
... {'id': 2, 'name': 'Bob'}
... ]
>>> cur.executemany(
... "INSERT INTO users VALUES (%(id)s, %(name)s)",
... users_data
... )fetchall()
从查询结果中获取所有剩余行。
语法
fetchall()返回值
| 返回类型 | 描述 |
|---|---|
list |
表示所有剩余行的 Tuple 列表 |
抛出
| 异常 | 条件 |
|---|---|
ProgrammingError |
如果尚未调用 execute() |
示例
>>> cursor.execute("SELECT id, name FROM users")
>>> all_rows = cursor.fetchall()
>>> print(len(all_rows)) # 总行数fetchmany
从查询结果中获取多行。
语法
fetchmany(size=1)参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
size |
int | 1 |
要拉取的行数。若未指定,则使用 cursor.arraysize |
返回值
| 返回类型 | 描述 |
|---|---|
list |
由已拉取行组成的元组列表 |
引发异常
| 异常 | 条件 |
|---|---|
ProgrammingError |
如果尚未先调用 execute() |
示例
>>> cursor.execute("SELECT id, name FROM users")
>>> rows = cursor.fetchmany(3)
>>> print(rows) # [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]fetchone
从查询结果中获取下一行。
语法
fetchone()返回值
| 返回类型 | 说明 |
|---|---|
tuple or None |
返回下一行的元组;如果没有更多行,则返回 None |
抛出
| 异常 | 条件 |
|---|---|
ProgrammingError |
如果尚未先调用 execute() |
示例
>>> cursor.execute("SELECT id, name FROM users LIMIT 3")
>>> row = cursor.fetchone()
>>> print(row) # (1, 'Alice')
>>> row = cursor.fetchone()
>>> print(row) # (2, 'Bob')max_stmt_length = 1024000
executemany() 生成的最大语句大小。
默认值为 1024000。
mogrify
返回将发送到数据库的精确查询字符串。
此方法会显示参数替换后的最终 SQL 查询, 这对调试和日志很有帮助。
语法
mogrify(query, args=None)参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
query |
str | 必填 | 包含参数占位符的 SQL 查询 |
args |
tuple/list/dict | None |
用于替换占位符的参数 |
返回值
| 返回类型 | 描述 |
|---|---|
str |
参数替换后的最终 SQL 查询字符串 |
示例
>>> cur.mogrify("SELECT * FROM users WHERE id = ?", (123,))
"SELECT * FROM users WHERE id = 123"nextset
切换到下一个结果集 (不支持) 。
语法
nextset()返回
| 返回类型 | 描述 |
|---|---|
None |
由于不支持多个结果集,因此始终返回 None |
setinputsizes
设置参数输入大小 (实际为空操作) 。
语法
setinputsizes(*args)参数
| 参数 | 类型 | 描述 |
|---|---|---|
*args |
- | 参数大小说明 (忽略) |
setoutputsizes
设置输出列的大小 (空操作实现) 。
语法
setoutputsizes(*args)参数
| 参数 | 类型 | 说明 |
|---|---|---|
*args |
- | 列大小说明 (将被忽略) |
错误类
用于 chdb database 操作的异常类。
该模块提供了完整的异常类层次结构,用于处理 chdb 中与 database 相关的 error,并遵循 Python Database API Specification v2.0。
异常类层次结构如下:
StandardError
├── Warning
└── Error
├── InterfaceError
└── DatabaseError
├── DataError
├── OperationalError
├── IntegrityError
├── InternalError
├── ProgrammingError
└── NotSupportedError每个异常类都表示一类特定的数据库错误:
| Exception | Description |
|---|---|
Warning |
数据库操作期间的非致命警告 |
InterfaceError |
数据库接口本身存在的问题 |
DatabaseError |
所有与数据库相关错误的基类 |
DataError |
数据处理问题 (如无效值、类型错误) |
OperationalError |
数据库运行问题 (如连接性、资源等) |
IntegrityError |
违反约束 (如外键、唯一性约束) |
InternalError |
数据库内部错误和数据损坏 |
ProgrammingError |
SQL 语法错误和 API 使用不当 |
NotSupportedError |
不受支持的功能或操作 |
另请参阅
- Python Database API Specification v2.0
chdb.dbapi.connections- 数据库连接管理chdb.dbapi.cursors- 数据库游标操作
示例
>>> try:
... cursor.execute("SELECT * FROM nonexistent_table")
... except ProgrammingError as e:
... print(f"SQL Error: {e}")
...
SQL Error: Table 'nonexistent_table' doesn't exist>>> try:
... cursor.execute("INSERT INTO users (id) VALUES (1), (1)")
... except IntegrityError as e:
... print(f"Constraint violation: {e}")
...
Constraint violation: Duplicate entry '1' for key 'PRIMARY'异常 chdb.dbapi.err.DataError
因处理中的数据存在问题而引发的异常。
当数据库操作因正在处理的数据出现问题而失败时,会引发此异常,例如:
- 除零操作
- 数值超出范围
- 无效的日期/时间值
- 字符串截断错误
- 类型转换失败
- 不符合列类型的数据格式
引发
| Exception | Condition |
|---|---|
DataError |
当数据验证或处理失败时 |
示例
>>> # SQL 中的除零操作
>>> cursor.execute("SELECT 1/0")
DataError: Division by zero>>> # 无效的日期格式
>>> cursor.execute("INSERT INTO table VALUES ('invalid-date')")
DataError: Invalid date format异常 chdb.dbapi.err.DatabaseError
基类:Error
为与数据库相关的错误而引发的异常。
这是所有数据库相关错误的基类。它涵盖数据库操作期间发生的、 与数据库本身相关而非与接口相关的 所有错误。
常见场景包括:
- SQL 执行错误
- 数据库连接问题
- 与事务相关的问题
- 违反数据库特定约束
异常 chdb.dbapi.err.Error
作为所有其他错误异常 (Warning 除外) 的基类异常。
这是 chdb 中所有错误异常的基类,不包括警告。 它是所有会导致操作无法成功完成的数据库错误情况的父类。
另请参阅
Warning- 用于不会阻止操作完成的非致命警告
异常 chdb.dbapi.err.IntegrityError
当数据库的关系完整性遭到破坏时引发的异常。
当数据库操作违反完整性约束时,会引发此异常, 包括:
- 违反外键约束
- 违反主键或唯一约束 (键重复)
- 违反检查约束
- 违反 NOT NULL 约束
- 违反参照完整性
引发
| 异常 | 条件 |
|---|---|
IntegrityError |
当违反数据库完整性约束时 |
示例
>>> # 重复的主键
>>> cursor.execute("INSERT INTO users (id, name) VALUES (1, 'John')")
>>> cursor.execute("INSERT INTO users (id, name) VALUES (1, 'Jane')")
IntegrityError: Duplicate entry '1' for key 'PRIMARY'>>> # 外键违规
>>> cursor.execute("INSERT INTO orders (user_id) VALUES (999)")
IntegrityError: Cannot add or update a child row: foreign key constraint fails异常 chdb.dbapi.err.InterfaceError
基类:Error
当错误与数据库接口而不是数据库本身相关时,会引发此异常。
当数据库接口实现出现问题时,会引发此异常,例如:
- 无效的连接参数
- API 使用不当 (在已关闭的连接上调用方法)
- 接口层协议错误
- 模块导入或初始化失败
引发
| Exception | Condition |
|---|---|
InterfaceError |
当数据库接口遇到与数据库操作无关的错误时 |
异常 chdb.dbapi.err.InternalError
当数据库遇到内部错误时引发的异常。
当数据库系统遇到并非由应用程序导致的内部错误时,会引发此异常,例如:
- 游标状态无效 (游标已失效)
- 事务状态不一致 (事务不同步)
- 数据库损坏
- 内部数据结构损坏
- 系统级数据库错误
引发
| 异常 | 条件 |
|---|---|
InternalError |
当数据库出现内部不一致时 |
异常 chdb.dbapi.err.NotSupportedError
在某个方法或数据库 API 不受支持时引发的异常。
当应用程序尝试使用当前数据库配置或版本不支持的数据库功能或 API 方法时,会引发此异常,例如:
- 在不支持事务的连接上调用
rollback() - 使用当前数据库版本不支持的高级 SQL 功能
- 调用当前 driver 中未实现的方法
- 尝试使用已禁用的数据库功能
引发
| 异常 | 条件 |
|---|---|
NotSupportedError |
访问不受支持的数据库功能时 |
示例
>>> # 在非事务性连接上回滚事务
>>> connection.rollback()
NotSupportedError: Transactions are not supported>>> # 使用不支持的 SQL 语法
>>> cursor.execute("SELECT * FROM table WITH (NOLOCK)")
NotSupportedError: WITH clause not supported in this database version异常 chdb.dbapi.err.OperationalError
因与数据库操作相关的错误而引发的异常。
当数据库操作期间发生错误, 且这些错误不一定在程序员的控制范围内时,会引发此异常,包括:
- 与数据库的连接意外中断
- 找不到数据库服务器或无法连接
- 事务处理失败
- 处理过程中发生内存分配错误
- 磁盘空间不足或资源耗尽
- 数据库服务器内部错误
- 身份验证或授权失败
引发
| 异常 | 条件 |
|---|---|
OperationalError |
当数据库操作因运行层面的问题而失败时 |
异常 chdb.dbapi.err.ProgrammingError
在数据库操作中发生编程错误时引发的异常。
当应用程序使用数据库时存在编程错误,就会引发此异常,包括:
- 找不到表或列
- 创建时表或索引已存在
- 语句中的 SQL 语法错误
- 预处理语句中指定的参数个数不正确
- 无效的 SQL 操作 (例如,对不存在的对象执行 DROP)
- 数据库 API 方法使用不当
引发
| Exception | Condition |
|---|---|
ProgrammingError |
当 SQL 语句或 API 用法存在错误时 |
示例
>>> # 表未找到
>>> cursor.execute("SELECT * FROM nonexistent_table")
ProgrammingError: Table 'nonexistent_table' doesn't exist>>> # SQL 语法错误
>>> cursor.execute("SELCT * FROM users")
ProgrammingError: You have an error in your SQL syntax>>> # 参数数量错误
>>> cursor.execute("INSERT INTO users (name, age) VALUES (%s)", ('John',))
ProgrammingError: Column count doesn't match value count异常 chdb.dbapi.err.StandardError
基类:Exception
与 chdb 操作相关的异常。
这是所有 chdb 相关异常的基类。它继承自 Python 内置的 Exception 类,并作为数据库操作异常层级结构的根类。
异常 chdb.dbapi.err.Warning
在插入过程中发生数据截断等重要警告时引发的异常。
当数据库操作已完成,但伴有 需要提醒应用程序注意的重要警告时,就会引发此异常。 常见场景包括:
- 插入过程中的数据截断
- 数值转换中的精度丢失
- 字符集转换警告
模块常量
chdb.dbapi.apilevel = '2.0'
str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str根据给定对象创建一个新的字符串对象。如果指定了 encoding 或
errors,则该对象必须提供一个数据缓冲区,
该缓冲区将使用给定的编码和错误处理器进行解码。
否则,返回 object._\_str_\_() (如果已定义)
或 repr(object) 的结果。
- encoding 默认为‘utf-8’。
- errors 默认为‘strict’。
chdb.dbapi.threadsafety = 1
int([x]) -> integer
int(x, base=10) -> integer将数值或字符串转换为整数;如果未提供参数,则返回 0。 如果 x 是数值,则返回 x._int_()。对于浮点数, 会向零方向截断。
如果 x 不是数值,或者给定了 base,那么 x 必须是一个表示给定 进制整数字面量的字符串、bytes 或 bytearray 实例。 该字面量前面可以有 ‘+’ 或 ‘-’,并且两侧可以包含空白字符。 base 默认为 10。有效的进制为 0 和 2-36。 base 为 0 表示根据字符串中的整数字面量来解释其进制。
>>> int(‘0b100’, base=0)
4chdb.dbapi.paramstyle = 'format'
str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str根据给定对象创建一个新的字符串对象。如果指定了 encoding 或 errors,那么该对象必须暴露一个数据缓冲区, 该缓冲区将使用给定的编码和错误处理程序进行解码。 否则,返回 object._str_() 的结果 (如果已定义) 或 repr(object)。 encoding 默认为‘utf-8’。 errors 默认为‘strict’。
类型常量
chdb.dbapi.STRING = frozenset({247, 253, 254})
用于 DB-API 2.0 类型比较的扩展型 frozenset。
此类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它可实现更灵活的类型检查,既可以使用相等运算符,也可以使用不等运算符, 将单个项与该集合进行比较。
这类用法适用于 STRING、BINARY、NUMBER 等类型常量,以便进行 诸如“field_type == STRING”之类的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.BINARY = frozenset({249, 250, 251, 252})
为 DB-API 2.0 类型比较扩展的 frozenset。
此类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它支持灵活的类型检查,允许使用相等和不等运算符 将单个项与该集合进行比较。
这用于 STRING、BINARY、NUMBER 等类型常量,从而支持 诸如 “field_type == STRING” 这样的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.NUMBER = frozenset({0, 1, 3, 4, 5, 8, 9, 13})
用于 DB-API 2.0 类型比较的扩展型 frozenset。
此类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它支持更灵活的类型检查,使单个项既可以使用相等运算符, 也可以使用不等运算符与该集合进行比较。
这用于 STRING、BINARY、NUMBER 等类型常量,以支持 类似“field_type == STRING”这样的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.DATE = frozenset({10, 14})
用于 DB-API 2.0 类型比较的扩展型 frozenset。
该类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它支持更灵活的类型检查,即可使用相等和不等运算符将单个项 与该集合进行比较。
它用于 STRING、BINARY、NUMBER 等类型常量,以便支持 诸如 “field_type == STRING” 这样的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.TIME = frozenset({11})
用于 DB-API 2.0 类型比较的扩展版 frozenset。
此类在 frozenset 的基础上进行了扩展,以支持 DB-API 2.0 的类型比较语义。 它支持更灵活的类型检查,使单个项既可以通过相等运算符,也可以通过不等运算符 与该集合进行比较。
这可用于 STRING、BINARY、NUMBER 等类型常量,从而支持 诸如“field_type == STRING”这样的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.TIMESTAMP = frozenset({7, 12})
用于 DB-API 2.0 类型比较的扩展版 frozenset。
此类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它支持更灵活的类型检查,可以使用相等或不等运算符, 将单个项与该集合进行比较。
它用于 STRING、BINARY、NUMBER 等类型常量,以支持 “field_type == STRING”这类比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.DATETIME = frozenset({7, 12})
用于 DB-API 2.0 类型比较的扩展型 frozenset。
此类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它支持更灵活的类型检查,使单个项既可使用相等运算符,也可使用不等运算符 与该集合进行比较。
这类对象用于 STRING、BINARY、NUMBER 等类型常量,从而支持 类似“field_type == STRING”的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # 返回 True
>>> FIELD_TYPE.INT != string_types # 返回 True
>>> FIELD_TYPE.BLOB in string_types # 返回 Falsechdb.dbapi.ROWID = frozenset({})
为 DB-API 2.0 类型比较扩展的 frozenset。
此类扩展了 frozenset,以支持 DB-API 2.0 的类型比较语义。 它支持灵活的类型检查,使单个项既可使用相等运算符,也可使用不等运算符 与该集合进行比较。
这用于 STRING、BINARY、NUMBER 等类型常量,以支持 诸如“field_type == STRING”这样的比较,其中 field_type 是单个类型值。
示例
>>> string_types = DBAPISet([FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING])
>>> FIELD_TYPE.STRING == string_types # Returns True
>>> FIELD_TYPE.INT != string_types # Returns True
>>> FIELD_TYPE.BLOB in string_types # Returns False使用示例
基本查询示例:
import chdb.dbapi as dbapi
print("chdb driver version: {0}".format(dbapi.get_client_info()))
# Create connection and cursor
conn = dbapi.connect()
cur = conn.cursor()
# Execute query
cur.execute('SELECT version()')
print("description:", cur.description)
print("data:", cur.fetchone())
# Clean up
cur.close()
conn.close()使用数据:
import chdb.dbapi as dbapi
conn = dbapi.connect()
cur = conn.cursor()
# Create table
cur.execute("""
CREATE TABLE employees (
id UInt32,
name String,
department String,
salary Decimal(10,2)
) ENGINE = Memory
""")
# Insert data
cur.execute("""
INSERT INTO employees VALUES
(1, 'Alice', 'Engineering', 75000.00),
(2, 'Bob', 'Marketing', 65000.00),
(3, 'Charlie', 'Engineering', 80000.00)
""")
# Query data
cur.execute("SELECT * FROM employees WHERE department = 'Engineering'")
# Fetch results
print("Column names:", [desc[0] for desc in cur.description])
for row in cur.fetchall():
print(row)
conn.close()连接管理:
import chdb.dbapi as dbapi
# Temporary database (default)
conn1 = dbapi.connect()
# Persistent database file
conn2 = dbapi.connect("./my_database.chdb")
# Connection with parameters
conn3 = dbapi.connect("./my_database.chdb?log-level=debug&verbose")
# Read-only connection
conn4 = dbapi.connect("./my_database.chdb?mode=ro")
# Automatic connection cleanup
with dbapi.connect("test.chdb") as conn:
cur = conn.cursor()
cur.execute("SELECT count() FROM numbers(1000)")
result = cur.fetchone()
print(f"Count: {result[0]}")
cur.close()最佳实践
- 连接管理:使用完毕后务必关闭连接和游标
- 上下文管理器:使用
with语句自动清理资源 - 批次处理:对于较大的结果集,使用
fetchmany() - 错误处理:将数据库操作放在 try-except 块中
- 参数绑定:尽可能使用参数化查询
- 内存管理:对于非常大的数据集,避免使用
fetchall()
Python 用户自定义函数 (UDF)
chDB 支持原生 Python UDF,可在进程内运行,支持类型化参数、自动类型推断以及可配置的 NULL/异常处理。注册为 UDF 的 Python 函数可直接在 SQL 查询中调用。
以下示例使用默认的 CSV 输出格式。行内注释显示逻辑结果值;原始输出会将 NULL 显示为 \N,并对字符串和日期值进行 CSV 引号转义。
chdb.create_function
将 Python 函数注册为 chDB SQL 函数。
语法
chdb.create_function(name, func, arg_types=None, return_type=None, *, on_null=None, on_error=None)参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
name |
str | (必填) | 要注册的 SQL 函数名称 |
func |
callable | (必填) | 要注册的 Python 函数 |
arg_types |
list of ChdbType/str/type, or None | None |
参数类型列表。若为 None,则根据类型注解推断 |
return_type |
ChdbType/str/type, or None | None |
返回类型。若为 None,则根据函数的返回类型注解推断;若该注解也缺失,则注册失败 |
on_null |
str or NullHandling | None (skip) |
如何处理 NULL 输入:"skip" 或 "pass"。仅限关键字参数 |
on_error |
str or ExceptionHandling | None (propagate) |
如何处理异常:"propagate" 或 "ignore"。仅限关键字参数 |
每个类型参数 (arg_types 的元素和 return_type) 均可接受:
ChdbType常量:INT64、STRING、FLOAT64等。- ClickHouse 类型字符串:
"Int64"、"String"、"DateTime64(6)"、"DateTime('UTC')"等。 - Python 类型:
int、float、str、bool、bytes、datetime.date、datetime.datetime— 根据自动类型映射进行映射
注册已注册的名称会引发错误 — UDFs 不会被静默替换。请先调用 drop_function,再重新注册函数。
示例
from chdb import create_function, drop_function, query
from chdb.sqltypes import INT64, STRING
create_function("strlen", len, arg_types=[STRING], return_type=INT64)
print(query("SELECT strlen('hello')")) # 5
drop_function("strlen")chdb.drop_function
移除先前注册的 Python UDF。如果该函数尚未注册,则不会执行任何操作,因此可放心无条件调用。
语法
chdb.drop_function(name)参数
| 参数 | 类型 | 描述 |
|---|---|---|
name |
str | 要删除的 SQL 函数名称 |
@func 装饰器
用于将 Python 函数注册为 chDB SQL 函数。该函数仍可像普通 Python 函数一样调用,同时可通过其 __name__ 在 SQL 查询中使用。
语法
from chdb import func
@func(arg_types=None, return_type=None, *, on_null=None, on_error=None)
def my_function(...):
...参数
与 create_function 相同 (name 和 func 除外,它们由被装饰的函数派生) 。
示例
from chdb import func, query
from chdb.sqltypes import INT64, STRING
# Explicit types
@func([INT64, INT64], INT64)
def add(a, b):
return a + b
# Types inferred from annotations
@func()
def multiply(a: int, b: int) -> int:
return a * b
# Explicit return_type, arg_types inferred from annotations
@func(return_type=STRING)
def greet(name: str):
return f"Hello, {name}!"
print(query("SELECT add(12, 22)")) # 34
print(query("SELECT multiply(3, 7)")) # 21
print(query("SELECT greet('world')")) # Hello, world!类型系统
可用类型
从 chdb.sqltypes 中导入类型:
from chdb.sqltypes import (
BOOL,
INT8, INT16, INT32, INT64, INT128, INT256,
UINT8, UINT16, UINT32, UINT64, UINT128, UINT256,
FLOAT32, FLOAT64,
STRING,
DATE, DATE32, DATETIME, DATETIME64,
)自动类型映射
根据 Python 类型注解推断类型时,使用以下映射:
| Python 类型 | ClickHouse 类型 |
|---|---|
bool |
Bool |
int |
Int64 |
float |
Float64 |
str |
String |
bytes |
String |
bytearray |
String |
datetime.date |
Date |
datetime.datetime |
DateTime64(6) |
类型指定方式
可以通过多种方式指定类型:
from chdb import create_function, func
from chdb.sqltypes import INT64
# 1. ChdbType constants
create_function("f1", lambda x: x, arg_types=[INT64], return_type=INT64)
# 2. ClickHouse type strings
create_function("f2", lambda x: x, arg_types=["Int64"], return_type="Int64")
# 3. Parameterized type strings
create_function("f3", lambda x: x, arg_types=["DateTime('UTC')"], return_type="DateTime('UTC')")
# 4. Python types — passed directly or used as annotations
create_function("f4", lambda x: x, arg_types=[int], return_type=int)
@func()
def f5(x: int) -> int:
return xNULL 处理
使用 on_null 参数控制 NULL 值的处理方式。
| 值 | 枚举 | 行为 |
|---|---|---|
"skip" |
NullHandling.SKIP |
不调用函数,直接返回 NULL (默认) |
"pass" |
NullHandling.PASS |
将 NULL 转换为 None 后调用函数 |
from chdb import func, query, NullHandling
# Default: NULL in → NULL out, function not called
@func(return_type="Int64")
def add_one(x: int) -> int:
return x + 1
print(query("SELECT add_one(NULL)")) # NULL
# Pass NULL as None
@func(return_type="Int64", on_null="pass")
def null_safe(x):
return 0 if x is None else x + 1
print(query("SELECT null_safe(NULL)")) # 0异常处理
使用 on_error 参数控制异常处理方式。
| 值 | 枚举 | 行为 |
|---|---|---|
"propagate" |
ExceptionHandling.PROPAGATE |
将异常作为 SQL 错误抛出 (默认) |
"ignore" |
ExceptionHandling.IGNORE |
捕获异常并返回 NULL |
from chdb import func, query
# Default: exception propagates
@func(arg_types=["Int64", "Int64"], return_type="Int64")
def divide(a, b):
return a // b
# print(query("SELECT divide(1, 0)")) # Error: division by zero
# Ignore: exception → NULL
@func(arg_types=["Int64", "Int64"], return_type="Int64", on_error="ignore")
def safe_divide(a, b):
return a // b
print(query("SELECT safe_divide(1, 0)")) # NULL
print(query("SELECT safe_divide(10, 2)")) # 5日期时间和时区支持
UDF 完全支持具有时区信息的 Date、Date32、DateTime 和 DateTime64 类型。
from chdb import func, query
from datetime import datetime, timedelta, date
@func(arg_types=["DateTime('UTC')"], return_type="DateTime('UTC')")
def add_one_hour(dt):
return dt + timedelta(hours=1)
@func()
def get_year(d: date) -> int:
return d.year
print(query("SELECT add_one_hour(toDateTime('2024-01-01 12:00:00', 'UTC'))")) # 2024-01-01 13:00:00
print(query("SELECT get_year(toDate('2024-06-15'))")) # 2024- 输入的 ClickHouse
DateTime/DateTime64值会转换为包含时区信息的 Pythondatetime对象 - Python
datetime输出对象返回至 ClickHouse 时会保留时区信息 chdb.sqltypes中的DATETIME64类型默认标度为 6 (微秒) ,相当于DateTime64(6)
旧版 API
chdb.udf.chdb_udf
chDB Python UDF (用户自定义函数) 装饰器。
语法
chdb.udf.chdb_udf(return_type='String')参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
return_type |
str | "String" |
函数的返回类型。应为 ClickHouse 数据类型之一。 |
注意
- 函数应为无状态。仅支持 UDFs,不支持 UDAFs。
- 默认返回类型为 String。返回类型应为 ClickHouse 数据类型之一。
- 函数应接受 String 类型的参数。所有参数均为字符串。
- 函数会对输入的每一行调用一次。
- 函数应为纯 Python 函数。请导入函数中使用的所有模块。
- 所用的 Python 解释器与运行脚本时使用的解释器相同。
示例
@chdb_udf()
def sum_udf(lhs, rhs):
return int(lhs) + int(rhs)
@chdb_udf()
def func_use_json(arg):
import json
# ... use json modulechdb.udf.generate_udf
生成 UDF 配置文件和可执行脚本。
此函数会为 chDB 中的用户自定义函数 (UDF) 创建所需的文件:
- 用于处理输入数据的 Python 可执行脚本
- 用于在 ClickHouse 中注册 UDF 的 XML 配置文件
语法
chdb.udf.generate_udf(func_name, args, return_type, udf_body)参数
| 参数 | 类型 | 描述 |
|---|---|---|
func_name |
str | UDF 函数名称 |
args |
list | 函数的参数名称列表 |
return_type |
str | 函数的 ClickHouse 返回类型 |
udf_body |
str | UDF 函数的 Python 源代码主体 |
工具
适用于 chDB 的实用函数和辅助工具。
本模块包含各种用于 chDB 的实用函数,例如数据类型推断、数据转换辅助函数以及调试工具。
chdb.utils.convert_to_columnar
将字典列表转换为列式格式。
该函数接受一个字典列表,并将其转换为一个字典, 其中每个键对应一列,每个值则是该列的值列表。 字典中缺失的值会表示为 None。
语法
chdb.utils.convert_to_columnar(items: List[Dict[str, Any]]) → Dict[str, List[Any]]参数
| 参数 | 类型 | 描述 |
|---|---|---|
items |
List[Dict[str, Any]] |
要转换的字典列表 |
返回值
| 返回类型 | 描述 |
|---|---|
Dict[str, List[Any]] |
一个字典,键为列名,值为对应列值的列表 |
示例
>>> items = [
... {"name": "Alice", "age": 30, "city": "New York"},
... {"name": "Bob", "age": 25},
... {"name": "Charlie", "city": "San Francisco"}
... ]
>>> convert_to_columnar(items)
{
'name': ['Alice', 'Bob', 'Charlie'],
'age': [30, 25, None],
'city': ['New York', None, 'San Francisco']
}chdb.utils.flatten_dict
将嵌套字典扁平化。
此函数接收一个嵌套字典并将其扁平化,使用分隔符拼接嵌套键。 由字典组成的列表会被序列化为 JSON 字符串。
语法
chdb.utils.flatten_dict(d: Dict[str, Any], parent_key: str = '', sep: str = '_') → Dict[str, Any]参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
d |
Dict[str, Any] |
必填 | 要扁平化的字典 |
parent_key |
str | "" |
要添加到每个键前的基础键名 |
sep |
str | "_" |
用于连接拼接键的分隔符 |
返回值
| 返回类型 | 描述 |
|---|---|
Dict[str, Any] |
扁平化后的字典 |
示例
>>> nested_dict = {
... "a": 1,
... "b": {
... "c": 2,
... "d": {
... "e": 3
... }
... },
... "f": [4, 5, {"g": 6}],
... "h": [{"i": 7}, {"j": 8}]
... }
>>> flatten_dict(nested_dict)
{
'a': 1,
'b_c': 2,
'b_d_e': 3,
'f_0': 4,
'f_1': 5,
'f_2_g': 6,
'h': '[{"i": 7}, {"j": 8}]'
}chdb.utils.infer_data_type
为一组值推断最合适的数据类型。
此函数会检查一组值,并确定能够表示列表中所有值的最合适数据类型。它会考虑整数、无符号整数、Decimal 和浮点类型;如果这些值无法用任何数值类型表示,或者所有值均为 None,则默认使用“string”。
语法
chdb.utils.infer_data_type(values: List[Any]) → str参数
| 参数 | 类型 | 描述 |
|---|---|---|
values |
List[Any] |
要分析的值列表。这些值可以是任意类型 |
返回值
| 返回类型 | 描述 |
|---|---|
str |
表示推断出的数据类型的字符串。可能的返回值包括:“int8”、“int16”、“int32”、“int64”、“int128”、“int256”、“uint8”、“uint16”、“uint32”、“uint64”、“uint128”、“uint256”、“decimal128”、“decimal256”、“float32”、“float64” 或 “string”。 |
chdb.utils.infer_data_types
为列式数据结构中的每一列推断数据类型。
此函数会分析各列中的值,并根据数据样本为每一列推断出最合适的 数据类型。
语法
chdb.utils.infer_data_types`(column_data: Dict[str, List[Any]], n_rows: int = 10000) → List[tuple]参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
column_data |
Dict[str, List[Any]] |
必填 | 一个字典,其中键为列名,值为对应列的值列表 |
n_rows |
int | 10000 |
用于类型推断的采样行数 |
返回值
| 返回类型 | 描述 |
|---|---|
List[tuple] |
一个元组列表,其中每个元组包含列名及其推断出的数据类型 |
抽象基类
类 chdb.rwabc.PyReader(data: Any)`
基类:ABC
class chdb.rwabc.PyReader(data: Any)abstractmethod read
从给定列中读取指定数量的行,并返回一个对象列表, 其中每个对象对应一列的值序列。
abstractmethod (col_names: List[str], count: int) → List[Any]参数
| 参数 | 类型 | 说明 |
|---|---|---|
col_names |
List[str] |
要读取的列名列表 |
count |
int | 要读取的最大行数 |
返回值
| 返回类型 | 说明 |
|---|---|
List[Any] |
序列列表,每一列对应一个序列 |
类 chdb.rwabc.PyWriter
基类:ABC
class chdb.rwabc.PyWriter(col_names: List[str], types: List[type], data: Any)abstractmethod finalize
组装并返回各个块中的最终数据。必须由子类实现。
abstractmethod finalize() → bytes返回值
| 返回类型 | 说明 |
|---|---|
bytes |
最终的序列化数据 |
abstractmethod write
将数据列写入块中。必须由子类实现。
abstractmethod write(col_names: List[str], columns: List[List[Any]]) → None参数
| 参数 | 类型 | 描述 |
|---|---|---|
col_names |
List[str] |
正在写入的列名列表 |
columns |
List[List[Any]] |
列数据列表,其中每列都由一个列表表示 |
异常处理
类 chdb.ChdbError
基类:Exception
与 chDB 相关错误的基础异常类。
当 chDB 查询执行失败或发生错误时,会引发此异常。它继承自标准 Python 的 Exception 类,并提供来自底层 ClickHouse engine 的错误信息。
异常消息通常包含来自 ClickHouse 的详细错误信息,包括语法错误、类型不匹配、缺失的表/列,以及其他查询执行问题。
变量
| 变量 | 类型 | 描述 |
|---|---|---|
args |
- | 包含错误消息及其他附加参数的 Tuple |
示例
>>> try:
... result = chdb.query("SELECT * FROM non_existent_table")
... except chdb.ChdbError as e:
... print(f"Query failed: {e}")
Query failed: Table 'non_existent_table' doesn't exist>>> try:
... result = chdb.query("SELECT invalid_syntax FROM")
... except chdb.ChdbError as e:
... print(f"Syntax error: {e}")
Syntax error: Syntax error near 'FROM'版本信息
chdb.chdb_version = ('3', '6', '0')
内置的不可变序列。
如果未提供参数,构造函数会返回一个空元组。 如果指定了 iterable,则会用 iterable 中的项初始化该元组。
如果参数本身就是元组,返回值就是该对象本身。
chdb.engine_version = '25.5.2.1'
str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str根据给定对象创建一个新的字符串对象。如果指定了 encoding 或 errors,则该对象必须提供一个数据缓冲区, 并将使用给定的编码和错误处理器对其进行解码。 否则,返回 object._str_() 的结果 (如果已定义) , 或 repr(object)。
- encoding 默认为 ‘utf-8’。
- errors 默认为 ‘strict’。
chdb.__version__ = '3.6.0'
str(object=’’) -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str根据给定对象创建一个新的字符串对象。如果指定了 encoding 或 errors,则该对象必须提供一个数据缓冲区, 该缓冲区将使用给定的编码和错误处理程序进行解码。 否则,返回 object._str_() 的结果 (如果已定义) 或 repr(object)。
- encoding 默认为 ‘utf-8’。
- errors 默认为 ‘strict’。