sqlalchemy-cubrid

🌐 Community translation of README.md — English is canonical; this translation may lag behind the original. The translation-sync CI check flags drift, and maintainers open resync PRs.

适用于 CUBRID 数据库的 SQLAlchemy 2.0–2.1 方言 — 为 SQLAlchemy 与 CUBRID 特有类型提供 Python ORM、模式反射、Alembic 迁移和类型映射。

🇰🇷 한국어 · 🇺🇸 English · 🇨🇳 中文 · 🇮🇳 हिन्दी · 🇩🇪 Deutsch · 🇷🇺 Русский

---

状态：Production/Stable — 支持 CUBRID 10.2–11.4 和 SQLAlchemy 2.0–2.1 的稳定维护方言。每个 PR 均通过实时数据库集成 CI 验证。

为什么选择 sqlalchemy-cubrid？

CUBRID 是一款高性能开源关系型数据库，在韩国公共部门和企业应用中被广泛采用。 此前一直没有一个积极维护、支持现代 2.0–2.1 API 的 SQLAlchemy 方言。

sqlalchemy-cubrid 填补了这一空白：

• 完整的 SQLAlchemy 2.0–2.1 方言，支持语句缓存和 PEP 561 类型标注
• 庞大的离线测试套件 —— 无需数据库即可运行；CI 在 offline-tests 任务中强制执行至少 95% 的行覆盖率（--cov-fail-under=95）
• 并发压力测试 —— 已在真实 CUBRID 上验证 QueuePool 同步线程和 asyncio.gather 工作负载
• 面向 SQLAlchemy 2.1 的兼容垫片 —— 私有 API 访问被封装在 _compat.py 中（在完成 SA 2.1 全面验证前仍固定为 <2.2）
• 在 Python 3.11 -- 3.15 上测试 4 个 CUBRID 版本（10.2、11.0、11.2、11.4）
• CUBRID 特有的 DML 构造：ON DUPLICATE KEY UPDATE、MERGE、REPLACE INTO
• 开箱即用的 Alembic 迁移支持
• 三种驱动选项 —— C 扩展（cubrid://）、纯 Python（cubrid+pycubrid://）或异步纯 Python（cubrid+aiopycubrid://）

架构

flowchart TD
    app["Application"] --> sa["SQLAlchemy Core/ORM"]
    sa --> dialect["CubridDialect"]
    dialect --> pycubrid["pycubrid driver"]
    dialect --> cext["CUBRIDdb driver"]
    dialect --> aio["pycubrid.aio async driver"]
    pycubrid --> server["CUBRID Server"]
    cext --> server
    aio --> server

Broker、CAS 与 CCI 的路径请参见完整通信架构（英文）。

flowchart TD
    expr["SQL Expression"] --> compiler["CubridSQLCompiler"] --> sql["SQL String"]

环境要求

• Python 3.11 或更高版本
• SQLAlchemy 2.0 – 2.1
• CUBRID-Python（C 扩展）或 pycubrid（纯 Python）

安装

pip install sqlalchemy-cubrid

使用纯 Python 驱动（同步和异步）：

pip install "sqlalchemy-cubrid[pycubrid]"

[pycubrid] extra 同时支持 cubrid+pycubrid:// 和 cubrid+aiopycubrid://， 并包含 SQLAlchemy 的 asyncio extra（greenlet）。如果没有兼容的 wheel， 安装 greenlet 可能需要构建工具。

包含 Alembic 支持：

pip install "sqlalchemy-cubrid[alembic]"

快速开始

Core（连接级别）

from sqlalchemy import create_engine, text

engine = create_engine("cubrid://dba:password@localhost:33000/demodb")

with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    print(result.scalar())

ORM（会话级别）

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)


engine = create_engine("cubrid://dba:password@localhost:33000/demodb")
Base.metadata.create_all(engine)

with Session(engine) as session:
    user = User(name="Alice", email="[email protected]")
    session.add(user)
    session.commit()

Async

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine


async def main() -> None:
    # Create the engine inside the running event loop, and always dispose of it.
    engine = create_async_engine("cubrid+aiopycubrid://dba:password@localhost:33000/demodb")
    try:
        async with AsyncSession(engine) as session:
            result = await session.execute(text("SELECT 1 + 1"))
            print(result.scalar())
    finally:
        await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

功能特性

• 为 SQLAlchemy 标准类型和 CUBRID 特有类型提供类型映射 —— 数值、字符串、日期/时间、位、LOB、集合和 JSON 类型
• SQL 编译 -- SELECT、JOIN、CAST、LIMIT/OFFSET、子查询、CTE、窗口函数
• DML 扩展 -- ON DUPLICATE KEY UPDATE、MERGE、REPLACE INTO、FOR UPDATE、TRUNCATE
• DDL 支持 -- COMMENT、IF NOT EXISTS / IF EXISTS、AUTO_INCREMENT
• 模式反射 -- 表、视图、列、主键、外键、索引、唯一约束、注释
• 通过 CubridImpl 提供 Alembic 迁移（加载方言时自动注册）
• 支持 CUBRID 的三种 MVCC 隔离级别 — READ COMMITTED（默认）、REPEATABLE READ、SERIALIZABLE
• Async 支持 —— 通过 pycubrid.aio 使用 create_async_engine("cubrid+aiopycubrid://...")

已知限制

