chDB Python API 参考文档

chdb.query(sql, output_format='CSV', path='', udf_path='')

>>> # 基础 CSV 查询
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello"

>>> # 输出 DataFrame 格式的查询
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
   id    msg
0   1  hello

>>> # 使用基于文件的数据库进行查询
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")

>>> # 使用 UDF 进行查询
>>> result = chdb.query("SELECT my_udf('test')", udf_path="/path/to/udfs")

chdb.sql(sql, output_format='CSV', path='', udf_path='')

>>> # 基础 CSV 查询
>>> result = chdb.query("SELECT 1, 'hello'")
>>> print(result)
"1,hello"

>>> # 以 DataFrame 格式输出的查询
>>> df = chdb.query("SELECT 1 as id, 'hello' as msg", "DataFrame")
>>> print(df)
   id    msg
0   1  hello

>>> # 使用基于文件的数据库进行查询
>>> result = chdb.query("CREATE TABLE test (id INT) ENGINE = Memory", path="mydb.chdb")

>>> # 使用 UDF 进行查询
>>> result = chdb.query("SELECT my_udf('test')", udf_path="/path/to/udfs")

chdb.to_arrowTable(res)

>>> result = chdb.query("SELECT 1 as id, 'hello' as msg", "Arrow")
>>> table = chdb.to_arrowTable(result)
>>> print(table.to_pandas())
   id    msg
0   1  hello

chdb.to_df(r)

>>> 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(connection_string: str = ':memory:') → Connection

>>> # 内存数据库
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # 基于文件的数据库
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # 带参数
>>> conn = connect("data.db?mode=ro")  # 只读模式
>>> conn = connect(":memory:?verbose&log-level=debug")  # 调试日志
>>>
>>> # 使用上下文管理器自动清理资源
>>> with connect("data.db") as conn:
...     result = conn.query("SELECT 1")
...     print(result)
>>> # 连接已自动关闭

class chdb.session.Session(path=None)

cleanup()

>>> session = Session("test.db")
>>> try:
...     session.query("INVALID SQL")
... finally:
...     session.cleanup()  # 无论是否发生错误，均可安全清理

close()

>>> session = Session("test.db")
>>> session.query("SELECT 1")
>>> session.close()  # 显式关闭会话

query(sql, fmt='CSV', udf_path='')

>>> session = Session("test.db")
>>>
>>> # 使用默认 CSV 格式的基础查询
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1

>>> # 使用 JSON 格式进行查询
>>> 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, fmt='CSV') → StreamingResult

>>> 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}")

sql(sql, fmt='CSV', udf_path='')

>>> session = Session("test.db")
>>>
>>> # 使用默认 CSV 格式的基础查询
>>> result = session.query("SELECT 1 as number")
>>> print(result)
number
1

>>> # 使用 JSON 格式查询
>>> 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

chdb.state.connect(connection_string: str = ':memory:') → Connection

>>> # 内存数据库
>>> conn = connect()
>>> conn = connect(":memory:")
>>>
>>> # 基于文件的数据库
>>> conn = connect("my_data.db")
>>> conn = connect("/path/to/data.db")
>>>
>>> # 带参数
>>> conn = connect("data.db?mode=ro")  # 只读模式
>>> conn = connect(":memory:?verbose&log-level=debug")  # 调试日志
>>>
>>> # 使用上下文管理器自动清理资源
>>> with connect("data.db") as conn:
...     result = conn.query("SELECT 1")
...     print(result)
>>> # 连接自动关闭

class chdb.state.sqlitelike.Connection(connection_string: str)

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

>>> 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'),)

query(query: str, format: str = 'CSV') → Any

>>> 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(query: str, format: str = 'CSV') → StreamingResult

>>> 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}")

class chdb.state.sqlitelike.StreamingResult

fetch() → Any

>>> stream = conn.send_query("SELECT * FROM large_table")
>>> chunk = stream.fetch()
>>> while chunk is not None:
...     process_data(chunk)
...     chunk = stream.fetch()

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

record_batch(rows_per_batch: int = 1000000) → pa.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)

>>> stream = conn.send_query("SELECT number FROM numbers(1000000)")
>>> for 数据块 in stream:
...     print(f"Chunk size: {len(chunk)} bytes")

