Поддержка SQLAlchemy

Поддержка SQLAlchemy и Alembic для ClickHouse

ClickHouse Connect включает диалект SQLAlchemy (clickhousedb), созданный на базе основного драйвера. Синхронный диалект поддерживает SQLAlchemy 1.4.40 и более поздние версии, включая SQLAlchemy 2.x, с акцентом на запросы Core, DDL ClickHouse, рефлексию и простые ORM-вставки. Для асинхронного диалекта требуется SQLAlchemy версии 2.0.44 или более поздней.

Установите зависимости SQLAlchemy с помощью дополнительного пакета:

pip install "clickhouse-connect[sqlalchemy]"

​

Подключение через SQLAlchemy

Создайте движок, указав URL в формате clickhousedb:// или clickhousedb+connect://:

from sqlalchemy import create_engine, text

engine = create_engine(
    "clickhousedb://user:password@host:8123/mydb?compression=zstd"
)

with engine.connect() as conn:
    version = conn.execute(text("SELECT version()")).scalar_one()
    print(version)

​

Идентификаторы сеансов ClickHouse

По умолчанию каждое соединение из пула — как в синхронном, так и в асинхронном диалекте — генерирует собственный идентификатор сеанса ClickHouse. Если запросы через это соединение попадают в один и тот же процесс сервера ClickHouse, настройки, изменённые с помощью SET, и временные таблицы сохраняются для этого соединения. Состояние именованного сеанса и проверка на пересечение запросов в одном сеансе локальны для процесса. В пределах одного процесса сервера пересекающийся по времени запрос от того же пользователя с тем же идентификатором сеанса не ставится в очередь, а сразу отклоняется с кодом ошибки сервера 373. Если вы задаёте фиксированный session_id, используйте pool_size=1, max_overflow=0 или сериализуйте доступ до того, как запросы попадут в ClickHouse. В ClickHouse Cloud и других развертываниях с балансировкой нагрузки запросы с одним и тем же идентификатором сеанса могут попадать на разные серверы, поэтому не используйте фиксированный session_id в качестве распределённого состояния или распределённого мьютекса.

​

Асинхронные соединения

Для асинхронного диалекта требуется SQLAlchemy версии 2.0.44 или новее; он использует нативный AsyncClient из ClickHouse Connect. Установите необходимые зависимости и создайте асинхронный движок с URL clickhousedb+async://:

pip install "clickhouse-connect[sqlalchemy-async]"

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine


async def main():
    engine = create_async_engine(
        "clickhousedb+async://user:password@host:8123/mydb"
    )
    try:
        async with engine.connect() as conn:
            version = (await conn.execute(text("SELECT version()"))).scalar_one()
            print(version)
    finally:
        await engine.dispose()


asyncio.run(main())

Результаты буферизуются. Серверные курсоры отключены, поэтому AsyncConnection.stream() вызывает InvalidRequestError. SQLAlchemy допускает вызов AsyncSession.stream(), однако диалект буферизует весь результат, прежде чем вернуть его. Для больших объёмов результатов используйте нативные потоковые методы AsyncClient. Исходный нативный клиент доступен через driver_connection, пока соответствующее соединение SQLAlchemy взято из пула:

async def stream_events(engine):
    async with engine.connect() as conn:
        raw_connection = await conn.get_raw_connection()
        client = raw_connection.driver_connection
        async with await client.query_rows_stream("SELECT * FROM events") as rows:
            async for row in rows:
                print(row)

Не используйте соединение SQLAlchemy параллельно с его низкоуровневым клиентом. Завершайте потоки низкоуровневого клиента до выхода из блока соединения SQLAlchemy и не сохраняйте ссылку на низкоуровневый клиент после возврата соединения в пул. Жизненным циклом заимствованного клиента управляет SQLAlchemy, поэтому никогда не вызывайте client.close() или какие-либо его приватные методы жизненного цикла. Параллелизмом соединений управляет пул SQLAlchemy. Каждому соединению в пуле принадлежит один нативный асинхронный клиент, и по умолчанию коннектор aiohttp ограничен одним соединением в целом и одним соединением на хост. Чтобы переопределить эти транспортные настройки, задайте connector_limit, connector_limit_per_host или keepalive_timeout в URL или в connect_args. При pool_pre_ping=True SQLAlchemy проверяет повторно используемые соединения запросом SELECT 1 при извлечении соединения из пула.

