ClickHouse Connect 驱动 API

import clickhouse_connect

client = clickhouse_connect.get_client()
print(client.server_version)
# 输出: '22.10.1.98'

import clickhouse_connect

client = clickhouse_connect.get_client(host='play.clickhouse.com', secure=True, port=443, user='play', password='clickhouse')
print(client.command('SELECT timezone()'))
# 输出: 'Etc/UTC'

import clickhouse_connect

client = clickhouse_connect.get_client(
    host='play.clickhouse.com',
    user='play',
    password='clickhouse',
    port=443,
    session_id='example_session_1',
    connect_timeout=15,
    database='github',
    settings={'distributed_ddl_task_timeout':300},
)
print(client.database)
# 输出: 'github'

import clickhouse_connect

# 启动时创建一次
client = clickhouse_connect.get_client(host='my-host', username='default', password='password')

# 所有查询复用同一客户端
for i in range(1000):
    result = client.query('SELECT count() FROM users')

# 关闭时销毁
client.close()

# 错误做法：创建 1000 个客户端，产生高昂的初始化开销
for i in range(1000):
    client = clickhouse_connect.get_client(host='my-host', username='default', password='password')
    result = client.query('SELECT count() FROM users')
    client.close()

import clickhouse_connect
import threading

# 选项 1：禁用会话（推荐用于共享客户端）
client = clickhouse_connect.get_client(
    host='my-host',
    username='default',
    password='password',
    autogenerate_session_id=False  # 线程安全必需
)

def worker(thread_id):
    # 所有线程现在均可安全共用同一客户端
    result = client.query(f"SELECT {thread_id}")
    print(f"Thread {thread_id}: {result.result_rows[0][0]}")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

client.close()
# 输出：
# Thread 0: 0
# Thread 7: 7
# Thread 1: 1
# Thread 9: 9
# Thread 4: 4
# Thread 2: 2
# Thread 8: 8
# Thread 5: 5
# Thread 6: 6
# Thread 3: 3

def worker(thread_id):
    # 每个线程拥有独立的客户端和会话
    client = clickhouse_connect.get_client(host='my-host', username='default', password='password')
    client.command('CREATE TEMPORARY TABLE temp (id UInt32) ENGINE = Memory')
    # ... 使用临时表 ...
    client.close()

client = clickhouse_connect.get_client(host='my-host', username='default', password='password')
try:
    result = client.query('SELECT 1')
finally:
    client.close()

with clickhouse_connect.get_client(host='my-host', username='default', password='password') as client:
    result = client.query('SELECT 1')

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {'table': 'my_table', 'v1': my_date, 'v2': "a string with a single quote'"}
client.query('SELECT * FROM {table:Identifier} WHERE date >= {v1:DateTime} AND string ILIKE {v2:String}', parameters=parameters)

SELECT *
FROM my_table
WHERE date >= '2022-10-01 15:20:05'
  AND string ILIKE 'a string with a single quote\''

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {'v1': my_date, 'v2': "a string with a single quote'"}
client.query('SELECT * FROM my_table WHERE date >= %(v1)s AND string ILIKE %(v2)s', parameters=parameters)

SELECT *
FROM my_table
WHERE date >= '2022-10-01 15:20:05'
  AND string ILIKE 'a string with a single quote\''

import ipaddress

parameters = (35200.44, ipaddress.IPv4Address(0x443d04fe))
client.query('SELECT * FROM some_table WHERE metric >= %s AND ip_address = %s', parameters=parameters)

SELECT *
FROM some_table
WHERE metric >= 35200.44
  AND ip_address = '68.61.4.254''

settings = {'merge_tree_min_rows_for_concurrent_read': 65535,
            'session_id': 'session_1234',
            'use_skip_indexes': False}
client.query("SELECT event_type, sum(timeout) FROM event_errors WHERE event_time > '2022-08-01'", settings=settings)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 创建表
result = client.command("CREATE TABLE test_command (col_1 String, col_2 DateTime) ENGINE MergeTree ORDER BY tuple()")
print(result)  # 返回带有 query_id 的 QuerySummary

# 显示表定义
result = client.command("SHOW CREATE TABLE test_command")
print(result)
# 输出：
# CREATE TABLE default.test_command
# (
#     `col_1` String,
#     `col_2` DateTime
# )
# ENGINE = MergeTree
# ORDER BY tuple()

