pycubrid

🌐 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 数据库的纯 Python DB-API 2.0 驱动 — 无需 C 扩展、无需编译，实现了 PEP 249（DB-API 2.0）接口。

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

---

状态：Stable（1.x）。 公共 API 遵循语义化版本控制：次版本发布添加向后兼容的功能，补丁发布仅修复 bug；破坏性变更保留到下一个主版本（2.0+），并由针对 api-baseline.json 的自动化 compat-check CI 作业进行门控。开发持续进行 — 完整契约请参见 RELEASE_POLICY.md。

为什么选择 pycubrid？

CUBRID 是一款高性能开源关系型数据库，在韩国公共部门和企业应用中被广泛采用。 现有的 C 扩展驱动（CUBRIDdb）存在构建依赖和平台兼容性问题。

pycubrid 解决了这些问题：

• 纯 Python 实现 — 无需 C 构建依赖，只需 pip install
• 实现 PEP 249（DB-API 2.0） — 标准异常层级、类型对象和游标接口
• 离线回归测试和完整覆盖率运行的 95% 下限 — 参见实测覆盖率和 CI 执行策略；常规 PR 冒烟测试不测量覆盖率
• 同步/异步连接均支持 TLS/SSL — 在 connect() 和 pycubrid.aio.connect() 中可选 ssl=True（已验证上下文，最低 TLS 1.2）或自定义 ssl.SSLContext。
• 原生 asyncio 支持 — 通过 pycubrid.aio 提供 async/await API，适用于高并发应用
• PEP 561 类型化包 — py.typed 标记支持现代 IDE 和静态分析工具
• 直接实现 CUBRID CAS 协议 — 无需额外中间件
• 支持 LOB（CLOB/BLOB） — 可处理大文本和二进制数据

环境要求

• Python 3.11 或更高版本
• CUBRID 数据库服务器 10.2+

安装

pip install pycubrid

快速开始

基本连接

import pycubrid

conn = pycubrid.connect(
    host="localhost",
    port=33000,
    database="testdb",
    user="dba",
    password="",
)

cur = conn.cursor()
cur.execute("SELECT 1 + 1")
print(cur.fetchone())  # (2,)

cur.close()
conn.close()

上下文管理器

import pycubrid

with pycubrid.connect(host="localhost", port=33000, database="testdb", user="dba") as conn:
    with conn.cursor() as cur:
        cur.execute("CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100))")
        cur.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
        conn.commit()

        cur.execute("SELECT * FROM users")
        for row in cur:
            print(row)

Async

import asyncio
import pycubrid.aio

async def main():
    conn = await pycubrid.aio.connect(
        host="localhost", port=33000, database="testdb", user="dba"
    )
    cur = conn.cursor()
    await cur.execute("SELECT 1 + 1")
    print(await cur.fetchone())  # (2,)
    await cur.close()
    await conn.close()

asyncio.run(main())

实用示例

在测试数据库中，在同一个 Python 会话中依次运行准备代码、三个示例和清理代码。请调整连接设置以匹配服务器。专用表不能已存在：准备代码会报错，避免重用或删除现有表。手动提交模式和初始 DDL 提交确保下面的回滚只影响 DML。

准备

import pycubrid

conn = pycubrid.connect(host="localhost", port=33000, database="testdb", user="dba")
conn.autocommit = False
cur = conn.cursor()
cur.execute("CREATE TABLE pycubrid_quickstart_327 (id INT PRIMARY KEY, name VARCHAR(100))")
conn.commit()

基本 CRUD

cur.execute("INSERT INTO pycubrid_quickstart_327 (id, name) VALUES (?, ?)", (1, "Alice"))
cur.execute("SELECT name FROM pycubrid_quickstart_327 WHERE id = ?", (1,))
print(cur.fetchone())  # ('Alice',)
cur.execute("UPDATE pycubrid_quickstart_327 SET name = ? WHERE id = ?", ("Alicia", 1))
cur.execute("SELECT name FROM pycubrid_quickstart_327 WHERE id = ?", (1,))
print(cur.fetchone())  # ('Alicia',)
cur.execute("DELETE FROM pycubrid_quickstart_327 WHERE id = ?", (1,))
conn.commit()

事务：提交与回滚

cur.execute("INSERT INTO pycubrid_quickstart_327 (id, name) VALUES (?, ?)", (2, "Bob"))
conn.commit()
cur.execute("UPDATE pycubrid_quickstart_327 SET name = ? WHERE id = ?", ("Temporary", 2))
conn.rollback()
cur.execute("SELECT name FROM pycubrid_quickstart_327 WHERE id = ?", (2,))
print(cur.fetchone())  # ('Bob',)
conn.rollback()

错误处理

from pycubrid.exceptions import ProgrammingError

try:
    cur.execute("SELECT missing_column FROM pycubrid_quickstart_327 WHERE id = ?", (2,))
except ProgrammingError as exc:
    conn.rollback()
    print(f"Query failed: {exc}")

清理

准备成功后，即使示例失败也应运行清理代码；它只删除上面创建的表。更多示例见 EXAMPLES.md。

conn.rollback()
cur.execute("DROP TABLE pycubrid_quickstart_327")
conn.commit()
cur.close()
conn.close()

PEP 249 兼容性

属性	值
apilevel	"2.0"
threadsafety	1（连接不能在线程之间共享）
paramstyle	"qmark"（位置参数 ?）

• 完整标准异常层级：Warning、Error、InterfaceError、DatabaseError、OperationalError、IntegrityError、InternalError、ProgrammingError、NotSupportedError
• 标准类型对象：STRING、BINARY、NUMBER、DATETIME、ROWID
• 标准构造函数：Date()、Time()、Timestamp()、Binary()、DateFromTicks()、TimeFromTicks()、TimestampFromTicks()