Асинхронные вставки SQLAlchemy через executemany в настоящее время отправляют отдельный HTTP-запрос для каждого набора параметров вместо использования нативного протокола массовой вставки драйвера. Используйте этот способ только для небольших пакетов. Для больших объёмов данных применяйте описанный выше шаблон доступа через driver_connection, принадлежащий пулу, и дожидайтесь завершения client.insert() до возврата соединения SQLAlchemy в пул. Поскольку асинхронный executemany использует привязку параметров запроса, наивные значения datetime обрабатываются согласно настройке naive_datetime_binding, а не naive_datetime_insert, которая применяется в синхронном нативном executemany. Типизированные привязки SQLAlchemy DateTime64 сохраняют доли секунды как для клиентских, так и для серверных параметров. Нетипизированные параметры %s или %(name)s, передаваемые в exec_driver_sql(), по умолчанию форматируют наивные значения datetime с точностью до целых секунд. Для однозначной обработки часовых поясов используйте значения с явно указанным часовым поясом. Для нативной семантики массовой вставки используйте client.insert().

Создавайте и освобождайте асинхронный движок в том цикле событий, в котором он используется. Верните все извлечённые соединения, а затем дождитесь выполнения engine.dispose() при завершении работы и перед использованием движка из другого цикла событий. Если цикл, которому принадлежит движок, уже закрыт, перед повторным использованием дождитесь выполнения engine.dispose() в текущем цикле. aiohttp может сообщать о незакрытом транспорте, если очистка начинается уже после закрытия цикла-владельца, поэтому по возможности освобождайте движок до его передачи. pool_pre_ping=True не заменяет освобождение при переносе асинхронного движка с пулом между циклами событий. Чтобы использовать один движок в нескольких циклах событий, не удерживая привязанные к циклу соединения, задайте poolclass=NullPool. Если освобождение выполняется, пока соединение ещё не возвращено в пул, диалект закроет это соединение при его возврате или сборке мусора. Не вызывайте engine.sync_engine.dispose() из синхронного кода: там SQLAlchemy не может дождаться асинхронной очистки соединений и может лишь записать ошибку в журнал, не закрыв транспорты пула.

Параметры запроса в URL могут содержать настройки ClickHouse, параметры клиента ClickHouse Connect, такие как compression, query_limit и тайм-ауты, или параметры HTTP/TLS, такие как ca_cert. При необходимости добавьте к настройке ClickHouse префикс ch_, чтобы она принудительно обрабатывалась как настройка сервера, например ch_http_max_field_name_size=99999.

Доступные параметры клиента см. в разделе Аргументы соединения и настройки.

Синхронные вспомогательные операции SQLAlchemy, например DDL и инспекцию схемы, выполняйте через AsyncConnection.run_sync():

from sqlalchemy import inspect


async def prepare_schema(engine, metadata):
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)
        return await conn.run_sync(
            lambda sync_conn: inspect(sync_conn).get_table_names()
        )

​

Настройки для отдельных запросов

Передавайте настройки ClickHouse через параметры выполнения SQLAlchemy. Настройки можно задавать на уровне движка, соединения или оператора. Если один и тот же ключ задан в нескольких местах, значение на уровне оператора имеет приоритет над значением на уровне соединения или движка.

from sqlalchemy import text

stmt = text("SELECT getSetting('max_threads')").execution_options(
    settings={"max_threads": 2}
)

with engine.connect() as conn:
    value = conn.execute(stmt).scalar_one()

​

Форматы чтения для отдельных запросов

Задавайте форматы чтения ClickHouse для движка, соединения или оператора с помощью параметра выполнения SQLAlchemy query_formats. Форматы оператора применяются первыми и переопределяют соответствующие ключи и подстановочные шаблоны соединения или движка.

from sqlalchemy import text

stmt = text("SELECT user_uuid FROM users").execution_options(
    query_formats={"UUID": "string"}
)

with engine.connect() as conn:
    rows = conn.execute(stmt).all()

​

Обработка ошибок

Ошибки, которые драйвер выбрасывает через соединение SQLAlchemy, представлены классами DB-API, экспортируемыми из clickhouse_connect.dbapi. Это те же самые объекты классов, что и соответствующие классы в clickhouse_connect.driver.exceptions, поэтому SQLAlchemy оборачивает их в соответствующий подкласс sqlalchemy.exc.DBAPIError. StreamFailureError является подклассом OperationalError и оборачивается в sqlalchemy.exc.OperationalError.

