SQLAlchemy サポート

ClickHouse SQLAlchemy および Alembic サポート

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

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

pip install "clickhouse-connect[sqlalchemy]"

​

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)

​

ClickHouseのセッションID

同期・非同期いずれのダイアレクトでも、プール内の各接続はデフォルトで個別のClickHouseセッションIDを生成します。その接続からのリクエストが同一のClickHouseサーバープロセスに到達する限り、SET で変更した設定や一時テーブルはその接続内で保持されます。名前付きセッションの状態と、同一セッションでの重複実行のチェックは、プロセス単位で管理されます。1つのサーバープロセス上で、同じユーザー・同じセッションIDのリクエストが重複した場合、そのリクエストはキューに入れられることなく、サーバーのエラーコード373で即座に拒否されます。固定の session_id を設定する場合は、pool_size=1, max_overflow=0 を指定するか、リクエストがClickHouseに到達する前にアクセスを直列化してください。ClickHouse Cloudなどロードバランサーを介した環境では、同じセッションIDを持つリクエストが別々のサーバーに到達する可能性があります。そのため、固定の session_id を分散状態や分散ミューテックスとして利用しないでください。

​

非同期接続

非同期ダイアレクトには SQLAlchemy 2.0.44 以降が必要で、ClickHouse Connect のネイティブな AsyncClient を使用します。必要な依存関係をインストールし、clickhousedb+async:// URL を指定して非同期エンジンを作成します:

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 を送出します。AsyncSession.stream() は SQLAlchemy 上では受け付けられますが、ダイアレクトは結果全体をバッファに格納してから返します。大きな結果を扱う場合は、ネイティブの AsyncClient のストリーミングメソッドを使用してください。SQLAlchemy の接続がチェックアウトされている間は、生のネイティブクライアントに driver_connection としてアクセスできます：

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 のプールが管理します。プールされた各接続はネイティブの非同期クライアントを 1 つ保持し、その aiohttp コネクタの上限はデフォルトで接続数 1、ホストあたりの接続数 1 に設定されます。これらのトランスポート設定を上書きするには、URL または connect_args で connector_limit、connector_limit_per_host、keepalive_timeout を設定します。pool_pre_ping=True を指定すると、SQLAlchemy はプールから接続をチェックアウトする際に、再利用する接続を SELECT 1 で確認します。

非同期 SQLAlchemy の executemany による挿入は、現時点ではドライバーの Native 一括挿入プロトコルを使用せず、パラメータセットごとに HTTP リクエストを 1 回送信します。この方法は小規模なバッチにのみ使用してください。大量のデータを扱う場合は、前述のプール管理の driver_connection アクセスパターンを使用し、SQLAlchemy の接続をプールに戻す前に client.insert() を await してください。非同期 executemany はクエリパラメータのバインディングを使用するため、naive な datetime 値は、同期の Native executemany で使用される naive_datetime_insert 設定ではなく、naive_datetime_binding に従います。型指定された SQLAlchemy の DateTime64 バインドでは、クライアント側・サーバー側どちらのパラメータでも秒の小数部が保持されます。exec_driver_sql() に渡す型指定のない %s や %(name)s パラメータでは、naive な datetime 値はデフォルトの秒単位のフォーマットのままとなります。タイムゾーンの扱いを明確にするには、タイムゾーン対応の値を使用してください。Native の一括挿入セマンティクスが必要な場合は client.insert() を使用してください。

非同期エンジンは、それを使用するイベントループ内で作成および破棄してください。チェックアウトした接続をすべて返却したうえで、シャットダウン時、および別のイベントループからエンジンを使用する前に engine.dispose() を await してください。エンジンを所有するループがすでに閉じられている場合は、再利用する前に現在のループで engine.dispose() を await してください。所有ループが閉じられた後にクリーンアップを開始すると、aiohttp が未クローズのトランスポートを報告することがあるため、可能であればループを移す前に破棄してください。プールされた非同期エンジンをイベントループ間で移動する場合、pool_pre_ping=True は破棄の代わりにはなりません。ループに紐づいた接続を保持せずに 1 つのエンジンを複数のイベントループで共有するには、poolclass=NullPool を設定してください。接続がチェックアウトされたままの状態で破棄が実行された場合、ダイアレクトはその接続が返却された時点、またはガベージコレクションされた時点で接続を閉じます。同期コードから engine.sync_engine.dispose() を呼び出さないでください。同期コードでは SQLAlchemy が非同期の接続クリーンアップを await できないため、プールされたトランスポートを閉じずにエラーをログに記録するだけになる場合があります。

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

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

DDL やインスペクションなどの同期 SQLAlchemy ヘルパーは、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()
        )

​

