API драйвера ClickHouse Connect

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

# Использование параметров словаря (в стиле 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)

# Использование параметров кортежа
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",
)