Если отмена на стороне вызывающего кода может прервать явный вызов AsyncConnection.invalidate(), выполняйте инвалидацию в отдельной собственной задаче и дожидайтесь её завершения, прежде чем передавать отмену дальше. Так SQLAlchemy сможет завершить служебный учёт записей соединений:

import asyncio


async def invalidate_safely(connection):
    invalidate_task = asyncio.create_task(connection.invalidate())
    cancellation = None
    while not invalidate_task.done():
        try:
            await asyncio.wait({invalidate_task})
        except asyncio.CancelledError as ex:
            cancellation = ex
    if cancellation is not None:
        try:
            invalidate_task.result()
        finally:
            raise cancellation
    invalidate_task.result()

Не используйте соединение, пока задача его инвалидации ещё выполняется. Если прямой вызов await connection.invalidate() был отменён, а connection.invalidated по-прежнему равно false, снова выполните await connection.invalidate(), чтобы завершить очистку, и только после этого используйте или закрывайте соединение.

​

Серверные параметры

SQLAlchemy обычно подставляет параметры на стороне клиента. Чтобы использовать серверные параметры ClickHouse, включите их при создании движка:

engine = create_engine(
    "clickhousedb://user:password@host:8123/mydb",
    server_side_params=True,
)

Для асинхронного диалекта передайте тот же аргумент server_side_params=True в create_async_engine().

В этом режиме каждому связанному значению должен соответствовать SQLAlchemy-тип, совместимый с ClickHouse. Поддерживаемые списки IN преобразуются в типизированные параметры ClickHouse Array. Компилятор выдаёт CompileError, если не может определить совместимый тип или безопасно обработать привязку.

Имена привязок должны быть ASCII-именами ClickHouse типа BareWord. Имена, которые начинаются и заканчиваются на $, отклоняются, поскольку основной драйвер резервирует их для параметров запроса в виде необработанных двоичных данных.

​

Основные запросы

Диалект поддерживает запросы SELECT в SQLAlchemy Core с JOIN, фильтрами, сортировкой, ограничением и смещением, а также DISTINCT и составные SELECT.

SQLAlchemy union(), intersect() и except_() компилируются в ClickHouse UNION DISTINCT, INTERSECT DISTINCT и EXCEPT DISTINCT. Их аналоги union_all(), intersect_all() и except_all() компилируются в соответствующие операторы ALL. Это явное сопоставление сохраняет семантику дубликатов SQLAlchemy независимо от настроек по умолчанию для операций над множествами в ClickHouse.

from sqlalchemy import MetaData, Table, select

metadata = MetaData(schema="mydb")
users = Table("users", metadata, autoload_with=engine)
orders = Table("orders", metadata, autoload_with=engine)
events = Table("events", metadata, autoload_with=engine)

stmt = (
    select(users.c.name, orders.c.product)
    .select_from(users.join(orders, users.c.id == orders.c.user_id))
    .order_by(users.c.name)
    .limit(10)
)

with engine.connect() as conn:
    rows = conn.execute(stmt).all()

Поддерживается легковесный DELETE, требующий явного условия WHERE:

from sqlalchemy import delete

stmt = delete(users).where(users.c.name.like("%temporary%"))
with engine.connect() as conn:
    conn.execute(stmt)

​

Подстановка литералов

Когда SQLAlchemy подставляет связанное значение в запрос через literal_binds или literal_execute, диалект использует правила экранирования ClickHouse для общих строковых типов и типов ClickHouse. Это также относится к обёрткам TypeDecorator и вариантам, выбранным с помощью with_variant(). Строковые значения сохраняют знаки процента и обратные косые черты, даже если другие связанные параметры не подставляются.

Значения Python datetime с типом SQLAlchemy ClickHouse DateTime64 сохраняют микросекунды как в параметрах на стороне клиента, так и во встроенных литералах, в том числе для значений Nullable и значений, вложенных в массивы и кортежи. ClickHouse применяет объявленную точность. Python datetime поддерживает до шести цифр в дробной части. Обычные значения DateTime форматируются с точностью до целых секунд. Для оператора text() явно укажите тип с помощью bindparam("ts", type_=DateTime64(6)), чтобы сохранить доли секунды.

Типы столбцов SQLAlchemy должны соответствовать схеме на сервере. Если объявить DateTime64 для серверного столбца типа DateTime, в запрос будут подставляться доли секунды, что может приводить к ошибкам преобразования при вставке и в сравнениях IN.