>>> with conn.send_query("SELECT * FROM data") as stream:
...     for chunk in stream:
...         process(chunk)
>>> # Stream 已自动关闭

class chdb.state.sqlitelike.Cursor(connection)

close() → None

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchone()
>>> cursor.close()  # 清理游标资源

column_names() → list

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT id, name, email FROM users LIMIT 1")
>>> print(cursor.column_names())
['id', 'name', 'email']

column_types() → list

>>> cursor = conn.cursor()
>>> cursor.execute("SELECT toInt32(1), toString('hello')")
>>> print(cursor.column_types())
['Int32', 'String']

commit() → None

>>> cursor = conn.cursor()
>>> cursor.execute("INSERT INTO test VALUES (1, 'data')")
>>> cursor.commit()

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

execute(query: str) → None

>>> 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'),)

fetchall() → 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}")

fetchmany(size: int = 1) → 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() → 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()

chdb.state.sqlitelike.to_arrowTable(res)

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

chdb.state.sqlitelike.to_df(r)

>>> 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: object

class chdb.dataframe.Table(*args: Any, **kwargs: Any)

chdb.dbapi.connect(*args, **kwargs)

chdb.dbapi.get_client_info()

chdb.dbapi.Binary(x)

class chdb.dbapi.connections.Connection(path=None)

>>> # 内存数据库
>>> conn = Connection()
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT 1")
>>> result = cursor.fetchall()
>>> conn.close()

>>> # 基于文件的数据库
>>> 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()

>>> # 上下文管理器用法
>>> with Connection() as cur:
...     cur.execute("SELECT version()")
...     version = cur.fetchone()

close()

commit()

cursor(cursor=None)

>>> conn = Connection()
>>> cur = conn.cursor()
>>> cur.execute("SELECT 1")
>>> result = cur.fetchone()

escape(obj, mapping=None)

>>> conn = Connection()
>>> safe_value = conn.escape("O'Reilly")
>>> query = f"SELECT * FROM users WHERE name = {safe_value}"

escape_string(s)

query(sql, fmt='CSV')

>>> conn = Connection()
>>> result = conn.query("SELECT 1, 'hello'", "CSV")
>>> print(result)
"1,hello\n"

rollback()

class chdb.dbapi.cursors.Cursor(connection)

>>> conn = Connection()
>>> cur = conn.cursor()
>>> cur.execute("SELECT 1 as id, 'test' as name")
>>> result = cur.fetchone()
>>> print(result)  # (1, 'test')
>>> cur.close()

callproc(procname, args=())

close()

execute(query, args=None)

>>> # 问号参数
>>> 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")

executemany(query, args)

>>> # 使用问号参数进行批量插入
>>> 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()

>>> cursor.execute("SELECT id, name FROM users")
>>> all_rows = cursor.fetchall()
>>> print(len(all_rows))  # 总行数

fetchmany(size=1)

>>> cursor.execute("SELECT id, name FROM users")
>>> rows = cursor.fetchmany(3)
>>> print(rows)  # [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]

fetchone()

>>> cursor.execute("SELECT id, name FROM users LIMIT 3")
>>> row = cursor.fetchone()
>>> print(row)  # (1, 'Alice')
>>> row = cursor.fetchone()
>>> print(row)  # (2, 'Bob')

mogrify(query, args=None)

>>> cur.mogrify("SELECT * FROM users WHERE id = ?", (123,))
"SELECT * FROM users WHERE id = 123"

nextset()

setinputsizes(*args)

setoutputsizes(*args)

StandardError
├── Warning
└── Error
    ├── InterfaceError
    └── DatabaseError
        ├── DataError
        ├── OperationalError
        ├── IntegrityError
        ├── InternalError
        ├── ProgrammingError
        └── NotSupportedError

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

>>> # SQL 中的除零操作
>>> cursor.execute("SELECT 1/0")
DataError: Division by zero

>>> # 无效的日期格式
>>> cursor.execute("INSERT INTO table VALUES ('invalid-date')")
DataError: Invalid date format

>>> # 重复的主键
>>> 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

>>> # 在非事务性连接上回滚事务
>>> connection.rollback()
NotSupportedError: Transactions are not supported

>>> # 使用不支持的 SQL 语法
>>> cursor.execute("SELECT * FROM table WITH (NOLOCK)")
NotSupportedError: WITH clause not supported in this database version

>>> # 表未找到
>>> 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