# pbatis-clickhouse **Repository Path**: pbatis/pbatis-clickhouse ## Basic Information - **Project Name**: pbatis-clickhouse - **Description**: No description available - **Primary Language**: Unknown - **License**: MIT - **Default Branch**: main - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-11-04 - **Last Updated**: 2026-11-04 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README PyBatis ClickHouse ================== .. image:: docs/pbatis-logo.png :alt: PyBatis Logo :align: center :width: 280px ClickHouse 方言插件 —— 让 `PyBatis `_ 直接连接 `ClickHouse 数据库 `_。 支持三种连接方式：native（TCP，推荐）、async native（异步 TCP）、http。 PyBatis 主页：https://gitee.com/pbatis/pbatis ---- 安装 ---- .. code-block:: bash pip install pbatis-clickhouse 依赖： .. code-block:: bash pip install pbatis # ORM 核心 pip install clickhouse-driver # native 驱动 pip install requests # http 驱动 pip install asynch # 异步 native 驱动 ---- 快速开始 -------- .. code-block:: python from pbatis import create_engine, Column, MetaData, func from pbatis_clickhouse import ( Table, MaterializedView, select, get_declarative_base, make_session, types, engines, ) # 1) 连接 # native（TCP，推荐） uri = 'clickhouse+native://localhost/default' # http # uri = 'clickhouse+http://localhost:8123/default' # 异步 native # uri = 'clickhouse+asynch://localhost:9000/default' engine = create_engine(uri) # 2) 声明式 metadata = MetaData() Base = get_declarative_base(metadata=metadata) class Rate(Base): __tablename__ = 'rates' day = Column(types.Date, primary_key=True) value = Column(types.Int32) __table_args__ = ( engines.Memory(), # 内存引擎（不落盘） ) # 3) 建表 Rate.__table__.create() # 4) 写入 from datetime import date, timedelta today = date.today() rates = [ {'day': today - timedelta(i), 'value': 200 - i} for i in range(100) ] session = make_session(engine) session.execute(Rate.__table__.insert(), rates) # 5) 查询 total = session.query(func.count(Rate.day)) \ .filter(Rate.day > today - timedelta(20)) \ .scalar() print(total) ---- 连接串格式 ---------- .. list-table:: :header-rows: 1 :widths: 25 18 28 29 * - URL 前缀 - 驱动 - 额外依赖 - 说明 * - ``clickhouse://`` - http - ``requests`` - 默认，走 HTTP 协议 * - ``clickhouse+http://`` - http - ``requests`` - 同上，显式写法 * - ``clickhouse+native://`` - native - ``clickhouse-driver`` - TCP 原生协议（推荐） * - ``clickhouse+asynch://`` - asynch - ``asynch`` - 异步 TCP 协议 连接串参数： .. code-block:: text clickhouse+native://user:password@host:9000/database?secure=True clickhouse+http://user:password@host:8123/database 常用参数： .. list-table:: :header-rows: 1 :widths: 25 75 * - 参数 - 说明 * - ``secure`` - 是否启用 TLS（native 驱动） * - ``client_name`` - 客户端标识，出现在 ClickHouse 的 ``system.query_log`` 里 * - ``verify`` - 是否校验服务端证书 ---- 类型系统 -------- ClickHouse 自有类型在 ``pbatis_clickhouse.types`` 中： .. code-block:: python from pbatis_clickhouse import types types.Int8 types.Int16 types.Int32 types.Int64 types.UInt8 types.UInt16 types.UInt32 types.UInt64 types.Float32 types.Float64 types.String types.FixedString types.Date types.DateTime types.DateTime64 types.Decimal types.UUID types.Array types.Nullable types.Map types.Tuple types.Struct types.Bool types.Enum8 types.Enum16 组合类型： .. code-block:: python types.Array(types.Int32) # Array(Int32) types.Nullable(types.String) # Nullable(String) types.Tuple(types.Int32, types.String) # Tuple(Int32, String) ---- 表引擎 -------- ``engines`` 模块提供 ClickHouse 表引擎声明： .. code-block:: python from pbatis_clickhouse import engines engines.MergeTree() # 默认合并树 engines.ReplacingMergeTree() # 替换合并树 engines.SummingMergeTree() # 汇总合并树 engines.AggregatingMergeTree() # 聚合合并树 engines.OrderedMergeTree() # 有序合并树 engines.Distributed() # 分布式表 engines.MaterializedView() # 物化视图 engines.View() # 普通视图 用法： .. code-block:: python class Events(Base): __tablename__ = 'events' id = Column(types.Int64, primary_key=True) ts = Column(types.DateTime) payload = Column(types.String) __table_args__ = ( engines.MergeTree( partition_by='toYYYYMM(ts)', order_by=('ts',), ), ) 带参数的引擎： .. code-block:: python engines.MergeTree( partition_by='toYYYYMM(ts)', # 分区键 order_by=('ts',), # 排序键 primary_key=('id',), # 主键 index_granularity=8192, # 索引粒度 ttl=('ts + INTERVAL 90 DAY',), # TTL ) ---- 物化视图 -------- .. code-block:: python from pbatis_clickhouse import MaterializedView class DailyAgg(MaterializedView): __tablename__ = 'daily_agg' day = Column(types.Date, primary_key=True) total = Column(types.Int64) __table_args__ = ( engines.MaterializedView( to='events', # 源表 engine='AggregatingMergeTree() ORDER BY day', populate='SELECT toDate(ts) AS day, count() AS total FROM events GROUP BY day', ), ) ---- 常见操作 -------- .. code-block:: python from pbatis import select, func, desc # 查询 session.execute(select(Rate).where(Rate.day > today)).fetchall() # 聚合 session.execute( select( Rate.day, func.sum(Rate.value).label('total'), ).group_by(Rate.day) ).fetchall() # 排序 + 分页 session.execute( select(Rate).order_by(desc(Rate.day)).limit(10) ).fetchall() # 批量写入 session.execute(Rate.__table__.insert(), rates) # executemany # 执行原生 SQL from pbatis import text session.execute(text('SELECT count() FROM rates')).scalar() # DDL from pbatis import DDL session.execute(DDL('ALTER TABLE rates DROP COLUMN IF EXISTS value')) ---- 异步使用 -------- .. code-block:: python from pbatis.ext.asyncio import create_async_engine from pbatis import text engine = create_async_engine('clickhouse+asynch://localhost:9000/default') async def main(): async with engine.begin() as conn: result = await conn.execute(text('SELECT 1')) print(result.scalar()) ---- 注意事项 -------- .. list-table:: :header-rows: 1 :widths: 28 72 * - 项目 - 说明 * - 主键 - ClickHouse 不强制主键，但 ``MergeTree`` 引擎建议指定 ``ORDER BY`` * - 事务 - ClickHouse 不支持常规事务，批量写入靠 ``INSERT`` 的批处理语义 * - UPDATE / DELETE - 是 ``ALTER TABLE ... UPDATE/DELETE`` 的异步变更，落地有延迟 * - 自增主键 - ClickHouse 无自增列，需要用雪花 ID 或 UUID（见 pbatis 的 ``pybatis.ext.snowflake``） * - 批量写入 - 单次写入建议 1 万行以上，或用 ``clickhouse-driver`` 的原生批量接口 ---- 相关项目 -------- - PyBatis 核心：https://gitee.com/pbatis/pbatis - ClickHouse 官方文档：https://clickhouse.com/docs ---- 版本与许可 ---------- 当前版本：1.0.0 协议：MIT（详见 ``LICENSE`` 文件）