> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-parallel-read-in-order-multi-part.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

> ClickHouse SQLAlchemy 和 Alembic 支持

# SQLAlchemy 支持

ClickHouse Connect 内置了一个基于核心驱动构建的 `clickhousedb` SQLAlchemy 方言。同步方言支持 SQLAlchemy 1.4.40 及更高版本 (包括 SQLAlchemy 2.x) ，重点关注 Core 查询、ClickHouse DDL、反射以及简单的 ORM 插入。异步方言要求 SQLAlchemy 2.0.44 或更高版本。

通过包扩展安装 SQLAlchemy 依赖项：

```bash theme={null}
pip install "clickhouse-connect[sqlalchemy]"
```

<h2 id="sqlalchemy-connect">
  使用 SQLAlchemy 进行连接
</h2>

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

```python theme={null}
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)
```

<h3 id="sqlalchemy-session-ids">
  ClickHouse 会话 ID
</h3>

默认情况下，无论使用同步还是异步方言，每个池化连接都会生成一个独立的 ClickHouse 会话 ID。只要该连接的请求到达同一个 ClickHouse 服务器进程，通过 `SET` 修改的设置以及临时表就会在该连接上持续保留。命名会话状态和同一会话的重叠检查仅在进程内生效。在同一个服务器进程上，针对相同用户和会话 ID 的重叠请求会被立即拒绝并返回服务器错误码 373，而不会进入队列等待。如果您配置了固定的 `session_id`，请使用 `pool_size=1, max_overflow=0`，或在请求到达 ClickHouse 之前将访问串行化。在 ClickHouse Cloud 或其他采用负载均衡的部署中，具有相同会话 ID 的请求可能会被路由到不同的服务器，因此请勿将固定的 `session_id` 用作分布式状态或分布式互斥锁。

<h3 id="sqlalchemy-async-connections">
  异步连接
</h3>

异步方言要求 SQLAlchemy 2.0.44 或更高版本，并使用 ClickHouse Connect 原生的 `AsyncClient`。请先安装所需的依赖项，然后使用 `clickhousedb+async://` URL 创建异步引擎：

```bash theme={null}
pip install "clickhouse-connect[sqlalchemy-async]"
```

```python theme={null}
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` 的流式方法。在对应的 SQLAlchemy 连接被签出期间，可通过 `driver_connection` 访问底层原生客户端：

```python theme={null}
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 连接器默认限制为总共一个连接、每个主机一个连接。可在 URL 或 `connect_args` 中设置 `connector_limit`、`connector_limit_per_host` 或 `keepalive_timeout` 来覆盖这些传输设置。启用 `pool_pre_ping=True` 后，SQLAlchemy 会在签出池化连接时使用 `SELECT 1` 检查被复用的连接。

目前，异步 SQLAlchemy 的 executemany 插入会为每个参数集单独发送一个 HTTP 请求，而不会使用驱动程序的 Native 批量插入协议。此方式仅适用于小批量数据。对于大批量数据，请使用上文介绍的由连接池管理的 `driver_connection` 访问模式，并在将 SQLAlchemy 连接归还到连接池之前 await `client.insert()`。由于异步 executemany 使用查询参数绑定，不带时区的 `datetime` 值遵循 `naive_datetime_binding`，而非同步 Native executemany 所使用的 `naive_datetime_insert` 设置。带类型的 SQLAlchemy `DateTime64` 绑定无论使用客户端参数还是服务器端参数，均会保留小数秒。对于传递给 `exec_driver_sql()` 的无类型 `%s` 或 `%(name)s` 参数，不带时区的 `datetime` 值仍沿用默认的整秒格式。如需明确的时区行为，请使用带时区信息的值。如需 Native 批量语义，请使用 `client.insert()`。

请在使用异步引擎的事件循环中创建并释放该引擎。在关闭时，以及在其他事件循环中使用该引擎之前，请先归还所有已签出的连接，然后 await `engine.dispose()`。如果引擎所属的事件循环已经关闭，请在复用前于当前循环中 await `engine.dispose()`。如果清理工作在所属循环关闭之后才开始，aiohttp 仍可能报告存在未关闭的传输，因此请尽可能在转移之前完成释放。在事件循环之间迁移池化异步引擎时，`pool_pre_ping=True` 不能替代释放操作。若要在多个事件循环之间共享同一个引擎，且不保留绑定到特定循环的连接，请配置 `poolclass=NullPool`。如果执行释放时仍有连接处于签出状态，方言会在该连接被归还或被垃圾回收时将其关闭。请勿在同步代码中调用 `engine.sync_engine.dispose()`，因为 SQLAlchemy 在此无法 await 异步连接的清理，可能只会记录错误日志，而不会关闭池化的传输。

URL 查询参数可以包含 ClickHouse 设置、ClickHouse Connect 客户端选项 (例如 `compression`、`query_limit` 和各类超时) ，或 HTTP/TLS 选项 (例如 `ca_cert`) 。必要时，可为 ClickHouse 设置添加 `ch_` 前缀，强制将其视为服务器设置，例如 `ch_http_max_field_name_size=99999`。

有关可用的客户端选项，请参阅[连接参数和设置](/zh/integrations/language-clients/python/driver-api#connection-arguments)。

DDL、元数据检查等同步 SQLAlchemy 辅助操作需通过 `AsyncConnection.run_sync()` 运行：

```python theme={null}
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()
        )