クエリごとの設定

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 の接続を介してドライバーが送出するエラーには、clickhouse_connect.dbapi からエクスポートされている DB-API クラスが使用されます。これらは 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 のままになっている場合は、接続を使用またはクローズする前に connection.invalidate() を再度 await して、クリーンアップを完了させてください。

​

サーバー側パラメータ

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

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

非同期ダイアレクトでは、create_async_engine() に同じ server_side_params=True 引数を指定してください。

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

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

​

Core クエリ

このダイアレクトは、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 の DateTime64 SQLAlchemy 型が指定された Python の datetime 値は、クライアント側のパラメータおよびインラインリテラルにおいてマイクロ秒が保持されます。これには Nullable な値や、配列およびタプル内にネストされた値も含まれます。ClickHouse は宣言された精度を適用します。Python の datetime が扱えるのは小数点以下最大 6 桁までです。通常の DateTime 値は秒単位のフォーマットのままです。text() ステートメントで秒未満の値を保持するには、bindparam("ts", type_=DateTime64(6)) のように型を明示的に指定してください。

SQLAlchemy のカラム型はサーバーのスキーマと一致させる必要があります。サーバー側の DateTime カラムに対して DateTime64 を宣言すると、秒未満の値がレンダリングされ、INSERT 時や IN 比較で変換エラーが発生する可能性があります。

SQLAlchemy 2.x では、ClickHouse の Tuple 要素を含む汎用 sqlalchemy.ARRAY 型をインラインリテラルとして使用する場合、SQLAlchemy が各タプルを 1 つの要素として扱えるよう、dimensions=1(ネストされた配列の場合は、それに応じたより大きな次元数)を指定する必要があります。SQLAlchemy 1.4 では、汎用 ARRAY 型のインラインリテラルはサポートされていません。

名前付きの datetime パラメータを再利用する場合、秒未満の値を保持するには、すべての出現箇所で互換性のある DateTime64 バインド型を指定する必要があります。型が指定されていない箇所や型が競合する箇所があると、秒単位のフォーマットのままになります。各 bindparam に type_=DateTime64(6) を設定するか、適切な型を指定した別々のパラメータ名を使用してください。

​

JSON型ヒント

typed_paths マッピングを使用して、型付きパスを宣言します。パスの型には、ClickHouse SQLAlchemyの型クラス、設定済みのインスタンス、またはClickHouseの型名文字列を指定できます。型名文字列は、Dynamic のようにSQLAlchemyのコンストラクタを持たない型にも対応しており、複雑な型式を設定する用途にも利用できます。また、名前付き Tuple の名前も保持されます。

型名文字列には、Array(JSON(`child` UInt32)) のような設定済みのネストされたJSON型を含めることができます。これらの文字列内で認識されるClickHouseの型名は大文字・小文字を区別せず、正規の表記で出力されます。文字列には、完全な型式を1つだけ含める必要があります。末尾に余分なテキストがある場合や、ネストされたJSON引数の形式が不正な場合は拒否されます。

空の Tuple() は、ClickHouseがJSON カラムのNative formatでシリアライズできないため、JSON パスとしてはサポートされません。コアドライバーは、クエリおよびinsertのカラムにおいて、位置依存タプルや名前付きタプルへのネスト、Array の内部、さらにserverで有効な場合の Nullable(Tuple()) を含め、任意の位置で 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)) 。ドットを含むパス、スペース、バッククォート、%2E でエンコードされたドット、あるいはコンストラクタのオプションと名前が衝突する場合には typed_paths を使用してください。SKIP という名前の型付きパスも、このマッピング経由でサポートされます。typed_paths のキーと skip_paths の値はデコード済みの名前です。先頭または末尾のバッククォートおよびダブルクォートは、あらかじめ適用されたSQLのクォートではなく、リテラルなパス文字として扱われます。生の型文字列の内部では、バッククォートとダブルクォートはClickHouseの識別子構文として扱われます。

型付きパスは最大1000個まで設定できます。max_dynamic_paths は0から10000まで、max_dynamic_types は0から254までを受け付けます。これらの範囲は、生のネストされたJSON型文字列の内部にも適用されます。サーバーのデフォルトである1024および32を明示的に指定した場合、生成されるDDLからは省略されます。プレーンなスキップパスは重複が除去されます。ClickHouseはRE2構文を使用するため、正規表現の文字列はPython側では検証されません。重複する正規表現はそのまま保持されます。

