SQLAlchemy 支持

ClickHouse Connect 内置了一个基于核心驱动构建的 clickhousedb SQLAlchemy 方言。它支持 SQLAlchemy 1.4.40 及更高版本 (包括 SQLAlchemy 2.x) ，重点关注 Core 查询、ClickHouse DDL、反射以及简单的 ORM 插入。通过包扩展安装 SQLAlchemy 依赖项：

pip install "clickhouse-connect[sqlalchemy]"

使用 SQLAlchemy 进行连接

使用 clickhousedb:// 或 clickhousedb+connect:// 这两种 URL 格式之一创建引擎：

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 设置、ClickHouse Connect 客户端选项 (例如 compression、query_limit 和超时设置) ，或 HTTP/TLS 选项 (例如 ca_cert) 。如有需要，可为 ClickHouse 设置添加 ch_ 前缀，以强制将其识别为服务器设置，例如 ch_http_max_field_name_size=99999。可用的客户端选项请参见连接参数和设置。

每个查询的设置

通过 SQLAlchemy 的执行选项传递 ClickHouse 设置。可以在引擎、连接或语句上设置这些参数。对于相同的键，语句上的值优先于连接或引擎上的值。

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

按查询设置读取格式

通过 SQLAlchemy 的执行选项 query_formats，可为引擎、连接或语句设置 ClickHouse 读取格式。语句级格式会优先应用，并覆盖匹配的连接级或引擎级键及通配符。

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 服务器端参数：

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

在此模式下，每个绑定值都必须具有与 ClickHouse 兼容的 SQLAlchemy 类型。受支持的 IN 列表会变为带类型的 ClickHouse Array 参数。如果编译器无法推导出兼容的类型，或无法安全地处理绑定值，就会引发 CompileError。

Core 查询

该方言支持 SQLAlchemy Core SELECT 查询，可使用 JOIN、过滤器、排序、LIMIT 和 OFFSET 以及 DISTINCT。

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)

JSON 子列

对于声明为或映射为 ClickHouse JSON 的列，请使用方括号逐段选择由存储支持的子列路径：

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()。每个分段都必须是非空字符串。向 .subcolumn() 传入 type_ 会将点分路径包装为 SQL CAST，并将该类型赋予 SQLAlchemy 表达式。未传入 type_ 时，.subcolumn("segment") 的行为与 ["segment"] 相同。未指定类型的路径具有 ClickHouse 的 Dynamic 类型。ClickHouse 不允许在 ORDER BY 或 GROUP BY 中直接使用 Dynamic 值。当在这些位置使用子列时，请传入 type_。对于静态类型代码，请从 clickhouse_connect.cc_sqlalchemy 导入 json_subcolumn。该辅助函数同样每次只接受一个分段，并保留 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 路径处理，反引号不会使点号成为字面量。启用 json_type_escape_dots_in_keys 后，键名中的字面点号应使用 ClickHouse 的 %2E 编码。对于名为 a.b 的键，应通过 payload["a%2Eb"] 而非 payload["a.b"] 访问。

ClickHouse 查询扩展

从 clickhouse_connect.cc_sqlalchemy 导入 select，即可向静态类型检查器公开带类型的 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 方法如下：

方法	SQL 特性
`.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 (...)`

例如，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 形式需要 SQLAlchemy 2.0.42 或更高版本，其中新增了 Values.cte()。

Materialized CTE

默认情况下，ClickHouse 会内联公共表表达式，因此被多次引用的 CTE 的主体会针对每次引用执行一次。向 .cte() 传入 materialized=True，即可生成 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})
)

仅当指定该关键字、enable_materialized_cte=1 且启用 analyzer 时，服务器才会 materialize CTE。如每个查询的设置所示，可在语句、连接或引擎上设置 enable_materialized_cte。在所有支持此功能的服务器上，analyzer 默认启用，因此显式设置 enable_analyzer=1 是一种防御性措施。enable_materialized_cte 是一项 Experimental ClickHouse 设置。使用 enable_materialized_cte=0 或 enable_analyzer=0 时，查询仍会成功执行并返回相同的行。ClickHouse 会静默忽略 MATERIALIZED 并重新内联 CTE，因此漏设该选项只会影响性能，不会报错。Materialized CTEs 需要 ClickHouse 26.3 或更高版本。旧版服务器会将该关键字视为语法错误而拒绝。对于使用标准 sqlalchemy.select 构建的语句，请改用模块级的 cte()。它将语句作为第一个 argument，其他行为与 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 方言下生效，因此与其他后端共享的语句在该方言下可原样编译。 ClickHouse 不支持递归 materialized CTE。若同时设置 recursive=True 和 materialized=True，SQLAlchemy helpers 将引发 ValueError。

DDL 与反射

ClickHouse Connect 提供 ClickHouse 数据类型、表引擎、字典结构、数据库 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。 MergeTree 键参数 (如 order_by、partition_by、primary_key、sample_by 和 ttl) 既接受 SQLAlchemy 列和 SQL 表达式，也接受普通字符串。

插入和基本 ORM 用法

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

Alembic 迁移

ClickHouse Connect 提供了适用于 ClickHouse schema 迁移的 Alembic 集成。使用以下命令安装：

pip install "clickhouse-connect[alembic]"

在 Alembic 的 env.py 中导入 clickhouse_connect.cc_sqlalchemy.alembic，以注册方言集成。自动生成支持常见的表结构变更，包括创建和删除表、添加/修改/删除列、默认值以及注释。表和列的重命名请使用手动操作。在应用每个生成的迁移之前，都应先进行审查。 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 端的工作，但 commit 和 rollback 在服务器端都是空操作。
该方言未实现 UPDATE、两阶段事务、序列、RETURNING 以及高级隔离级别。需要执行服务器端变更时，请显式使用 ClickHouse SQL。
Column(..., primary_key=True) 提供的是 SQLAlchemy 的对象标识。它不会创建服务器端的唯一性约束。请通过表引擎定义排序和可选的主键表达式。
传统的外键、唯一约束以及标准索引元数据不可用，因为 ClickHouse 不会强制执行这些约束。
ORM 关系管理、工作单元更新、级联，以及立即或延迟的关系加载，不属于受支持的 ORM 范围。

首页

JavaScript

Python

Go

Java

C#

Rust

C++

C

第三方客户端

使用 SQLAlchemy 进行连接

每个查询的设置

按查询设置读取格式

服务器端参数

Core 查询

JSON 子列

ClickHouse 查询扩展

Materialized CTE

DDL 与反射

插入和基本 ORM 用法

Alembic 迁移

范围和限制

​使用 SQLAlchemy 进行连接

​每个查询的设置

​按查询设置读取格式

​服务器端参数

​Core 查询

​JSON 子列

​ClickHouse 查询扩展

​Materialized CTE

​DDL 与反射

​插入和基本 ORM 用法

​Alembic 迁移

​范围和限制

使用 SQLAlchemy 进行连接

每个查询的设置

按查询设置读取格式

服务器端参数

Core 查询

JSON 子列

ClickHouse 查询扩展

Materialized CTE

DDL 与反射

插入和基本 ORM 用法

Alembic 迁移

范围和限制