В SQLAlchemy 2.x для встроенных литералов общих типов sqlalchemy.ARRAY, содержащих элементы ClickHouse Tuple, необходимо указать dimensions=1 (или соответствующую большую размерность для вложенных массивов), чтобы SQLAlchemy рассматривал каждый кортеж как один элемент. SQLAlchemy 1.4 не поддерживает встроенные литералы для общих типов ARRAY.

Если именованный параметр даты и времени используется повторно, для сохранения долей секунды каждое его вхождение должно иметь совместимый тип привязки DateTime64. Вхождение без типа или с конфликтующим типом форматируется с точностью до целых секунд. Укажите type_=DateTime64(6) для каждого bindparam или используйте разные имена параметров с соответствующими типами.

​

Подсказки типов JSON

Типизированные JSON-пути объявляются через отображение typed_paths. Типом пути может быть класс типа ClickHouse SQLAlchemy, настроенный экземпляр или строка с именем типа ClickHouse. Строки с именами типов позволяют задавать типы, у которых нет конструктора SQLAlchemy, например Dynamic, и при этом подходят для сложных настроенных выражений типов. Они сохраняют имена в именованном Tuple.

Строки с именами типов могут содержать настроенные вложенные типы JSON, например Array(JSON(`child` UInt32)). Распознаваемые имена типов ClickHouse в таких строках регистронезависимы и выводятся в канонической капитализации. Строка должна содержать ровно одно полное выражение типа. Завершающий текст и некорректные вложенные аргументы JSON отклоняются.

Пустой Tuple() не поддерживается в качестве типизированного JSON-пути, так как ClickHouse не может сериализовать его через Native format JSON-столбца. Основной драйвер поддерживает Tuple() в столбцах запроса и вставки в любой позиции, в том числе вложенным в позиционные или именованные кортежи, внутри Array, а также как Nullable(Tuple()) там, где это включено на сервере.

from sqlalchemy import Column, MetaData, Table

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import JSON, UInt32

events = Table(
    "events",
    MetaData(),
    Column(
        "payload",
        JSON(
            typed_paths={
                "event.id": UInt32,
                "details": "Tuple(id UInt32, label Nullable(String))",
                "attributes": "Variant(String, Array(String))",
            },
            max_dynamic_paths=256,
            max_dynamic_types=16,
            skip_paths=["internal.debug"],
            skip_regexps=[r"^private\."],
        ),
    ),
)

Для простых путей, являющихся идентификаторами Python, именованные аргументы служат сокращённой записью для typed_paths, например JSON(user_id=UInt32). Используйте typed_paths для путей с точками, пробелами, обратными кавычками, точками в кодировке %2E, а также для имён, совпадающих с параметрами конструктора. Типизированный путь с именем SKIP поддерживается через это отображение. Ключи в typed_paths и значения в skip_paths — это декодированные имена. Обратные и двойные кавычки в начале или в конце имени трактуются как обычные символы пути, а не как заранее применённое SQL-экранирование. Внутри необработанной строки типа обратные и двойные кавычки являются синтаксисом идентификаторов ClickHouse.

Можно задать до 1000 типизированных путей. max_dynamic_paths принимает значения от 0 до 10000, max_dynamic_types — от 0 до 254. Эти диапазоны действуют и внутри необработанных строк вложенных JSON-типов. Явные серверные значения по умолчанию 1024 и 32 в генерируемый DDL не включаются. Для обычных пропускаемых путей выполняется дедупликация. Строки регулярных выражений не проверяются средствами Python, поскольку ClickHouse использует синтаксис RE2. Дублирующиеся регулярные выражения сохраняются.

Обычный пропускаемый путь не может называться в точности REGEXP, так как ClickHouse резервирует этот токен для SKIP REGEXP. Имена вида REGEXP_foo остаются допустимыми. В необработанной строке JSON-типа операнд обычного SKIP должен быть одним идентификатором ClickHouse либо составным идентификатором, разделённым точками. Составной идентификатор без кавычек не может начинаться с REGEXP; заключайте первый компонент в кавычки, если он является данными пути. У SKIP REGEXP должен быть ровно один строковый литерал в одинарных кавычках. Части идентификатора заключайте в обратные или двойные кавычки, если они содержат пробелы или знаки пунктуации. Необработанные подсказки JSON-типа поддерживают Variant(...); для отдельного Variant публичного конструктора SQLAlchemy нет. Члены Variant упорядочиваются и дедуплицируются по тем же каноническим именам, которые использует ClickHouse.