```

<h3 id="sqlalchemy-per-query-settings">
  每个查询的设置
</h3>

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

```python theme={null}
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()
```

<h3 id="sqlalchemy-per-query-read-formats">
  按查询设置读取格式
</h3>

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

```python theme={null}
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()
```

<h3 id="sqlalchemy-error-handling">
  错误处理
</h3>

驱动通过 SQLAlchemy 连接引发的错误，使用的是从 `clickhouse_connect.dbapi` 导出的 DB-API 类。这些类与 `clickhouse_connect.driver.exceptions` 中的对应类是同一个类对象，因此 SQLAlchemy 会将其包装为相应的 `sqlalchemy.exc.DBAPIError` 子类。`StreamFailureError` 属于 `OperationalError`，会被包装为 `sqlalchemy.exc.OperationalError`。

如果调用方的取消操作可能会中断显式调用的 `AsyncConnection.invalidate()`，请在自行管理的任务中执行失效操作，并等待该任务完成后再传播取消。这样 SQLAlchemy 便能完成连接记录的簿记工作：

```python theme={null}
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()` 以完成清理，之后再使用或关闭该连接。

<h3 id="sqlalchemy-server-side-parameters">
  服务器端参数
</h3>

SQLAlchemy 通常会在客户端渲染参数。创建引擎时，可选择启用 ClickHouse 服务器端参数：

```python theme={null}
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 名称。以 `$` 开头和结尾的名称会被拒绝，因为核心驱动程序将其保留用于原始二进制查询参数。

<h2 id="sqlalchemy-core-queries">
  Core 查询
</h2>

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

SQLAlchemy `union()`、`intersect()` 和 `except_()` 会编译为 ClickHouse `UNION DISTINCT`、`INTERSECT DISTINCT` 和 `EXCEPT DISTINCT`。对应的 `union_all()`、`intersect_all()` 和 `except_all()` 会编译为相应的 `ALL` 运算符。此显式映射可保留 SQLAlchemy 的重复项语义，而不受 ClickHouse 集合操作默认设置的影响。

```python theme={null}
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` 子句：

