SQLAlchemy サポート

ClickHouse Connect には、コアドライバー上に構築された clickhousedb SQLAlchemy ダイアレクトが含まれています。これは SQLAlchemy 1.4.40 以降 (SQLAlchemy 2.x を含む) をサポートしており、Core クエリ、ClickHouse DDL、リフレクション、およびシンプルな ORM insert に重点を置いています。

パッケージ extra を使用して SQLAlchemy の依存関係をインストールします:

pip install "clickhouse-connect[sqlalchemy]"

clickhousedb:// または clickhousedb+connect:// のいずれかの URL 形式で engine を作成します。

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)

URLクエリパラメータには、ClickHouse設定、compression、query_limit、タイムアウトなどの ClickHouse Connect クライアントオプション、または ca_cert などの HTTP/TLS オプションを含めることができます。必要に応じて、ClickHouse設定をサーバー設定として扱わせるには、先頭に ch_ を付けます。たとえば ch_http_max_field_name_size=99999 です。

利用可能なクライアントオプションについては、接続引数と設定 を参照してください。

SQLAlchemy の実行オプションを通じて ClickHouse の設定を渡します。設定は engine、connection、またはステートメントに指定できます。同じキーが指定されている場合、ステートメントの値が connection または engine の値よりも優先されます。

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

query_formats を指定した SQLAlchemy の実行オプションを使用して、ClickHouse の読み取りフォーマットを engine、connection、またはステートメントに設定できます。ステートメントのフォーマットが先に適用されるため、一致する connection または engine のオプションやワイルドカードよりも優先されます。

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 は通常、クライアント側でパラメータを展開します。engine の作成時に ClickHouse のサーバー側パラメータを有効にしてください。

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

このモードでは、バインドされるすべての値に ClickHouse と互換性のある SQLAlchemy の型が必要です。サポートされている IN リストは、型付きの ClickHouse Array パラメータになります。コンパイラは、互換性のある型を導出できない場合や、バインドを安全に処理できない場合に CompileError を発生させます。

バインド名は ClickHouse の ASCII BareWord 名である必要があります。先頭と末尾が $ の名前は、コアドライバーが生バイナリクエリパラメータ用に予約しているため拒否されます。

このダイアレクトは、JOIN、フィルター、並べ替え、LIMIT と OFFSET、DISTINCT、複合 SELECT を含む SQLAlchemy Core の SELECT クエリをサポートしています。

SQLAlchemy の union()、intersect()、except_() は、ClickHouse の UNION DISTINCT、INTERSECT DISTINCT、EXCEPT DISTINCT にコンパイルされます。これらに対応する union_all()、intersect_all()、except_all() は、それぞれ対応する ALL 演算子にコンパイルされます。この明示的なマッピングにより、ClickHouse の集合演算のデフォルト設定にかかわらず、SQLAlchemy の重複に関するセマンティクスが保持されます。

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

論理削除がサポートされており、明示的な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 によってバインド値をインライン展開する場合、ダイアレクト は汎用 String 型および ClickHouse 型に ClickHouse のクォーティングを使用します。これは TypeDecorator ラッパーおよび with_variant() の選択にも適用されます。他のバインドパラメータが残っている場合でも、文字列値内のパーセント記号とバックスラッシュは保持されます。

