API драйвера ClickHouse Connect

Используйте clickhouse_connect.get_client, чтобы создать синхронный Client, или установите дополнительный пакет async и вызовите clickhouse_connect.get_async_client с await, чтобы создать нативный AsyncClient.

Асинхронная фабрика также принимает connector_limit=100, connector_limit_per_host=20 и keepalive_timeout=30.0 для настройки пула соединений aiohttp. Она не принимает pool_mgr. Синхронное backend-соединение chDB принимает path и chdb_options; см. Встроенное backend-соединение chDB.

Наконец, аргумент settings для get_client используется для передачи серверу дополнительных настроек ClickHouse с каждым клиентским запросом. Обратите внимание, что в большинстве случаев пользователи с доступом readonly=1 не могут изменять настройки, передаваемые вместе с запросом, поэтому ClickHouse Connect отбрасывает такие настройки в итоговом запросе и записывает предупреждение в журнал. Следующие настройки применяются только к HTTP-запросам/сеансам, используемым ClickHouse Connect, и не документированы как общие настройки ClickHouse.

О других настройках ClickHouse, которые можно передавать с каждым запросом, см. в документации ClickHouse.

• Без параметров клиент ClickHouse Connect подключится к HTTP-порту по умолчанию на localhost с пользователем по умолчанию default и без пароля:

import clickhouse_connect

client = clickhouse_connect.get_client()
print(client.server_version)

• Подключение к защищённому (HTTPS) внешнему серверу ClickHouse

import clickhouse_connect

client = clickhouse_connect.get_client(
    host="play.clickhouse.com",
    secure=True,
    port=443,
    username="play",
    password="clickhouse",
)
print(client.command("SELECT timezone()"))

• Подключение с идентификатором сеанса, а также с другими пользовательскими параметрами подключения и настройками ClickHouse.

import clickhouse_connect

client = clickhouse_connect.get_client(
    host="play.clickhouse.com",
    username="play",
    password="clickhouse",
    port=443,
    secure=True,
    session_id="example_session_1",
    connect_timeout=15,
    database="github",
    settings={"distributed_ddl_task_timeout": 300},
)
print(client.database)
# Output: github

Установите clickhouse-connect[chdb], чтобы использовать экспериментальное backend-соединение chDB, работающее в процессе приложения. Оно предоставляет синхронные методы клиента: запросы, вставка, стриминг и методы Arrow:

import clickhouse_connect

with clickhouse_connect.get_client(interface="chdb") as client:
    result = client.query("SELECT sum(number) FROM numbers(10)")
    print(result.first_row)
    # Output: (45,)

По умолчанию используется база данных в оперативной памяти. Передайте path="/data/my_chdb" или используйте dsn="chdb:///data/my_chdb" для постоянного хранения данных. Это backend-соединение поддерживает только один путь к движку на процесс и не поддерживает get_async_client или внешние данные.

Создание клиента ClickHouse Connect — ресурсоемкая операция, включающая установление соединения, получение метаданных сервера и инициализацию настроек. Следуйте этим рекомендациям для оптимальной производительности:

• Повторно используйте клиенты: Создавайте клиенты один раз при запуске приложения и используйте их повторно в течение всего жизненного цикла приложения
• Избегайте частого создания: Не создавайте новый клиент для каждого запроса или обращения
• Корректно освобождайте ресурсы: Всегда закрывайте клиенты при завершении работы, чтобы освободить ресурсы пула соединений
• По возможности используйте совместно: Один клиент может обрабатывать множество параллельных запросов через свой пул соединений (см. примечания о потоках ниже)

Повторно используйте один экземпляр клиента:

import clickhouse_connect

# Create once at startup
client = clickhouse_connect.get_client(
    host="my-host",
    username="default",
    password="password",
)

# Reuse for all queries
for i in range(1000):
    result = client.query("SELECT count() FROM users")

# Close on shutdown
client.close()

Избегайте повторного создания клиентов:

# BAD: Creates 1000 clients with expensive initialization overhead
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

# Option 1: Disable sessions (recommended for shared clients)
client = clickhouse_connect.get_client(
    host="my-host",
    username="default",
    password="password",
    autogenerate_session_id=False,
)

def worker(thread_id):
    # All threads can now safely use the same client
    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()

Альтернатива при использовании сеансов: Если вам нужны сеансы (например, для временных таблиц), создайте отдельный клиент для каждого потока:

def worker(thread_id):
    # Each thread gets its own client with isolated session
    client = clickhouse_connect.get_client(
        host="my-host",
        username="default",
        password="password",
    )
    client.command("CREATE TEMPORARY TABLE temp (id UInt32) ENGINE = Memory")
    # ... use temp table ...
    client.close()