```python theme={null}
from sqlalchemy import delete

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

<h3 id="sqlalchemy-literal-rendering">
  字面量渲染
</h3>

当 SQLAlchemy 通过 `literal_binds` 或 `literal_execute` 内联绑定值时，方言会针对通用 String 类型和 ClickHouse 类型采用 ClickHouse 的引用规则。这同样适用于 `TypeDecorator` 包装器以及 `with_variant()` 选择。即使其他绑定参数保持不变，String 值中的百分号和反斜杠仍会保留。

对于使用 ClickHouse `DateTime64` SQLAlchemy 类型的 Python `datetime` 值，其微秒部分会在客户端参数和内联字面量中得到保留，包括 Nullable 值以及嵌套在数组和元组中的值。ClickHouse 会按所声明的精度进行处理。Python `datetime` 最多提供六位小数。普通 `DateTime` 值仍按整秒格式输出。对于 `text()` 语句，请通过 `bindparam("ts", type_=DateTime64(6))` 显式指定类型，以保留小数秒。

SQLAlchemy 列类型必须与服务器 schema 保持一致。若在服务器端类型为 `DateTime` 的列上声明 `DateTime64`，渲染结果会带有小数秒，并可能在 insert 时以及 `IN` 比较中引发转换出错。

在 SQLAlchemy 2.x 中，若通用 `sqlalchemy.ARRAY` 类型包含 ClickHouse `Tuple` 元素，其内联字面量需要设置 `dimensions=1` (嵌套数组则需设置相应更高的维数) ，以便 SQLAlchemy 将每个元组视为单个元素。SQLAlchemy 1.4 不支持通用 `ARRAY` 类型的内联字面量。

如果重复使用某个命名的 datetime 参数，则该参数每次出现时都需要兼容的 `DateTime64` 绑定类型，才能保留小数部分。只要有一处未指定类型或类型冲突，就会保持整秒格式。请在每个 `bindparam` 上设置 `type_=DateTime64(6)`，或改用具有相应类型的不同参数名。

<h3 id="sqlalchemy-json-type-hints">
  JSON type hints
</h3>

使用 `typed_paths` 映射声明带类型的 JSON 路径。路径类型可以是 ClickHouse SQLAlchemy 类型类、已配置的类型实例，或 ClickHouse 类型名称字符串。类型名称字符串支持没有 SQLAlchemy 构造函数的类型 (例如 `Dynamic`) ，也同样适用于复杂的已配置类型表达式，并能在命名 `Tuple` 中保留字段名称。

类型名称字符串中可以包含已配置的嵌套 JSON 类型，例如 ``Array(JSON(`child` UInt32))``。在这些字符串中，可识别的 ClickHouse 类型名称不区分大小写，输出时会统一采用其规范大小写形式。字符串必须且只能包含一个完整的类型表达式，尾随文本以及格式错误的嵌套 JSON 参数都会被拒绝。

空的 `Tuple()` 不能用作 JSON 类型化路径，因为 ClickHouse 无法通过 JSON 列的 Native 格式对其进行序列化。核心驱动支持在查询列和插入列的任意位置使用 `Tuple()`，包括嵌套在位置元组或命名元组中、位于 `Array` 内，以及在服务端启用的情况下以 `Nullable(Tuple())` 的形式使用。

```python theme={null}
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 中。普通的 skip 路径会被去重。由于 ClickHouse 使用 RE2 语法，Python 不会校验正则表达式字符串。重复的正则表达式会被保留。

普通 skip 路径的名称不能正好是 `REGEXP`，因为 ClickHouse 将该标记保留给 `SKIP REGEXP` 使用。而 `REGEXP_foo` 这类名称仍然有效。在原始 JSON 类型字符串中，普通 `SKIP` 的操作数必须是一个 ClickHouse 标识符，或以点号分隔的复合标识符。未加引号的复合标识符不能以 `REGEXP` 开头；当第一个组成部分本身就是路径数据时，请为其加上引号。`SKIP REGEXP` 必须带有一个单引号字符串字面量。当标识符各部分包含空格或标点符号时，请使用反引号或双引号引用。原始 JSON 类型提示支持 `Variant(...)`；独立的 `Variant` 没有公开的 SQLAlchemy 构造函数。`Variant` 成员会按照 ClickHouse 所用的同一套规范名称进行排序和去重。

构造函数对参数的排序与 ClickHouse 返回的规范形式一致。反射得到的类型、SQLAlchemy 类型副本以及 Alembic 自动生成均会保留该配置。

