SQLAlchemy 支持 - ClickHouse Documentation

ClickHouse Connect 内置了一个基于核心驱动构建的 SQLAlchemy 方言 (clickhousedb) 。它面向 SQLAlchemy Core API，支持 SQLAlchemy 1.4.40+ 和 2.0.x。

通过 SQLAlchemy 连接

使用 clickhousedb:// 或 clickhousedb+connect:// URL 创建引擎。查询参数会映射为 ClickHouse 设置、客户端选项以及 HTTP/TLS 传输选项。

from sqlalchemy import create_engine, text

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

with engine.begin() as conn:
    rows = conn.execute(text("SELECT version()"))
    print(rows.scalar())

关于 URL/查询参数的说明：

ClickHouse 设置：作为查询参数传入 (例如，use_skip_indexes=0) 。
客户端选项：compression (compress 的别名) 、query_limit、超时等。
HTTP/TLS 选项：用于 HTTP 连接池和 TLS 的选项 (例如，ch_http_max_field_name_size=99999、ca_cert=certifi) 。

如需查看受支持选项的完整列表，请参阅下方各节中的连接参数和设置。这些选项也可以通过 SQLAlchemy DSN 传入。

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)

# 基本 SELECT
with engine.begin() as conn:
    rows = conn.execute(select(users.c.id, users.c.name).order_by(users.c.id).limit(10)).fetchall()

# JOIN（INNER/LEFT OUTER/FULL OUTER/CROSS）
with engine.begin() as conn:
    stmt = (
        select(users.c.name, orders.c.product)
        .select_from(users.join(orders, users.c.id == orders.c.user_id))
    )
    rows = conn.execute(stmt).fetchall()

支持必须带有 WHERE 子句的轻量级 DELETE：

from sqlalchemy import delete

with engine.begin() as conn:
    conn.execute(delete(users).where(users.c.name.like("%temp%")))

DDL 和反射

你可以使用提供的 DDL 辅助工具以及 type/引擎构造来创建数据库和表。支持表反射 (包括列类型和引擎) 。

import sqlalchemy as db
from sqlalchemy import MetaData
from clickhouse_connect.cc_sqlalchemy.ddl.custom import CreateDatabase, DropDatabase
from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import MergeTree
from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import UInt32, String, DateTime64

with engine.begin() as conn:
    # 数据库
    conn.execute(CreateDatabase("example_db", exists_ok=True))

    # 表
    metadata = MetaData(schema="example_db")
    table = 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"),
    )
    table.create(conn)

    # 反射
    reflected = db.Table("events", metadata, autoload_with=engine)
    assert reflected.engine is not None

反映出的列会在服务器上存在相关信息时，包含方言特有的属性，例如 clickhousedb_default_type、clickhousedb_codec_expression 和 clickhousedb_ttl_expression。

插入 (Core 和基本 ORM)

为了方便起见，既可以通过 SQLAlchemy Core 进行插入，也可以使用简单的 ORM 模型。

# 核心插入
with engine.begin() as conn:
    conn.execute(table.insert().values(id=1, user="joe"))

# 基本 ORM 插入
from sqlalchemy.orm import declarative_base, Session

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=1, name="Alice"))
    session.bulk_save_objects([User(id=2, name="Bob")])
    session.commit()

范围与限制

核心关注点：支持 SQLAlchemy Core 功能，例如带 JOIN (INNER、LEFT OUTER、FULL OUTER、CROSS) 的 SELECT，以及 WHERE、ORDER BY、LIMIT/OFFSET 和 DISTINCT。
仅支持带 WHERE 的 DELETE：该方言支持轻量级 DELETE，但要求显式指定 WHERE 子句，以避免误删整张表。如需清空表，请使用 TRUNCATE TABLE。
不支持 UPDATE：ClickHouse 针对追加写入进行了优化。该方言未实现 UPDATE。如果需要修改数据，请在上游完成转换后重新插入，或自行承担风险使用显式文本 SQL (例如 ALTER TABLE ... UPDATE) 。
DDL 和反射：支持创建数据库和表，反射会返回列类型和表引擎元数据。传统的主键/外键/索引元数据不存在，因为 ClickHouse 不会强制执行这些约束。
ORM 范围：为方便起见，支持声明式模型以及通过 Session.add(...)/bulk_save_objects(...) 进行插入。不支持高级 ORM 功能 (关系管理、工作单元更新、级联、预加载/懒加载语义) 。
主键语义：SQLAlchemy 中的 Column(..., primary_key=True) 仅用于对象标识，不会在 ClickHouse 中创建服务端约束。请通过表引擎定义 ORDER BY (以及可选的 PRIMARY KEY) (例如 MergeTree(order_by=...)) 。
事务和服务端功能：不支持两阶段事务、序列、RETURNING 和高级隔离级别。engine.begin() 提供了一个用于将语句分组的 Python 上下文管理器，但不会执行任何实际的事务控制 (commit/rollback 均为空操作) 。

​通过 SQLAlchemy 连接

​Core 查询

​DDL 和反射

​插入 (Core 和基本 ORM)

​范围与限制

通过 SQLAlchemy 连接

Core 查询

DDL 和反射

插入 (Core 和基本 ORM)

范围与限制