Всегда закрывайте клиенты при завершении работы. Обратите внимание: client.close() освобождает клиент и закрывает HTTP-соединения из пула только в том случае, если клиент управляет собственным менеджером пула (например, если он создан с пользовательскими параметрами TLS/прокси). Для общего пула по умолчанию используйте client.close_connections(), чтобы принудительно очистить сокеты; в противном случае соединения будут автоматически освобождены по истечении периода бездействия и при завершении процесса.

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

Несколько клиентов уместны в следующих случаях:

• Разные серверы: один клиент на каждый сервер ClickHouse или кластер
• Разные учетные данные: отдельные клиенты для разных пользователей или уровней доступа
• Разные базы данных: когда нужно работать с несколькими базами данных
• Изолированные сеансы: когда нужны отдельные сеансы для временных таблиц или настроек, специфичных для сеанса
• Изоляция на уровне потоков: когда потокам нужны независимые сеансы (как показано выше)

В некоторых методах клиента используются один или оба стандартных именованных аргумента: parameters и settings. Они описаны ниже.

Методы query* и command клиента ClickHouse Connect принимают необязательный именованный аргумент parameters, который используется для привязки выражений Python к выражению значения в ClickHouse. Доступны два типа привязки.

ClickHouse поддерживает привязку на стороне сервера для значений в запросе. Привязанное значение передаётся отдельно от запроса в виде HTTP-параметра. ClickHouse Connect использует этот режим, когда обнаруживает выражение в формате {<name>:<datatype>}. Передавайте значения в виде словаря Python.

Имена параметров должны быть ASCII-именами ClickHouse BareWord. Драйвер принимает $ в начале, внутри или в конце имени, если сервер допускает его использование, например {$tenant_id:String}. Ключ словаря, который начинается и заканчивается символом $ и содержит буферное значение, например bytes, bytearray или memoryview, зарезервирован для соглашения ClickHouse Connect об использовании необработанных двоичных параметров. Если такой ключ используется для небинарного параметра с привязкой на стороне сервера, используйте для него только один заполнитель {name:Type}. Повторяющиеся имена $tag$ могут быть разобраны ClickHouse как маркеры heredoc.

Используйте Python None для допускающих NULL значений. Вложенные значения None поддерживаются внутри параметров Array и Tuple, а также внутри литералов Map, когда dict_parameter_format имеет значение "map".

• Привязка на стороне сервера со словарём Python, значением DateTime и строковым значением

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

ClickHouse Connect также поддерживает привязку параметров на стороне клиента, что дает больше гибкости при формировании шаблонизированных SQL-запросов. Для привязки на стороне клиента аргумент parameters должен быть словарём или последовательностью. При привязке на стороне клиента для подстановки параметров используется форматирование строк Python в стиле "printf".

Обратите внимание: в отличие от привязки на стороне сервера, привязка на стороне клиента не работает с идентификаторами баз данных, таблиц и столбцов, поскольку форматирование в стиле Python не различает разные типы строк, а для них требуется разное оформление (обратные кавычки или двойные кавычки для идентификаторов базы данных и одинарные кавычки для значений данных).

• Пример с Python-словарём, значением DateTime и экранированием строк

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

• Пример с последовательностью Python Sequence (Tuple), Float64 и IPv4Address

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'

Все основные методы клиента ClickHouse Connect — "insert" и "select" — принимают необязательный именованный аргумент settings, который позволяет передавать пользовательские настройки сервера ClickHouse для данного SQL-оператора. Аргумент settings должен быть словарём. Каждый элемент должен содержать имя настройки ClickHouse и соответствующее ей значение. Обратите внимание, что при отправке на сервер в качестве параметров запроса значения будут преобразованы в строки.

Как и в случае с настройками на уровне клиента, ClickHouse Connect отбрасывает любые настройки, которые сервер помечает как readonly=1, с соответствующим сообщением в журнале. Настройки, применимые только к запросам через HTTP-интерфейс ClickHouse, всегда допустимы. Эти настройки описаны в API get_client.

Пример использования настроек ClickHouse:

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

Используйте Client.command для операторов, которые не возвращают табличный набор данных, или для запросов, которые возвращают одно примитивное значение или одну строку. В зависимости от ответа метод возвращает строку, целое число, последовательность строк или QuerySummary. Если чтение даёт пустой результирующий набор, возвращается пустая строка.

import clickhouse_connect

client = clickhouse_connect.get_client()

# Create a table. A successful DDL returns QuerySummary.
summary = client.command(
    "CREATE TABLE test_command "
    "(col_1 String, col_2 DateTime) "
    "ENGINE MergeTree ORDER BY tuple()"
)
print(summary.query_id())

# Show table definition
result = client.command("SHOW CREATE TABLE test_command")
print(result)
# Output:
# CREATE TABLE default.test_command
# (
#     `col_1` String,
#     `col_2` DateTime
# )
# ENGINE = MergeTree
# ORDER BY tuple()

