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)  # QuerySummary と query_id を返す

# テーブル定義を表示する
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)
# Output:
# ('CHARACTER_SETS', 'INFORMATION_SCHEMA')
# ('COLLATIONS', 'INFORMATION_SCHEMA')
# ('COLUMNS', 'INFORMATION_SCHEMA')

# カラム名と型を取得
print(result.column_names)
# Output: ("name", "database")
print([col_type.name for col_type in result.column_types])
# Output: ['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()

# Dictionaryパラメータを使用する場合（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()

# 行指向データ: 各内部リストが1行を表す
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()

# カラム指向データ: 各内部リストが1つのカラム
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",
)