Конструктор упорядочивает аргументы в той же канонической форме, которую возвращает ClickHouse. Отражённые типы, копии типов SQLAlchemy и автогенерация Alembic сохраняют эту конфигурацию.

​

Подстолбцы JSON

Для столбца, объявленного или представленного как JSON в ClickHouse, используйте квадратные скобки, чтобы выбирать по одному сегменту пути к подстолбцу, хранящемуся в базе:

from sqlalchemy import Column, MetaData, Table, select

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import JSON, UInt32

events = Table(
    "events",
    MetaData(),
    Column("payload", JSON),
)

request_id = events.c.payload["context"]["request"].subcolumn(
    "id",
    type_=UInt32,
)

stmt = select(
    events.c.payload["severity"].label("severity"),
    request_id.label("request_id"),
)

payload["severity"] компилируется в синтаксис точечного идентификатора ClickHouse. Каждая часть заключается в кавычки отдельно, например `events`.`payload`.`severity`. При этом считывается сохранённый подстолбец JSON ClickHouse без вызова getSubcolumn. Для каждого сегмента пути последовательно применяйте [] или .subcolumn(). Каждый сегмент должен быть непустой строкой.

Передача type_ в .subcolumn() оборачивает точечный путь в SQL CAST и назначает этот тип выражению SQLAlchemy. Без type_ .subcolumn("segment") работает так же, как ["segment"].

Нетипизированный путь имеет тип Dynamic ClickHouse. ClickHouse не допускает использование значений Dynamic непосредственно в ORDER BY или GROUP BY. Передавайте type_, если подстолбец используется в них.

Для статически типизированного кода импортируйте json_subcolumn из clickhouse_connect.cc_sqlalchemy. Эта вспомогательная функция также принимает по одному сегменту за раз и сохраняет тип результата Python из type_:

from clickhouse_connect.cc_sqlalchemy import json_subcolumn

context = json_subcolumn(events.c.payload, "context")
request = json_subcolumn(context, "request")
request_id = json_subcolumn(request, "id", type_=UInt32)

В этом примере средства проверки типов определяют request_id как ColumnElement[int].

Каждый сегмент заключается в кавычки отдельно, в том числе имена с пробелами или обратными кавычками. Обратные кавычки не делают точку литеральной при обработке JSON-путей в ClickHouse. Если включен json_type_escape_dots_in_keys, используйте кодирование ClickHouse %2E для литеральных точек в ключах. Обращайтесь к ключу с именем a.b через payload["a%2Eb"], а не через payload["a.b"].

​

Расширения запросов к ClickHouse

Импортируйте select из clickhouse_connect.cc_sqlalchemy, чтобы типизированные методы ClickHouse были доступны средствам статической проверки типов. Эти методы также доступны в стандартном sqlalchemy.select при выполнении.

from clickhouse_connect.cc_sqlalchemy import select

stmt = (
    select(events.c.user_id, events.c.event_type)
    .final()
    .prewhere(events.c.event_date >= "2026-01-01")
    .sample(0.1)
    .limit_by([events.c.user_id], 3)
)

Методы ClickHouse Select:

Method	SQL feature
.final()	FINAL для таблицы
.sample(value)	SAMPLE с использованием доли, количества строк или выражения
.prewhere(expression)	PREWHERE; повторные вызовы объединяются с помощью AND
.limit_by(columns, limit, offset=None)	LIMIT ... BY
.array_join(...)	ARRAY JOIN
.left_array_join(...)	LEFT ARRAY JOIN
.ch_join(...)	JOIN в ClickHouse с параметрами strictness, distribution, using и cross
.cte(name, materialized=True)	WITH name AS MATERIALIZED (...)

Select.with_hint() в SQLAlchemy — это API подсказок для таблиц. Диалект ClickHouse не формирует подсказки для таблиц. Подходящая подсказка с wildcard или подсказка clickhousedb вызывает SAWarning и не изменяет сгенерированный SQL. Для этих секций ClickHouse используйте final(), sample(), prewhere() или limit_by().

Select.with_statement_hint() — это API для необработанных завершающих директив. Он добавляет переданный текст в конец SELECT без проверки, специфичной для ClickHouse. Этот API по-прежнему доступен для доверенного статического SQL, такого как SETTINGS max_threads=1:

stmt = select(events.c.id).with_statement_hint("SETTINGS max_threads=1")

Для настроек ClickHouse рекомендуется использовать параметры выполнения, чтобы драйвер обрабатывал их отдельно от текста SQL:

stmt = select(events.c.id).execution_options(settings={"max_threads": 1})

Например, GLOBAL ANY LEFT JOIN в ClickHouse можно вызывать по цепочке без вложения пользовательского FromClause:

stmt = (
    select(events.c.id, users.c.name)
    .select_from(events)
    .ch_join(
        users,
        events.c.user_id == users.c.id,
        isouter=True,
        strictness="ANY",
        distribution="GLOBAL",
    )
)

Используйте явную конструкцию Lambda для функций высшего порядка в ClickHouse:

from sqlalchemy import column, func

from clickhouse_connect.cc_sqlalchemy import Lambda, select

stmt = select(
    func.arrayMap(
        Lambda("x", column("x") * 2),
        events.c.metrics,
    ).label("doubled")
)

Стандартная конструкция SQLAlchemy values() компилируется в синтаксис табличной функции ClickHouse VALUES, в том числе при использовании в общем табличном выражении (CTE). Для формы CTE требуется SQLAlchemy 2.0.42 или более поздней версии, в которой был добавлен Values.cte().

​

Материализованные CTE

По умолчанию ClickHouse подставляет тело общего табличного выражения (CTE), поэтому при каждом обращении к CTE его тело выполняется заново. Передайте materialized=True в .cte(), чтобы сгенерировать WITH <name> AS MATERIALIZED (...), при котором тело вычисляется один раз:

from sqlalchemy import func

from clickhouse_connect.cc_sqlalchemy import select

ranked = (
    select(book.c.book_id, func.row_number().over(order_by=book.c.score.desc()).label("result_rank"))
    .where(book.c.genre == "sci-fi")
    .order_by(book.c.score.desc())
    .limit(100)
    .cte("ranked", materialized=True)
)

stmt = (
    select(book.c.book_id, ranked.c.result_rank)
    .select_from(book)
    .ch_join(ranked, book.c.book_id == ranked.c.book_id, strictness="ANY")
    .where(book.c.book_id.in_(select(ranked.c.book_id)))
    .execution_options(settings={"enable_materialized_cte": 1, "enable_analyzer": 1})
)

Сервер материализует CTE только при наличии ключевого слова MATERIALIZED, значении enable_materialized_cte=1 и включенном analyzer. Установите enable_materialized_cte для оператора, подключения или движка, как показано в разделе Настройки для отдельных запросов. Analyzer по умолчанию включен на всех серверах, поддерживающих эту возможность, поэтому явная установка enable_analyzer=1 служит дополнительной мерой предосторожности. enable_materialized_cte — экспериментальная настройка ClickHouse. При enable_materialized_cte=0 или enable_analyzer=0 запрос успешно выполняется и возвращает те же строки. ClickHouse молча игнорирует MATERIALIZED и снова разворачивает CTE, поэтому пропущенная настройка снижает производительность без каких-либо сообщений. Для материализованных CTE требуется ClickHouse 26.3 или более поздней версии. Более старые серверы отклоняют ключевое слово с синтаксической ошибкой.

Для оператора, построенного с помощью стандартного sqlalchemy.select, вместо этого используйте cte() уровня модуля. В качестве первого аргумента она принимает оператор, а в остальном повторяет Select.cte():

from sqlalchemy import select as sa_select

from clickhouse_connect.cc_sqlalchemy import cte

ranked = cte(sa_select(book.c.book_id), "ranked", materialized=True)

Ключевое слово применяется только в диалекте ClickHouse, поэтому оператор, используемый с другим backend-соединением, компилируется там без изменений.

ClickHouse не поддерживает рекурсивные материализованные CTE. Вспомогательные функции SQLAlchemy вызывают ValueError, если одновременно заданы recursive=True и materialized=True.

​

DDL и рефлексия

ClickHouse Connect предоставляет типы данных ClickHouse, движки таблиц, конструкции для словарей, DDL для баз данных и рефлексию таблиц.

Отдельные столбцы Variant отражаются через внутренний тип SQLAlchemy, а автогенерация Alembic сохраняет их канонические исходные имена типов без повторяющихся изменений типа. Столбцы Geometry и MultiPoint отражаются как публичные типы SQLAlchemy.

import sqlalchemy as db
from sqlalchemy import MetaData

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import DateTime64, String, UInt32
from clickhouse_connect.cc_sqlalchemy.ddl.custom import CreateDatabase
from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import MergeTree