<h3 id="sqlalchemy-json-subcolumns">
  JSON 子列
</h3>

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

```python theme={null}
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 结果类型：

```python theme={null}
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"]` 访问。

<h3 id="sqlalchemy-query-extensions">
  ClickHouse 查询扩展
</h3>

从 `clickhouse_connect.cc_sqlalchemy` 导入 `select`，即可向静态类型检查器公开带类型的 ClickHouse 方法。标准的 `sqlalchemy.select` 在运行时也提供这些方法。

```python theme={null}
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 (...)` |

SQLAlchemy's `Select.with_hint()` 是表提示 API。ClickHouse 方言不会渲染表提示。适用的通配符提示或 `clickhousedb` 提示会发出 `SAWarning`，并保持生成的 SQL 不变。对于这些 ClickHouse 子句，请使用 `final()`、`sample()`、`prewhere()` 或 `limit_by()`。

`Select.with_statement_hint()` 是原始尾部指令 API。它会将提供的文本附加到 `SELECT` 末尾，而不进行 ClickHouse 特有的验证。它仍可用于受信任的静态 SQL，例如 `SETTINGS max_threads=1`：

```python theme={null}
stmt = select(events.c.id).with_statement_hint("SETTINGS max_threads=1")
```

对于 ClickHouse 设置，建议优先使用执行选项，以便驱动程序将设置与 SQL 文本分开处理：

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

例如，ClickHouse `GLOBAL ANY LEFT JOIN` 可以链式调用，无需嵌套自定义 `FromClause`：

```python theme={null}
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` 构造：

```python theme={null}
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()`。

<h3 id="sqlalchemy-materialized-ctes">
  Materialized CTE
</h3>

默认情况下，ClickHouse 会内联公共表表达式，因此被多次引用的 CTE 的主体会针对每次引用执行一次。向 `.cte()` 传入 `materialized=True`，即可生成 `WITH <name> AS MATERIALIZED (...)`，使主体只计算一次：

```python theme={null}
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。如[每个查询的设置](#sqlalchemy-per-query-settings)所示，可在语句、连接 或 引擎 上设置 `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()` 一致：

```python theme={null}
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`。

<h2 id="sqlalchemy-ddl-reflection">
  DDL 与反射
</h2>

ClickHouse Connect 提供 ClickHouse 数据类型、表引擎、字典结构、数据库 DDL 和表反射功能。

独立的 `Variant` 列通过 SQLAlchemy 内部类型进行反射，Alembic 自动生成会保留其规范的原始类型名称，不会反复产生类型变更。`Geometry` 和 `MultiPoint` 列则反射为公开的 SQLAlchemy 类型。

```python theme={null}
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` 子句中的 String 值使用 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`。

