dm-pure：纯 Python 达梦 DM8 数据库驱动

零 native 依赖、100% Python 实现的达梦数据库驱动。支持达梦 DM8 常用功能，适合受限环境（无编译工具/无法加载 native 库的机器）。

特性

• ✅ 连接：DH 密钥交换 + DES-CFB 加密登录（内置纯 Python 加密，零依赖）
• ✅ 查询：SELECT / INSERT / UPDATE / DELETE，支持参数绑定（含 NULL）
• ✅ 类型：INT / BIGINT / DECIMAL / DATE / TIME / DATETIME / TIME_TZ / DATETIME_TZ / DATETIME2 / DATETIME2_TZ（含时区偏移）/ INTERVAL / FLOAT / 中文等
• ✅ 字符集：自动检测服务器编码（GB18030 / UTF-8 / EUC-KR），也可手动指定，中文读写不乱码
• ✅ 大结果集：分批 fetch（cmd=7）
• ✅ LOB：CLOB / BLOB 读写（含大 LOB 分段传输）
• ✅ 事务：begin / commit / rollback / 隔离级别 / with c.transaction(): 上下文管理器
• ✅ DSN：DMClient.from_dsn("dm://user:pwd@host:port/?encoding=utf8")
• ✅ 事务：begin / commit / rollback / 隔离级别
• ✅ 存储过程：CALL + IN/OUT 参数
• ✅ 批量插入：executemany（协议级 batch）
• ✅ 连接池：线程安全，自动重连
• ✅ DB-API 2.0 兼容层：惰性游标（fetchone/fetchmany/fetchall 按需拉取）
• ✅ 健壮性：超时控制、断线重连、错误码表

安装

pip install dm-pure

零依赖：加密算法（DES/AES/RC4）内置纯 Python 实现。若环境已安装 pycryptodome 会自动优先使用（性能更优），否则用内置实现，功能完全一致。

快速上手

底层 API

from dmpy import DMClient

c = DMClient("127.0.0.1", 5236, "SYSDBA", "password")   # 默认自动检测服务器字符集
c = DMClient(..., encoding="UTF-8")                       # 也可手动指定编码
c.connect()
c.execute("SELECT id, name FROM t WHERE id = ?", [1])     # → (cols, rows)
c.execute("INSERT INTO t VALUES (?, ?)", [1, "x"])        # → (None, 影响行数)
c.begin(); c.execute(...); c.commit()                     # 事务
c.close()

中文/字符集支持

达梦 DM8 服务器初始化时可指定字符集（GB18030 或 UTF-8 等），服务器在握手响应参数区偏移 28 （int32：1=UTF8, 2=EUCKR, 其他=GB18030，对照 Go 驱动 zq.go Dm_build_1282=28）告知客户端。

• 默认自动检测：DMClient(host, port, user, pwd)（encoding=None）连接握手时读取偏移 28 并 切换到对应编码，SQL 文本、参数绑定、结果集、CLOB 读写在检测后统一按该编码编解码。
• 手动覆盖：DMClient(..., encoding="UTF-8") 强制指定（服务器不返回编码或需要特殊处理时使用）。

中文乱码的根因：服务器为 UTF-8 而客户端固定按 GB18030 编解码（或反之）。 旧版驱动硬编码 GB18030；本版握手时自动适配，GB18030 服务器行为与旧版完全一致。

已知限制

1. XDEC_INT32 / XDEC_INT64（24/25）：达梦内部类型，返回原始 bytes（Go 驱动同样回退）。
2. 超大结果集：cmd=7 只发一次拉取（请求带 maxRows=INT64_MAX），未对响应"还有更多" 标志（0xFFFF）做循环验证；实测中量级结果正常，建议对 >6 万行结果集做回归。
3. 连接池：每次 acquire 都执行 SELECT 1 保活检查，高频借还场景有网络开销。
4. TZ 参数绑定：TZ 类型目前仅支持读取（解码带 tzinfo），参数绑定仍按普通 TIME/DATETIME。
5. 发送帧间 2ms sleep 无条件（_send 固定 0.002s，Rust 版可配置 send_delay=0）。

已修复：时区类型（TIME_TZ/DATETIME_TZ/DATETIME2/DATETIME2_TZ）完整解码（含时区偏移）； DML 影响行数改为 int64（>65535 不截断，对照 Go Dm_build_354）； 事务中断线不再静默重连（明确报错，防未提交事务丢失）。

DB-API 2.0（sqlite3/pymysql 风格）

import dmpy.dbapi as dm

conn = dm.connect(host="127.0.0.1", port=5236, user="SYSDBA", password="...")
cur = conn.cursor()
cur.execute("SELECT id, name FROM t WHERE id >= ?", [1])
rows = cur.fetchall()          # 大结果集自动分批拉取
cur.executemany("INSERT INTO t VALUES (?, ?)", [[1, "a"], [2, "b"]])
conn.commit()
conn.close()

存储过程

from dmpy import DMClient, OUT

c = DMClient("127.0.0.1", 5236, "SYSDBA", "password")
c.connect()
r = c.execute("CALL sp_test(?, ?, ?, ?)", [5, "hello", OUT(int), OUT(str)])
print(r[2])  # → [10, "hello"]（OUT 参数返回值）

连接池

from dmpy.pool import ConnectionPool

pool = ConnectionPool(host="127.0.0.1", port=5236, user="SYSDBA", password="...",
                      min_connections=2, max_connections=10)
with pool.connection() as conn:
    cur = conn.cursor()
    cur.execute("SELECT 1")
pool.close()

大 LOB

# 写入 1MB CLOB/BLOB（自动分段传输）
c.execute("INSERT INTO t (id, content, data) VALUES (?, ?, ?)",
          [1, "大文本..." * 50000, b"\x00\x01..." * 100000])
# 读取（自动完整拉取）
rows = c.execute("SELECT content, data FROM t WHERE id = ?", [1])[1]

项目结构

dmpy/
├── client.py       # 核心：连接/查询/参数绑定/LOB/事务/存储过程/惰性查询
├── crypto.py       # 加密（内置纯 Python + 可选 pycryptodome）
├── _pure_crypto.py # 纯 Python DES/AES/RC4 实现
├── frame.py        # 协议帧封装/解析
├── result.py       # 列描述符/行数据/类型解码
├── dbapi.py        # DB-API 2.0 兼容层
└── pool.py         # 连接池

环境要求

• Python >= 3.8
• 可达的达梦 DM8 服务器（TCP 5236 端口）

测试

python test_all.py   # 22 项回归断言（需可达的达梦服务器，修改 test_all.py 中的连接参数）

发布

发布到 PyPI 的流程见 PUBLISHING.md。