with engine.connect() as conn:
    conn.execute(CreateDatabase("example_db", exists_ok=True))

    metadata = MetaData(schema="example_db")
    events = db.Table(
        "events",
        metadata,
        db.Column("id", UInt32, primary_key=True),
        db.Column("user", String),
        db.Column("created_at", DateTime64(3)),
        MergeTree(order_by="id"),
    )
    events.create(conn)

    reflected = db.Table("events", MetaData(schema="example_db"), autoload_with=conn)
    assert reflected.engine is not None

Отражённые столбцы содержат server_default для выражений DEFAULT, а также специфичные для диалекта атрибуты, такие как clickhouse_codec, clickhouse_ttl, clickhouse_materialized и clickhouse_alias, если они заданы.

Для строковых значений в секциях DEFAULT, MATERIALIZED, ALIAS и TTL используется экранирование строк ClickHouse. Такое же экранирование применяется к комментариям таблиц, словарей и столбцов, включая комментарии, создаваемые Alembic.

Аргументы ключа MergeTree, такие как order_by, partition_by, primary_key, sample_by и ttl, принимают столбцы SQLAlchemy, SQL-выражения, а также обычные строки.

Memory(), Log(), StripeLog(), TinyLog(), Null() и Set() можно вызывать без аргументов, и они без потерь проходят полный цикл автогенерации Alembic. Существующий аргумент-словарь по-прежнему поддерживается. Для передачи настроек движка используйте settings={...}.

SummingMergeTree и ReplicatedSummingMergeTree принимают необязательный аргумент columns, который можно передать только как именованный. Существующие позиционные аргументы сохраняют прежнее значение, поэтому SummingMergeTree("id") по-прежнему задаёт ORDER BY id.

from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import SummingMergeTree

engine_clause = SummingMergeTree("id", columns=("delta", "n_tx"))
# Sum delta and n_tx for rows with the same id.

Передайте строку, столбец SQLAlchemy, сопоставленный атрибут столбца или непустой список либо кортеж таких значений. Строковые элементы списка и кортежа заключаются в кавычки как идентификаторы. Скалярная строка задаёт Raw SQL, например "delta" или "(delta, n_tx)". Сервер допускает для этих столбцов только идентификаторы. Если не указывать columns, ClickHouse сам выберет столбцы для суммирования. Рефлексия и автогенерация Alembic сохраняют явно заданный список столбцов.

​

Вставка данных и базовое использование ORM

Поддерживаются вставки через Core и простые модели ORM. Для синхронного диалекта при массовой загрузке данных предпочтительнее использовать вставки через Core с executemany там, где это поддерживается. Для асинхронной массовой вставки используйте нативный метод AsyncClient.insert(), описанный в разделе Асинхронные соединения.

with engine.connect() as conn:
    conn.execute(
        events.insert(),
        [
            {"id": 13, "user": "user_1"},
            {"id": 79, "user": "user_2"},
        ],
    )

Для синхронного диалекта обычные вставки Core через executemany, сгенерированные компилятором SQLAlchemy, выполняются как одна массовая вставка в формате Native. Асинхронный executemany отправляет отдельный запрос для каждого набора параметров, как описано в разделе Асинхронные соединения. Raw SQL, а также вставки с выражениями или иной семантикой, которую нельзя безопасно перенаправить, сохраняют исходный SQL и выполняются отдельно для каждого набора параметров. Если один из последующих наборов параметров завершится с ошибкой, строки, записанные предыдущими наборами, останутся закоммиченными.

Явные многострочные операторы insert(events).values([...]) поддерживают строки в виде словарей, кортежи в порядке столбцов таблицы и SQL-выражения для отдельных строк. Именно эту форму использует Pandas to_sql(method="multi"). Строки при этом вставляются, но возвращается 0, поскольку текстовые операторы INSERT сообщают через курсор DB-API количество строк 0. SQLAlchemy определяет список столбцов по первой строке. Лишние ключи словарей в последующих строках и значения кортежей, выходящие за пределы этого списка столбцов, игнорируются. Если в одной из последующих строк отсутствует значение для выбранного столбца, компиляция завершается ошибкой. Задавайте во всех строках одинаковый набор столбцов.

При стандартных ограничениях HTTP-формы в ClickHouse 26.4 и новее server_side_params=True подходит только для небольших явных батчей — примерно до 1000 привязываемых значений с запасом для других полей. Этот предел можно повысить в конфигурации сервера. Для больших обычных батчей в синхронном диалекте передавайте строки вторым аргументом execute(), чтобы драйвер мог выполнить массовую вставку в формате Native. Для асинхронной загрузки больших объёмов данных используйте await с нативным методом AsyncClient.insert().