# Drop table
client.command("DROP TABLE test_command")

import clickhouse_connect

client = clickhouse_connect.get_client()

# Single value result
count = client.command("SELECT count() FROM system.tables")
print(count)

# Server version
version = client.command("SELECT version()")
print(version)

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

Client.query получает табличный набор данных в Native format ClickHouse и возвращает QueryResult. Полный результат материализуется при обращении к его свойству. Для результатов, которые не следует хранить в памяти, используйте стриминговый метод.

import clickhouse_connect

client = clickhouse_connect.get_client()

# Simple SELECT query
result = client.query(
    "SELECT number, toString(number) AS label FROM numbers(3)"
)

# Access results as rows
for row in result.result_rows:
    print(row)
# Output:
# (0, '0')
# (1, '1')
# (2, '2')

# Access column names and types
print(result.column_names)
# Output: ('number', 'label')
print([col_type.name for col_type in result.column_types])
# Output: ['UInt64', 'String']

import clickhouse_connect

client = clickhouse_connect.get_client()

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

# Row-oriented access (default)
print(result.result_rows)
# Output: [(0, '0'), (1, '1'), (2, '2')]

# Column-oriented access
print(result.result_columns)
# Output: [[0, 1, 2], ['0', '1', '2']]

# Named results (list of dictionaries)
for row_dict in result.named_results():
    print(row_dict)
# Output:
# {'number': 0, 'str': '0'}
# {'number': 1, 'str': '1'}
# {'number': 2, 'str': '2'}

# First row as dictionary
print(result.first_item)
# Output: {'number': 0, 'str': '0'}

# First row as tuple
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
    }
)

Базовый метод query возвращает объект QueryResult со следующими публичными свойствами:

• result_rows – Матрица результатов, представленная в виде строк.
• result_columns – Матрица результатов, представленная в виде столбцов.
• result_set – result_rows или result_columns в зависимости от ориентации запроса.
• column_names – Кортеж имен столбцов результата.
• column_types – Кортеж объектов ClickHouseType.
• row_count – Количество материализованных строк результата.
• query_id – Query id, указанный или сгенерированный для этого запроса. Пустая строка означает, что он недоступен.
• summary – Словарь, декодированный из заголовка ответа X-ClickHouse-Summary.
• first_item – Первая строка в виде словаря или None для пустого результата.
• first_row – Первая строка в виде последовательности или None для пустого результата.
• column_block_stream, row_block_stream и rows_stream – Внутренние контексты потоков. Вместо них используйте соответствующие стриминговые методы клиента.

См. Стриминговые запросы, чтобы узнать о поддерживаемых API StreamContext.

ClickHouse Connect предоставляет специализированные методы запросов для форматов данных NumPy, Pandas и Arrow. Подробную информацию об использовании этих методов, включая примеры, возможности стриминга и расширенную обработку типов, см. в разделе Расширенное выполнение запросов (запросы NumPy, Pandas и Arrow).

Для потоковой передачи больших результирующих наборов ClickHouse Connect предоставляет несколько методов. Подробности и примеры см. в разделе Расширенные запросы (потоковые запросы).

Для типичного сценария вставки нескольких записей в ClickHouse предусмотрен метод Client.insert. Он принимает следующие параметры:

Этот метод возвращает QuerySummary. Его словарь summary содержит значения, сообщаемые сервером. written_rows — это удобное свойство, а written_bytes() и query_id() возвращают соответствующие значения. При ошибке вставки будет вызвано исключение.

Описание специализированных методов вставки, работающих с Pandas DataFrames, PyArrow Tables и DataFrames на базе Arrow, см. в разделе Расширенная вставка (Специализированные методы вставки).

В примерах ниже предполагается наличие таблицы users со схемой (id UInt32, name String, age UInt8).

import clickhouse_connect

client = clickhouse_connect.get_client()

# Row-oriented data: each inner list is a row
data = [
    [13, "user_1", 25],
    [79, "user_2", 30],
]

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

import clickhouse_connect

client = clickhouse_connect.get_client()

# Column-oriented data: each inner list is a column
data = [
    [13, 79],  # id column
    ["user_1", "user_2"],  # name column
    [25, 30],  # age column
]

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

import clickhouse_connect

client = clickhouse_connect.get_client()

# Useful when you want to avoid a DESCRIBE query to the server
data = [
    [13, "user_1", 25],
    [79, "user_2", 30],
]

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

import clickhouse_connect

client = clickhouse_connect.get_client()

data = [
    [13, "user_1", 25],
    [79, "user_2", 30],
]

# Insert into a table in a specific database
client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    database="production",
)

Чтобы напрямую вставлять данные из файлов в таблицы ClickHouse, см. Расширенная вставка (вставка из файлов).