ClickHouse JSON として宣言または反映されたカラムでは、角括弧を使用して、ストレージでサポートされるサブカラムパスのセグメントを一度に1つずつ選択します。

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` となります。これは ClickHouse に格納されている JSON サブカラムを読み取り、getSubcolumn は呼び出しません。パスセグメントごとに [] または .subcolumn() を 1 回ずつ連結します。各セグメントは空でない文字列である必要があります。

.subcolumn() に type_ を渡すと、ドット付きパスが SQL の CAST でラップされ、その型が SQLAlchemy 式に割り当てられます。type_ を指定しない場合、.subcolumn("segment") は ["segment"] と同様に動作します。

型なしパスの型は ClickHouse の Dynamic です。ClickHouse では、Dynamic 値を ORDER BY や GROUP BY で直接使用できません。そこでサブカラムを使用する場合は、type_ を渡してください。

静的型付けされたコードでは、clickhouse_connect.cc_sqlalchemy から json_subcolumn をインポートします。このヘルパーも一度に 1 つのセグメントを受け取り、type_ で指定した Python の結果型を保持します。

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] として認識します。

スペースやバッククォートを含む名前も含め、各セグメントは個別に引用符で囲まれます。バッククォートを使用しても、ClickHouse の JSON path 処理でドットがリテラルとして扱われるわけではありません。json_type_escape_dots_in_keys が有効な場合、キー内のリテラルなドットには ClickHouse's %2E エンコーディングを使用します。a.b という名前のキーには、payload["a.b"] ではなく payload["a%2Eb"] でアクセスします。

静的型チェッカーで型付きの ClickHouse メソッドを利用できるようにするには、clickhouse_connect.cc_sqlalchemy から select をインポートします。標準の 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 メソッドは次のとおりです。

SQLAlchemy の Select.with_hint() はテーブルヒント用の API です。ClickHouse ダイアレクトではテーブルヒントは生成されません。適用可能なワイルドカードまたは clickhousedb ヒントを指定すると、SAWarning が発行され、生成された SQL は変更されません。これらの ClickHouse 句には、final()、sample()、prewhere()、または limit_by() を使用してください。

Select.with_statement_hint() は生の末尾ディレクティブ用 API です。ClickHouse 固有の検証を行わず、指定したテキストを SELECT の末尾に追加します。これは、SETTINGS max_threads=1 のような信頼できる静的 SQL で引き続き使用できます。

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

たとえば、ClickHouse の GLOBAL ANY LEFT JOIN は、カスタム 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",
    )
)

ClickHouseの高階関数では、明示的なLambda構文を使用します。

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 形式には、Values.cte() が追加された SQLAlchemy 2.0.42 以降が必要です。

デフォルトでは、ClickHouse は共通テーブル式をインライン化するため、複数回参照される CTE では、参照のたびにボディが実行されます。.cte() に materialized=True を渡すと、WITH <name> AS MATERIALIZED (...) が出力され、ボディは 1 回だけ計算されます。

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 をマテリアライズするのは、キーワードが指定され、enable_materialized_cte=1 が設定され、アナライザが有効な場合に限られます。クエリごとの設定に示すように、ステートメント、接続、またはエンジンで enable_materialized_cte を設定します。この機能をサポートするすべてのサーバーではアナライザがデフォルトで有効になっているため、enable_analyzer=1 を明示的に設定するのは予防的な措置です。enable_materialized_cte は実験的な ClickHouse 設定です。enable_materialized_cte=0 または enable_analyzer=0 の場合でも、クエリは成功し、同じ行を返します。ClickHouse は MATERIALIZED を通知なく無視して CTE を再びインライン化するため、設定を忘れてもエラーは発生せず、パフォーマンスが低下します。マテリアライズド CTE には ClickHouse 26.3 以降が必要です。古いサーバーでは、このキーワードは構文エラーとして拒否されます。

標準の sqlalchemy.select で構築したステートメントでは、代わりにモジュールレベルの cte() を使用します。これはステートメントを第 1 引数として受け取り、それ以外は 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 helpers は、recursive=True と materialized=True の両方が設定されている場合、ValueError を送出します。

ClickHouse Connect は、ClickHouse データ型、テーブルエンジン、Dictionary 機能、データベース DDL、テーブルリフレクションをサポートしています。

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

リフレクションで取得されたカラムには、DEFAULT 式に対応する server_default に加え、存在する場合は clickhouse_codec、clickhouse_ttl、clickhouse_materialized、clickhouse_alias などのダイアレクト 固有の属性も含まれます。

DEFAULT、MATERIALIZED、ALIAS、TTL 句内の文字列値では、ClickHouse の文字列エスケープを使用します。同じエスケープは、Alembic によって出力されるコメントを含む、テーブル、Dictionary、カラムのコメントにも適用されます。

order_by、partition_by、primary_key、sample_by、ttl などの MergeTree のキー引数では、SQLAlchemy のカラムや SQL 式に加えて、プレーンな文字列も指定できます。

Core による挿入とシンプルな ORM モデルをサポートしています。大量データを扱う処理には、Core による挿入を推奨します。

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

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

ClickHouse Connect には、ClickHouse のスキーマ移行向けの Alembic インテグレーションが含まれています。インストールするには、次を実行します。

pip install "clickhouse-connect[alembic]"

ダイアレクトのインテグレーションを登録するには、Alembic の env.py で clickhouse_connect.cc_sqlalchemy.alembic をインポートします。自動生成では、テーブルの作成と削除、カラムの追加/変更/削除、デフォルト値、コメントなど、一般的なテーブル変更に対応しています。テーブル名やカラム名のリネームには手動操作を使用してください。生成された移行は、適用前に必ずすべて確認してください。

ClickHouse 固有の op.* ヘルパーでは、次の操作をサポートしています。

• データスキッピングインデックス (追加、マテリアライズ、削除) 。
• プロジェクション (追加、マテリアライズ、削除) 。
• MergeTree テーブル設定の変更とリセット。
• materialized view の作成と削除。
• Dictionary の作成、削除、再読み込み。

ClickHouse のデータスキッピングインデックスは SQLAlchemy の索引ではありません。部分的または不正確な DDL を避けるため、Index、Column(index=True)、op.create_index、op.drop_index は使用できません。op.add_clickhouse_index と op.drop_clickhouse_index を使用してください。

完全な Alembic の実例 を参照してください。clickhouse-sqlalchemy から移行するユーザーは、移行ガイド も確認してください。

• ClickHouse は、この HTTP ダイアレクト では従来型のトランザクションを提供しません。engine.begin() と Session.commit() は Python 側の処理を整理しますが、commit と rollback はサーバー側では no-op です。
• UPDATE、二相トランザクション、シーケンス、RETURNING、および高度な分離レベルは、この ダイアレクト では実装されていません。必要に応じて、サーバー側のミューテーションには明示的に ClickHouse SQL を使用してください。
• Column(..., primary_key=True) は SQLAlchemy におけるオブジェクトの識別情報を提供します。これはサーバー側の一意制約を作成するものではありません。ソート順や必要に応じたプライマリキー式は、テーブルエンジン で定義してください。
• 従来の外部キー、一意制約、標準的な索引のメタデータは、ClickHouse がそれらの制約を強制しないため利用できません。
• ORM のリレーションシップ管理、unit-of-work による更新、カスケード、およびリレーションシップの即時または遅延ロードは、サポート対象の ORM の範囲外です。