# 删除表
client.command("DROP TABLE test_command")

import clickhouse_connect

client = clickhouse_connect.get_client()

# 单值结果
count = client.command("SELECT count() FROM system.tables")
print(count)
# 输出：151

# 服务器版本
version = client.command("SELECT version()")
print(version)
# 输出："25.8.2.29"

import clickhouse_connect

client = clickhouse_connect.get_client()

# 使用客户端参数
table_name = "system"
result = client.command(
    "SELECT count() FROM system.tables WHERE database = %(db)s",
    parameters={"db": table_name}
)

# 使用服务端参数
result = client.command(
    "SELECT count() FROM system.tables WHERE database = {db:String}",
    parameters={"db": "system"}
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 使用特定设置执行命令
result = client.command(
    "OPTIMIZE TABLE large_table FINAL",
    settings={"optimize_throw_if_noop": 1}
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 简单的 SELECT 查询
result = client.query("SELECT name, database FROM system.tables LIMIT 3")

# 以行的形式访问结果
for row in result.result_rows:
    print(row)
# 输出：
# ('CHARACTER_SETS', 'INFORMATION_SCHEMA')
# ('COLLATIONS', 'INFORMATION_SCHEMA')
# ('COLUMNS', 'INFORMATION_SCHEMA')

# 访问列名和类型
print(result.column_names)
# 输出：("name", "database")
print([col_type.name for col_type in result.column_types])
# 输出：['String', 'String']

import clickhouse_connect

client = clickhouse_connect.get_client()

result = client.query("SELECT number, toString(number) AS str FROM system.numbers LIMIT 3")

# 按行访问（默认）
print(result.result_rows)
# Output: [[0, "0"], [1, "1"], [2, "2"]]

# 按列访问
print(result.result_columns)
# Output: [[0, 1, 2], ["0", "1", "2"]]

# 命名结果（字典列表）
for row_dict in result.named_results():
    print(row_dict)
# Output: 
# {"number": 0, "str": "0"}
# {"number": 1, "str": "1"}
# {"number": 2, "str": "2"}

# 第一行（字典形式）
print(result.first_item)
# Output: {"number": 0, "str": "0"}

# 第一行（元组形式）
print(result.first_row)
# Output: (0, "0")

import clickhouse_connect

client = clickhouse_connect.get_client()

# 使用字典参数（printf 风格）
query = "SELECT * FROM system.tables WHERE database = %(db)s AND name LIKE %(pattern)s"
parameters = {"db": "system", "pattern": "%query%"}
result = client.query(query, parameters=parameters)

# 使用 Tuple 参数
query = "SELECT * FROM system.tables WHERE database = %s LIMIT %s"
parameters = ("system", 5)
result = client.query(query, parameters=parameters)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 服务器端绑定（更安全，SELECT 查询性能更佳）
query = "SELECT * FROM system.tables WHERE database = {db:String} AND name = {tbl:String}"
parameters = {"db": "system", "tbl": "query_log"}

result = client.query(query, parameters=parameters)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 在查询中传递 ClickHouse 设置
result = client.query(
    "SELECT sum(number) FROM numbers(1000000)",
    settings={
        "max_block_size": 100000,
        "max_execution_time": 30
    }
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 行式数据：每个内部列表代表一行
data = [
    [1, "Alice", 25],
    [2, "Bob", 30],
    [3, "Joe", 28],
]

client.insert("users", data, column_names=["id", "name", "age"])

import clickhouse_connect

client = clickhouse_connect.get_client()

# 列式数据：每个内部列表对应一列
data = [
    [1, 2, 3],  # id 列
    ["Alice", "Bob", "Joe"],  # name 列
    [25, 30, 28],  # age 列
]

client.insert("users", data, column_names=["id", "name", "age"], column_oriented=True)

import clickhouse_connect

client = clickhouse_connect.get_client()

# 当您希望避免向服务器发送 DESCRIBE 查询时很有用
data = [
    [1, "Alice", 25],
    [2, "Bob", 30],
    [3, "Joe", 28],
]

client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    column_type_names=["UInt32", "String", "UInt8"],
)

import clickhouse_connect

client = clickhouse_connect.get_client()

data = [
    [1, "Alice", 25],
    [2, "Bob", 30],
]

# 将数据插入指定数据库中的表
client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    database="production",
)