import sqlalchemy as db
from sqlalchemy import MetaData
from sqlalchemy.orm import Session, declarative_base

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import String, UInt32
from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import MergeTree

Base = declarative_base(metadata=MetaData(schema="example_db"))


class User(Base):
    __tablename__ = "users"
    __table_args__ = (MergeTree(order_by=["id"]),)

    id = db.Column(UInt32, primary_key=True)
    name = db.Column(String)


Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(User(id=13, name="user_1"))
    session.bulk_save_objects([User(id=79, name="user_2")])
    session.commit()

​

Миграции Alembic

ClickHouse Connect поддерживает интеграцию с Alembic для миграций схем ClickHouse. Установите её с помощью:

pip install "clickhouse-connect[alembic]"

Для миграций через асинхронный диалект установите оба дополнительных компонента:

pip install "clickhouse-connect[alembic,sqlalchemy-async]"

Создайте асинхронный проект Alembic, затем замените сгенерированное окружение примером с поддержкой ClickHouse:

alembic init -t async alembic

Сгенерированный alembic.ini использует script_location = %(here)s/alembic. Оставьте эту настройку, если каталог миграций называется alembic, или укажите в ней каталог, переданный в alembic init. Замените alembic/env.py на асинхронный пример env.py для Alembic из репозитория, затем задайте sqlalchemy.url в alembic.ini.

Импортируйте clickhouse_connect.cc_sqlalchemy.alembic в env.py Alembic, чтобы зарегистрировать интеграцию диалекта. Автогенерация поддерживает типовые изменения таблиц, включая создание и удаление таблиц, добавление/изменение/удаление столбцов, значения по умолчанию и комментарии. Для переименования таблиц и столбцов используйте ручные операции. Проверяйте каждую сгенерированную миграцию перед её применением.

Функции миграций Alembic остаются синхронными. Асинхронное окружение создаёт AsyncEngine, открывает AsyncConnection и передаёт синхронную функцию миграции в await connection.run_sync(...). Офлайн-миграции напрямую вызывают context.configure(url=..., literal_binds=True, dialect_opts={"paramstyle": "named"}) и не создают движок. Асинхронный пример env.py для Alembic из репозитория поддерживает оба варианта и считывает URL соединения из стандартного параметра конфигурации Alembic sqlalchemy.url. В нём сохранены хуки и параметры ClickHouse для Alembic из подробного примера, включая include_object, make_include_name(...), clickhouse_writer и version_table. Не используйте engine.sync_engine для запуска асинхронных миграций или освобождения их ресурсов.

Специальные для ClickHouse хелперы op.* охватывают:

• Индексы пропуска данных, включая операции добавления, материализации и удаления.
• Проекции, включая операции добавления, материализации и удаления.
• Изменение и сброс настроек таблиц семейства MergeTree.
• Создание и удаление materialized view.
• Создание, удаление и перезагрузку словарей.

Индексы пропуска данных ClickHouse — это не индексы SQLAlchemy. Index, Column(index=True), op.create_index и op.drop_index отклоняются, чтобы избежать частичного или некорректного DDL. Используйте op.add_clickhouse_index и op.drop_clickhouse_index.

См. полный пример работы с Alembic. Пользователям, переходящим с clickhouse-sqlalchemy, также следует ознакомиться с руководством по миграции.

​

Область применения и ограничения

• ClickHouse не поддерживает традиционные транзакции через этот HTTP-диалект. engine.begin() и Session.commit() организуют работу на стороне Python, но коммит и rollback на сервере ничего не меняют.
• UPDATE, двухфазные транзакции, последовательности, RETURNING и расширенные уровни изоляции в этом диалекте не реализованы. При необходимости используйте явный ClickHouse SQL для серверных мутаций.
• Column(..., primary_key=True) задает identity объекта SQLAlchemy. Это не создает ограничение уникальности на стороне сервера. Задавайте сортировку и необязательные выражения первичного ключа через движок таблицы.
• Метаданные для традиционных внешних ключей, ограничений уникальности и стандартных индексов недоступны, поскольку ClickHouse не применяет такие ограничения.
• Управление relationship в ORM, обновления unit of work, каскады, а также немедленная или отложенная загрузка relationship не входят в поддерживаемую область ORM.