```python theme={null}
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 列、映射的列属性，或由上述值组成的非空列表或元组。列表和元组中的字符串项会作为标识符加引号。单个标量字符串则直接作为原始 SQL 使用，例如 `"delta"` 或 `"(delta, n_tx)"`。服务器要求这些列以标识符形式指定。省略 `columns` 时，由 ClickHouse 自行选择要求和的列。反射和 Alembic 自动生成会保留显式指定的列列表。

<h2 id="sqlalchemy-inserts">
  插入和基本 ORM 用法
</h2>

支持 Core 插入以及简单的 ORM 模型。对于同步方言，在兼容的批量数据路径中优先使用 Core executemany 插入。对于异步批量插入，请使用[异步连接](#sqlalchemy-async-connections)中介绍的原生 `AsyncClient.insert()` 路径。

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

对于同步方言，由 SQLAlchemy 编译器生成的普通 Core `executemany` 插入会通过一次 Native 批量插入完成。异步 executemany 则会为每个参数集各发送一个请求，详见 [异步连接](#sqlalchemy-async-connections)。原始 SQL，以及包含表达式或其他无法安全路由的语义的插入，会保留原始 SQL，并针对每个参数集各执行一次。如果后续某个参数集执行失败，之前的参数集已写入的行仍会保持已提交状态。

显式多行 `insert(events).values([...])` 语句支持字典形式的行、按表列顺序排列的元组，以及逐行指定的 SQL 表达式。Pandas `to_sql(method="multi")` 采用的就是这种形式。它能够插入这些行，但会返回 `0`，因为文本形式的 INSERT 语句通过 DB-API 游标报告的行数始终为 `0`。SQLAlchemy 根据第一行确定列列表。后续行中多出的字典键，以及超出所选列列表范围的元组值，都会被忽略。如果后续某行缺少所选列的值，编译将会失败。请确保每一行都包含相同的列。

在 ClickHouse 26.4 及更高版本的默认 HTTP 表单限制下，`server_side_params=True` 仅适用于小型显式批次，即绑定值少于约 1000 个，同时还需为其他字段预留余量。可以通过服务器配置提高这一上限。对于同步方言下的大型普通批次，请将行作为 `execute()` 的第二个参数传入，以便驱动程序走其 Native 批量插入路径。对于异步批量数据，请 await 原生的 `AsyncClient.insert()` 方法。

```python theme={null}
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()
```

<h2 id="sqlalchemy-alembic">
  Alembic 迁移
</h2>

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

```bash theme={null}
pip install "clickhouse-connect[alembic]"
```

如需通过异步方言执行迁移，请同时安装这两个 extras：

```bash theme={null}
pip install "clickhouse-connect[alembic,sqlalchemy-async]"
```

创建一个异步 Alembic 项目，然后将其自动生成的环境替换为适配 ClickHouse 的示例：

```bash theme={null}
alembic init -t async alembic
```

生成的 `alembic.ini` 使用 `script_location = %(here)s/alembic`。如果迁移目录名为 `alembic`，请保留该设置；否则，请将其改为传递给 `alembic init` 的目录。将 `alembic/env.py` 替换为仓库中提供的[异步 Alembic `env.py` 示例](https://github.com/ClickHouse/clickhouse-connect/blob/main/examples/alembic_async/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` 示例](https://github.com/ClickHouse/clickhouse-connect/blob/main/examples/alembic_async/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 的创建与删除。
* 字典的创建、删除和重新加载。

ClickHouse 数据跳过索引不是 SQLAlchemy 索引。`Index`、`Column(index=True)`、`op.create_index` 和 `op.drop_index` 都会被拒绝，以避免生成不完整或不正确的 DDL。请使用 `op.add_clickhouse_index` 和 `op.drop_clickhouse_index`。

请参阅完整的 [Alembic 示例](https://github.com/ClickHouse/clickhouse-connect/blob/main/clickhouse_connect/cc_sqlalchemy/alembic/WORKED_EXAMPLE.md)。从 `clickhouse-sqlalchemy` 迁移的用户还应阅读[迁移指南](https://github.com/ClickHouse/clickhouse-connect/blob/main/clickhouse_connect/cc_sqlalchemy/MIGRATING_FROM_CLICKHOUSE_SQLALCHEMY.md)。

<h2 id="scope-and-limitations">
  范围和限制
</h2>

* ClickHouse 不通过此 HTTP 方言提供传统事务。`engine.begin()` 和 `Session.commit()` 用于组织 Python 端的工作，但 commit 和 rollback 在服务器端都是空操作。
* 该方言未实现 `UPDATE`、两阶段事务、序列、`RETURNING` 以及高级隔离级别。需要执行服务器端变更时，请显式使用 ClickHouse SQL。
* `Column(..., primary_key=True)` 提供的是 SQLAlchemy 的对象标识。它不会创建服务器端的唯一性约束。请通过表引擎定义排序和可选的主键表达式。
* 传统的外键、唯一约束以及标准索引元数据不可用，因为 ClickHouse 不会强制执行这些约束。
* ORM 关系管理、工作单元更新、级联，以及立即或延迟的关系加载，不属于受支持的 ORM 范围。