Для продвинутых сценариев, требующих прямого доступа к HTTP-интерфейсам ClickHouse без преобразования типов, см. Расширенное использование (Raw API).

Модуль clickhouse_connect.dbapi реализует интерфейсы connection и cursor, определённые в PEP 249. Он объявляет уровень API 2.0, threadsafety=2 и paramstyle="pyformat". Модуль также предоставляет конструкторы типов PEP 249 Date, Time, Timestamp и Binary, а также функции DateFromTicks, TimeFromTicks и TimestampFromTicks.

from clickhouse_connect import dbapi

connection = dbapi.connect(
    host="localhost",
    username="default",
    password="password",
    database="default",
)
cursor = connection.cursor()

try:
    cursor.execute(
        "SELECT name FROM system.tables "
        "WHERE database = %(database)s ORDER BY name LIMIT 5",
        {"database": "system"},
    )
    print(cursor.description)
    print(cursor.fetchall())
finally:
    cursor.close()
    connection.close()

Cursor.execute и Cursor.executemany принимают дополнительные именованные аргументы settings и query_formats. settings передаёт настройки ClickHouse. query_formats применяет форматы чтения по типам ClickHouse, когда оператор возвращает строки, используя то же сопоставление, что и Client.query. Cursor.execute также принимает доступный только по имени аргумент pyformat_encoded. Его значение по умолчанию True соответствует контракту DB-API pyformat. Диалект SQLAlchemy устанавливает его в False, когда компилятор оператора генерирует необработанные знаки процента, поэтому приложениям обычно не следует задавать его. executemany использует реализованный в драйвере механизм массовой вставки Native для совместимых операторов INSERT ... VALUES с материализованной последовательностью строк. fetchone, fetchmany и fetchall считывают текущий материализованный результат.

Cursor.description определяет null_ok по типу каждого столбца результата. Типы, не допускающие NULL, возвращают False, а допускающие NULL — True, включая обёртки Nullable, Variant и Dynamic. None означает, что допустимость NULL неизвестна. Если запрос, начинающийся с SELECT или WITH без учёта начальных комментариев, не возвращает строк и метаданных столбцов, курсор выполняет запрос метаданных с LIMIT 0, чтобы заполнить description. Если этот запрос метаданных завершается с ошибкой, description остаётся пустым.

ClickHouse не поддерживает традиционные транзакции через этот HTTP-интерфейс. Connection.commit() и Connection.rollback() не выполняют никаких действий. Правила параллелизма для идентификатора сеанса по-прежнему действуют, если соединение используется совместно.

Следующие модули предоставляют дополнительные общедоступные вспомогательные средства, используемые клиентскими приложениями.

Версия установленного пакета доступна как строка clickhouse_connect.__version__.

Пользовательские исключения, включая иерархию исключений DB-API 2.0, определены в clickhouse_connect.driver.exceptions. DatabaseError и OperationalError предоставляют числовой атрибут code с кодом ошибки ClickHouse и атрибут name с символьным именем, например UNKNOWN_TABLE, поэтому приложения могут строить логику на основе exc.code, а не разбирать сообщение. code задается, даже если show_clickhouse_errors отключен, тогда как для name требуется отображение подробностей ошибки (True или "scrub"). Оба имеют значение None, когда недоступны, например при ошибках передачи данных. Используйте show_clickhouse_errors="scrub", когда конечные пользователи должны видеть ошибки SQL без информации о хосте или версии сервера. Этот параметр также управляет сообщениями StreamFailureError в процессе потоковой передачи и общими сообщениями об ошибках передачи данных. Он влияет только на str(exc). Ошибки передачи данных по-прежнему прикрепляются как __cause__, а трассировки стека могут содержать исходные сведения о хосте, URL или текст ошибки библиотеки.

Функции и класс DT64Param в модуле clickhouse_connect.driver.binding можно использовать для корректного формирования и экранирования запросов ClickHouse SQL. Аналогично, функции из модуля clickhouse_connect.driver.parser можно использовать для разбора названий типов данных ClickHouse.

Подробнее об использовании ClickHouse Connect в многопоточных, многопроцессных и асинхронных/событийно-ориентированных приложениях см. в разделе Расширенное использование (многопоточные, многопроцессные и асинхронные/событийно-ориентированные сценарии использования).

Сведения о непосредственном использовании AsyncClient с asyncio см. в разделе Расширенное использование (AsyncClient).

Сведения об управлении идентификаторами сеансов ClickHouse в многопоточных или параллельно работающих приложениях см. в разделе Расширенное использование (Управление идентификаторами сеансов ClickHouse).

Сведения о настройке пула HTTP-соединений для крупных многопоточных приложений см. в разделе Расширенное использование (настройка пула HTTP-соединений).