功能特性

• 纯 Python — 无需 C 扩展、无需编译，在 Python 能运行的地方都能工作
• 完整 DB-API 2.0 — connect()、Cursor、fetchone/many/all、executemany、callproc
• 参数化查询 — cursor.execute(sql, params)，使用驱动端参数绑定（? 占位符在本地转义并插入 SQL）
• 批量操作 — executemany() 和 executemany_batch() 用于批量插入
• LOB 支持 — create_lob()，读写 CLOB 和 BLOB 列
• 模式自省 — get_schema_info() 用于表、列、索引和约束
• 自动提交控制 — 使用 connection.autocommit 属性管理事务
• 服务器版本检测 — connection.get_server_version() 返回版本字符串（例如 "11.2.0.0378"）
• 迭代器协议 — 可通过 for row in cursor 遍历游标结果
• 上下文管理器 — 连接和游标都支持 with 语句
• Async 支持 — pycubrid.aio.connect() 搭配 AsyncConnection 和 AsyncCursor，适用于 asyncio 事件循环

支持的 CUBRID 版本

该项目面向 CUBRID 10.x 和 11.x，并在 CI 中针对以下版本进行验证：

• 10.2
• 11.0
• 11.2
• 11.4

SQLAlchemy 集成

pycubrid 可作为 sqlalchemy-cubrid 的驱动使用——它是面向 CUBRID 的 SQLAlchemy 2.0 方言：

pip install "sqlalchemy-cubrid[pycubrid]"

from sqlalchemy import create_engine, text

engine = create_engine("cubrid+pycubrid://dba@localhost:33000/testdb")

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

与 sqlalchemy-cubrid 一起使用时，可通过 pycubrid 驱动访问 SQLAlchemy 的 ORM、Core、Alembic 迁移和模式反射等功能。

文档

指南	描述
连接	连接字符串、URL 格式、配置
类型映射	完整类型映射、CUBRID 特有类型、集合类型
API 参考	完整 API 文档 — 模块、类、函数
协议	CAS 线路协议参考
开发指南	开发环境设置、测试、Docker、覆盖率、CI/CD
示例	实用使用示例与代码
故障排除	连接错误、查询问题、LOB 处理、调试

兼容性

支持：Python 3.11–3.15 和 CUBRID 10.2、11.0、11.2、11.4。 普通 PR 使用一条 Ubuntu/Python 3.12 离线冒烟通道；高风险变更额外运行 Python 3.14/CUBRID 11.4。main 分支及每周变更验证还会运行 Python 3.11/CUBRID 10.2 实时测试，以及 Python 3.11 和 3.15 上的完整离线套件。完整的 5 × 4 实时矩阵会在显式触发 dispatch 时以及每次发布时运行。 参见 CI 执行策略 和 支持矩阵。

架构

graph TD
    app[Application]
    pycubrid[pycubrid Connection/Cursor]
    cas[CAS Protocol]
    server[CUBRID Server]

    app --> pycubrid
    pycubrid --> cas
    cas --> server

graph TD
    root[pycubrid/]
    init["__init__.py - Public API connect(), types, exceptions, __version__"]
    connection[connection.py - Connection class connect/commit/rollback/cursor/LOB]
    cursor[cursor.py - Cursor class execute/fetch/executemany/callproc/iterator]
    types[types.py - DB-API 2.0 type objects and constructors]
    exceptions[exceptions.py - PEP 249 exception hierarchy]
    constants[constants.py - CAS function codes, data types, protocol constants]
    protocol["protocol.py - CAS wire protocol packet classes (20 packet types)"]
    packet[packet.py - Low-level packet reader/writer]
    lob[lob.py - LOB support]
    typed[py.typed - PEP 561 marker]

    root --> init
    root --> connection
    root --> cursor
    root --> types
    root --> exceptions
    root --> constants
    root --> protocol
    root --> packet
    root --> lob
    root --> typed
    root --> aio
    aio["aio/ - AsyncConnection, AsyncCursor, async connect()"]

FAQ

如何使用 Python 连接到 CUBRID？

import pycubrid
conn = pycubrid.connect(host="localhost", port=33000, database="testdb", user="dba")

如何安装 pycubrid？

pip install pycubrid — 无需 C 扩展或构建工具。

pycubrid 使用什么参数风格？

问号风格（qmark）：cursor.execute("SELECT * FROM users WHERE id = ?", (1,))

pycubrid 可以与 SQLAlchemy 一起使用吗？

可以。安装 pip install "sqlalchemy-cubrid[pycubrid]"，并使用连接 URL cubrid+pycubrid://dba@localhost:33000/testdb。

支持哪些 Python 版本？

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

pycubrid 支持 LOB（CLOB/BLOB）吗？

支持。可以直接将字符串/字节插入 CLOB/BLOB 列。读取时，LOB 列会返回可通过游标访问的数据。

pycubrid 是线程安全的吗？

pycubrid 的 threadsafety = 1，表示连接不能在线程之间共享。请为每个线程创建单独的连接。

支持哪些 CUBRID 版本？

CUBRID 10.2、11.0、11.2 和 11.4 已在 CI 中测试。

pycubrid 支持 async/await 吗？

支持。使用 pycubrid.aio.connect() 获得原生 asyncio 支持。async 接口与 sync API 相似：await conn.ping(reconnect=...) 可执行与 sync Connection.ping() 相同的原生 CHECK_CAS 健康检查，create_lob() 仍然仅限同步接口，自动提交变更也使用 await conn.set_autocommit(...)，而不是属性 setter。

相关项目

• sqlalchemy-cubrid — 面向 CUBRID 的 SQLAlchemy 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。