• 不支持 RETURNING —— 不支持 INSERT/UPDATE/DELETE ... RETURNING；请改用 cursor.lastrowid 或 LAST_INSERT_ID()
• 不支持序列 —— CUBRID 仅使用 AUTO_INCREMENT
• 不支持多 schema —— 每个数据库只有单一 schema
• 未提交的 DDL 会持有模式锁 —— CUBRID 的 DDL 是事务性的（ROLLBACK 可撤销；只有方言已关闭的客户端自动提交才会提前提交它），因此默认情况下整个 Alembic 升级是一个事务（transactional_ddl = True），在提交前会一直锁住涉及的表；耗时较长或针对大表的迁移请使用 transaction_per_migration=True
• 仅支持 SQLAlchemy 2.0–2.1 —— 由于内部 API 依赖，版本固定为 <2.2（详情）
• Async 需要 pycubrid >= 1.8.0,<2.0 —— cubrid+aiopycubrid:// 驱动需要本项目当前支持的 async 能力 pycubrid 包线

文档

指南	描述
连接	连接字符串、URL 格式、驱动设置、连接池调优
类型映射	完整类型映射、CUBRID 特有类型、集合类型
DML 扩展	ON DUPLICATE KEY UPDATE、MERGE、REPLACE INTO、查询跟踪
隔离级别	CUBRID 的三种 MVCC 隔离级别、配置
Alembic 迁移	设置、配置、限制、批量解决方案
功能支持	与 MySQL、PostgreSQL、SQLite 的对比
ORM 指南	实用 ORM 示例、关系、查询
开发指南	开发设置、测试、Docker、覆盖率、CI/CD
驱动兼容性	CUBRID-Python 驱动版本和已知问题
故障排除	常见问题、错误解决方案、调试技巧
异步连接	使用 cubrid+aiopycubrid:// 设置异步引擎

兼容性矩阵

组件	支持的版本
Python	3.11、3.12、3.13、3.14、3.15
CUBRID	10.2、11.0、11.2、11.4
SQLAlchemy	2.0–2.1
Alembic	>=1.7.2
pycubrid（sync）	>=1.8.0,<2.0
pycubrid（async）	>=1.8.0,<2.0

FAQ

如何通过 SQLAlchemy 连接到 CUBRID？

from sqlalchemy import create_engine
engine = create_engine("cubrid://dba:password@localhost:33000/demodb")

对于纯 Python 驱动（无需 CUBRID 原生库）：create_engine("cubrid+pycubrid://dba@localhost:33000/demodb")。[pycubrid] extra 中的 greenlet 在没有兼容 wheel 时可能需要构建工具。

sqlalchemy-cubrid 支持 SQLAlchemy 2.0–2.1 吗？

支持。sqlalchemy-cubrid 是为 SQLAlchemy 2.0–2.1 构建的，并支持 2.0 风格 API，包括 Session.execute()、带类型标注的 Mapped[] 列以及语句缓存。

sqlalchemy-cubrid 支持 Alembic 迁移吗？

支持。请通过 pip install "sqlalchemy-cubrid[alembic]" 安装。CUBRID 迁移实现会在方言加载时自动注册，因此默认的 env.py 在同步 URL 下无需修改即可使用；cubrid+aiopycubrid:// 请使用 Alembic 的异步模板（alembic init -t async）。CUBRID 的 DDL 是事务性的，因此默认情况下失败的 alembic upgrade 会整体回滚（包括版本号更新）；耗时较长或针对大表的迁移可设置 transaction_per_migration=True，在每个修订后提交。

支持哪些 Python 版本？

支持 Python 3.11、3.12、3.13、3.14 和 3.15（标准 GIL 构建）。

CUBRID 支持 RETURNING 子句吗？

不支持。CUBRID 不支持 INSERT ... RETURNING 或 UPDATE ... RETURNING。请改用 cursor.lastrowid 或 SELECT LAST_INSERT_ID()。

如何在 CUBRID 中使用 ON DUPLICATE KEY UPDATE？

from sqlalchemy_cubrid import insert
stmt = insert(users).values(name="Alice").on_duplicate_key_update(name="Alice Updated")

cubrid:// 和 cubrid+pycubrid:// 有什么区别？

cubrid:// 使用需要编译的 C 扩展驱动（CUBRIDdb）。cubrid+pycubrid:// 使用无需 CUBRID 原生库的纯 Python 驱动。[pycubrid] extra 包含 greenlet，没有兼容 wheel 时可能需要构建工具。cubrid+aiopycubrid:// 使用纯 Python 驱动的异步变体，可与 create_async_engine 和 AsyncSession 一起使用。

sqlalchemy-cubrid 支持 async 吗？

支持。请配合 pycubrid 异步驱动使用 create_async_engine("cubrid+aiopycubrid://...")。需要 pycubrid>=1.8.0,<2.0。两个 pycubrid 方言现在都会在 pool_pre_ping 中使用原生 Connection.ping(False) / AsyncConnection.ping(False)，所有 Core 和 ORM 功能都可在 AsyncSession 中使用。

相关项目

• pycubrid — 适用于 CUBRID 的纯 Python DB-API 2.0 驱动
• cubrid-cookbook-python — 面向 CUBRID 的生产级 Python 示例

路线图

项目方向和后续里程碑请参见 ROADMAP.md。

生态系统全貌请参见 CUBRID Labs Ecosystem Roadmap。

贡献

贡献指南请参阅 CONTRIBUTING.md，开发环境设置请参阅 docs/DEVELOPMENT.md。

安全

请通过 GitHub 私密漏洞报告（private vulnerability reporting）报告漏洞（备用方式为电子邮件）-- 详见 SECURITY.md。请勿就安全问题创建公开 issue。

许可证

MIT -- 参见 LICENSE。