プレーンなスキップパスに REGEXP という名前をそのまま付けることはできません。ClickHouseがこのトークンを SKIP REGEXP 用に予約しているためです。REGEXP_foo のような名前は引き続き有効です。生のJSON型文字列では、プレーンな SKIP のオペランドは1つのClickHouse識別子か、ドット区切りの複合識別子でなければなりません。クォートされていない複合識別子を REGEXP で始めることはできません。先頭の部分がパスのデータである場合は、その部分をクォートしてください。SKIP REGEXP には、シングルクォートで囲まれた文字列リテラルを1つ指定する必要があります。識別子の各部分にスペースや記号が含まれる場合は、バッククォートまたはダブルクォートでクォートしてください。生のJSON型ヒントは Variant(...) をサポートしますが、単体の Variant には公開されたSQLAlchemyコンストラクタはありません。Variant のメンバーは、ClickHouseが使用するものと同じ正規化された名前に基づいて順序付けおよび重複除去が行われます。

コンストラクタは、ClickHouseが返す正規形と同じ形式で引数を並べます。リフレクションされた型、SQLAlchemyの型のコピー、およびAlembicの自動生成では、この設定が保持されます。

​

JSON サブカラム

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 メソッドを利用できるようにするには、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 メソッドは次のとおりです。

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(...)	strictness、distribution、using、cross オプション付きの ClickHouse JOIN
.cte(name, materialized=True)	WITH name AS MATERIALIZED (...)

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 以降が必要です。

​

マテリアライズド CTE

デフォルトでは、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 を送出します。

​

DDL とリフレクション

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

単独の Variant カラムは SQLAlchemy の内部型を通じてリフレクションされ、Alembic の 自動生成 は型変更を繰り返すことなく、その canonical な生の型名を保持します。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

リフレクションで取得されたカラムには、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 式に加えて、プレーンな文字列も指定できます。

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 のカラム、マッピングされたカラム属性、またはそれらからなる空でないリストまたはタプルを渡します。リストおよびタプル内の文字列要素は、識別子としてクォートされます。単一の文字列を渡した場合は、"delta" や "(delta, n_tx)" のような生の SQL として扱われます。サーバー側では、これらのカラムを識別子で指定する必要があります。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"},
        ],
    )

同期ダイアレクトでは、SQLAlchemy コンパイラが生成する通常の Core executemany による挿入は、1 回の Native 一括挿入として実行されます。非同期の executemany は、非同期接続 で説明しているとおり、パラメータセットごとに 1 回ずつリクエストを送信します。Raw SQL や、式を含むなど安全にルーティングできない挿入では、元の SQL がそのまま保持され、パラメータセットごとに 1 回ずつ実行されます。途中のパラメータセットで失敗した場合でも、それより前のパラメータセットで書き込まれた行はコミットされたまま残ります。

明示的な複数行の insert(events).values([...]) ステートメントは、辞書形式の行、テーブルのカラム順に並べたタプル、および行ごとの SQL 式に対応しています。Pandas の to_sql(method="multi") はこの形式を使用します。この場合、行は挿入されますが、テキスト形式の INSERT 文は DB-API カーソル経由で行数を 0 として報告するため、戻り値は 0 になります。SQLAlchemy は最初の行からカラムリストを決定します。2 行目以降に含まれる余分な辞書キーや、決定されたカラムリストの範囲外にあるタプル値は無視されます。2 行目以降でいずれかのカラムの値が欠けている場合は、コンパイルに失敗します。すべての行で同じカラムを指定してください。

ClickHouse 26.4 以降のデフォルトの HTTP フォーム制限では、server_side_params=True が適しているのは小規模な明示的バッチのみです。他のフィールド分のヘッドルームを確保したうえで、バインド値を約 1000 未満に抑えてください。この上限はサーバー設定で引き上げることができます。同期ダイアレクトで大規模な通常のバッチを扱う場合は、ドライバーが Native 一括挿入パスを使用できるよう、行を execute() の第 2 引数として渡してください。非同期で大量データを扱う場合は、ネイティブの AsyncClient.insert() メソッドを await してください。

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

pip install "clickhouse-connect[alembic]"

非同期ダイアレクトで移行を行う場合は、両方の extras をインストールしてください。

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 をリポジトリに含まれている 非同期 Alembic env.py の例 に置き換え、alembic.ini で sqlalchemy.url を設定します。

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

Alembic の移行関数は同期のままです。非同期環境では AsyncEngine を作成し、AsyncConnection を開いたうえで、同期の移行関数を await connection.run_sync(...) に渡します。オフライン移行ではエンジンを作成せず、context.configure(url=..., literal_binds=True, dialect_opts={"paramstyle": "named"}) を直接呼び出します。リポジトリに含まれている 非同期 Alembic env.py の例 は両方の処理に対応しており、Alembic 標準の sqlalchemy.url 設定から接続 URL を読み取ります。また、実例で使用している ClickHouse 用の Alembic フックとオプション (include_object、make_include_name(...)、clickhouse_writer、version_table など) もそのまま引き継いでいます。非同期移行の実行や破棄には engine.sync_engine を使用しないでください。

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 の範囲外です。