达梦数据库 DB-API 2.0 驱动 — macOS 用原生 dmclient，非 macOS 透传官方 dmPython

Project description

ns-pydm

达梦（DM）数据库 Python 驱动，DB-API 2.0 风格接口，跨平台智能路由。

操作系统	后端	依赖	说明
macOS (Intel / M系列)	GraalVM native-image dmclient	无需 Java	原生可执行文件，~43MB
Linux	官方 dmPython 透传	`pip install dmPython`	零包装，零性能损失
Windows	官方 dmPython 透传	`pip install dmPython`	零包装，零性能损失

💡 非 macOS 上不做任何包装，直接透传 dmPython，零性能损失。

特性

完整的 DB-API 2.0 飞行合规（apilevel="2.0"）
跨平台统一 API，一套代码跑遍 macOS / Linux / Windows
macOS 原生后端无需 JVM，彻底避免 OOM 和进程被杀问题
内置线程安全连接池（ConnectionPool）
支持 ? 占位符（qmark），自动兼容 %s 格式
支持多架构：macOS ARM64 (M1/M2/M3/M4) + Intel x64
事务管理、自动提交、上下文管理器
BLOB / CLOB / 日期 / Decimal 类型自动转换

安装

pip install ns-pydm

# 非 macOS 系统（Linux/Windows）需额外安装：
pip install dmPython

快速开始

基本连接与查询

import nspydm

# 创建连接
conn = nspydm.connect(
    user="SYSDBA",
    password="SYSDBA",
    server="127.0.0.1",
    port=5236,
)

# 执行查询
cur = conn.cursor()
cur.execute("SELECT ? AS x", [1])
print(cur.fetchone())  # (1,)

# 关闭连接
conn.close()

使用上下文管理器（推荐）

自动提交/回滚 + 自动关闭连接：

import nspydm

with nspydm.connect(user="SYSDBA", password="SYSDBA", server="127.0.0.1", port=5236) as conn:
    cur = conn.cursor()
    cur.execute("SELECT 1")
    print(cur.fetchone())  # (1,)
# 正常退出 → 自动 commit + close
# 异常退出 → 自动 rollback + close

连接参数

nspydm.connect(
    user="SYSDBA",              # 用户名
    password="SYSDBA",          # 密码
    server="127.0.0.1",         # 服务器地址（也可用 host）
    port=5236,                  # 端口，默认 5236
    dsn=None,                   # DSN 字符串
    url=None,                   # JDBC URL（优先级最高，仅 macOS dmclient 后端）
    schema=None,                # 默认 schema
    catalog=None,               # 数据库 catalog
    autocommit=False,           # 自动提交（也接受 autoCommit）
    properties=None,            # 额外的连接属性（仅 macOS dmclient 后端）
    # 高级参数（仅 macOS 后端生效）
    ssl_path=None,              # SSL 证书路径
    ssl_pwd=None,               # SSL 密码
    login_timeout=None,         # 登录超时（秒）
    connection_timeout=None,    # 连接超时（秒）
    txn_isolation=None,         # 事务隔离级别
)

参数说明：

server 和 host 效果相同，只能设其中一个
url 优先级最高，设置后将忽略 server/host/port/dsn
dsn 若以 jdbc: 开头则直接使用，否则自动拼接为 jdbc:dm://{dsn}
autocommit 和 autoCommit 都支持

参数绑定

推荐使用 ? 占位符（qmark 风格）：

cur = conn.cursor()

# 单参数
cur.execute("SELECT * FROM users WHERE id = ?", [42])

# 多参数
cur.execute("SELECT * FROM users WHERE age > ? AND city = ?", [18, "Beijing"])

# INSERT
cur.execute("INSERT INTO users (name, age) VALUES (?, ?)", ["Alice", 30])

# 也兼容 %s 格式（自动转换）
cur.execute("SELECT * FROM users WHERE id = %s", [42])

事务管理

import nspydm

conn = nspydm.connect(user="SYSDBA", password="SYSDBA", server="127.0.0.1", port=5236)
cur = conn.cursor()

try:
    cur.execute("INSERT INTO orders (product, qty) VALUES (?, ?)", ["Widget", 10])
    cur.execute("UPDATE inventory SET stock = stock - ? WHERE product = ?", [10, "Widget"])
    conn.commit()        # 提交事务
except Exception as e:
    conn.rollback()      # 回滚事务
    raise
finally:
    conn.close()

自动提交模式

# 方式一：连接时设置
conn = nspydm.connect(..., autocommit=True)

# 方式二：运行时切换
conn.autocommit = True   # 开启
conn.autocommit = False  # 关闭

数据操作 CRUD

建表

cur.execute("""
    CREATE TABLE employees (
        id INT PRIMARY KEY,
        name VARCHAR(100),
        salary DECIMAL(10,2),
        hire_date DATE,
        active BOOLEAN
    )
""")
conn.commit()

插入数据

# 单行插入
cur.execute(
    "INSERT INTO employees (id, name, salary, hire_date, active) VALUES (?, ?, ?, ?, ?)",
    [1, "Alice", 85000.50, "2023-01-15", True]
)

# 批量插入
employees = [
    [2, "Bob", 72000.00, "2023-03-20", True],
    [3, "Carol", 91000.75, "2022-11-01", True],
    [4, "David", 68000.00, "2024-01-10", False],
]
cur.executemany(
    "INSERT INTO employees (id, name, salary, hire_date, active) VALUES (?, ?, ?, ?, ?)",
    employees
)
conn.commit()
print(f"插入 {cur.rowcount} 行")

查询数据

# fetchone — 取一行
cur.execute("SELECT * FROM employees WHERE id = ?", [1])
row = cur.fetchone()
print(row)  # (1, 'Alice', Decimal('85000.50'), datetime.date(2023, 1, 15), True)

# fetchmany — 取 N 行
cur.execute("SELECT * FROM employees ORDER BY id")
rows = cur.fetchmany(2)
print(rows)  # [(1, 'Alice', ...), (2, 'Bob', ...)]

# fetchall — 取全部
cur.execute("SELECT name, salary FROM employees WHERE active = ?", [True])
for row in cur.fetchall():
    print(f"{row[0]}: {row[1]}")

# 迭代器方式
cur.execute("SELECT name FROM employees")
for row in cur:
    print(row[0])

更新数据

cur.execute("UPDATE employees SET salary = ? WHERE id = ?", [95000.00, 1])
conn.commit()
print(f"更新 {cur.rowcount} 行")

删除数据

cur.execute("DELETE FROM employees WHERE active = ?", [False])
conn.commit()
print(f"删除 {cur.rowcount} 行")

删表

cur.execute("DROP TABLE employees")
conn.commit()

数据类型映射

达梦数据库类型	Python 类型	示例
INT / BIGINT / SMALLINT	`int`	`42`
DECIMAL / NUMERIC	`Decimal`	`Decimal('85000.50')`
DOUBLE / FLOAT / REAL	`float`	`3.14`
VARCHAR / CHAR / CLOB	`str`	`'hello'`
DATE	`datetime.date`	`date(2023, 1, 15)`
TIMESTAMP	`datetime.datetime`	`datetime(2023, 1, 15, 10, 30, 0)`
BLOB / BINARY	`bytes`	`b'\x89PNG...'`
BOOLEAN	`bool`	`True`

写入时的参数转换：

Python 类型	达梦数据库接受方式
`int` / `float`	直接传递
`Decimal`	自动转字符串
`datetime` / `date` / `time`	自动转 ISO 格式字符串
`bytes` / `bytearray`	自动 Base64 编码
`str`	直接传递

连接池

适用于高频连接场景，避免反复创建/销毁连接的开销：

创建连接池

from nspydm import create_pool

pool = create_pool(
    user="SYSDBA",
    password="SYSDBA",
    server="127.0.0.1",
    port=5236,
    min_size=2,       # 初始连接数
    max_size=20,      # 最大连接数
    max_wait=30.0,    # 获取连接超时（秒）
)

使用连接池

# 方式一：上下文管理器（推荐）
with pool.get_connection() as conn:
    cur = conn.cursor()
    cur.execute("SELECT 1")
    print(cur.fetchone())
# 退出 with 块后连接自动归还到池

# 方式二：手动获取/归还
conn = pool.get_connection()
try:
    cur = conn.cursor()
    cur.execute("SELECT 1")
    print(cur.fetchone())
finally:
    conn.close()  # 不是真关闭，而是归还到池

连接池上下文管理器

# 池级别的上下文管理器
with create_pool(user="SYSDBA", password="SYSDBA", server="127.0.0.1") as pool:
    with pool.get_connection() as conn:
        cur = conn.cursor()
        cur.execute("SELECT 1")
        print(cur.fetchone())
# 退出时自动关闭所有连接

监控连接池状态

pool = create_pool(user="SYSDBA", password="SYSDBA", server="127.0.0.1",
                   min_size=2, max_size=10)

print(pool.size)           # 当前总连接数
print(pool.idle_count)     # 空闲连接数
print(pool.in_use_count)   # 正在使用的连接数
print(pool.total_created)  # 历史创建的总连接数
print(pool.closed)         # 池是否已关闭
print(repr(pool))          # <ConnectionPool(open) idle=2 in_use=0 max=10>

关闭连接池

pool.close_all()  # 关闭所有连接，池进入关闭状态

多线程使用

import threading
import nspydm

def worker(pool, thread_id):
    with pool.get_connection() as conn:
        cur = conn.cursor()
        cur.execute("SELECT ? AS thread", [thread_id])
        result = cur.fetchone()
        print(f"Thread {thread_id}: {result}")

pool = nspydm.create_pool(
    user="SYSDBA", password="SYSDBA", server="127.0.0.1",
    min_size=2, max_size=10,
)

threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(pool, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

pool.close_all()

DB-API 2.0 兼容性

import nspydm

# 模块属性
print(nspydm.apilevel)       # "2.0"
print(nspydm.threadsafety)   # 1
print(nspydm.paramstyle)     # "qmark"

# 类型常量
print(nspydm.STRING)         # "STRING"
print(nspydm.BINARY)         # "BINARY"
print(nspydm.NUMBER)         # "NUMBER"
print(nspydm.DATETIME)       # "DATETIME"
print(nspydm.ROWID)          # "ROWID"

# 类型构造函数
d = nspydm.Date(2024, 1, 15)              # datetime.date(2024, 1, 15)
t = nspydm.Time(10, 30, 0)                # datetime.time(10, 30)
ts = nspydm.Timestamp(2024, 1, 15, 10, 30, 0)  # datetime.datetime(2024, 1, 15, 10, 30)
b = nspydm.Binary(b"hello")               # b'hello'

# 异常层次
# Warning → Error → InterfaceError / DatabaseError → DataError / OperationalError / ...

完整使用案例

案例1：Web 服务数据库层

import nspydm
from nspydm import create_pool

# 应用启动时创建连接池
pool = create_pool(
    user="SYSDBA",
    password="your_password",
    server="192.168.1.100",
    port=5236,
    min_size=5,
    max_size=50,
    max_wait=10.0,
)

def get_user_by_id(user_id: int):
    """根据 ID 查询用户"""
    with pool.get_connection() as conn:
        cur = conn.cursor()
        cur.execute("SELECT id, name, email FROM users WHERE id = ?", [user_id])
        return cur.fetchone()

def create_user(name: str, email: str):
    """创建新用户"""
    with pool.get_connection() as conn:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            [name, email]
        )
        # with 块正常退出自动 commit

def update_user_email(user_id: int, new_email: str):
    """更新用户邮箱"""
    with pool.get_connection() as conn:
        cur = conn.cursor()
        cur.execute(
            "UPDATE users SET email = ? WHERE id = ?",
            [new_email, user_id]
        )

# 应用关闭时
# pool.close_all()

案例2：数据迁移脚本

import nspydm

def migrate_data(source_conn_params, target_conn_params):
    """从源库读取数据，写入目标库"""
    src = nspydm.connect(**source_conn_params)
    dst = nspydm.connect(**target_conn_params)

    try:
        # 读取源数据
        src_cur = src.cursor()
        src_cur.execute("SELECT id, name, value FROM source_table")
        rows = src_cur.fetchall()

        # 批量写入目标库
        dst_cur = dst.cursor()
        dst_cur.executemany(
            "INSERT INTO target_table (id, name, value) VALUES (?, ?, ?)",
            rows
        )
        dst.commit()
        print(f"迁移完成，共 {len(rows)} 条记录")

    except Exception:
        dst.rollback()
        raise
    finally:
        src.close()
        dst.close()

# 使用
migrate_data(
    source_conn_params={"user": "SYSDBA", "password": "src_pwd", "server": "10.0.0.1", "port": 5236},
    target_conn_params={"user": "SYSDBA", "password": "dst_pwd", "server": "10.0.0.2", "port": 5236},
)

案例3：定时报表生成

import nspydm
from datetime import date

def generate_daily_report():
    """生成每日报表"""
    with nspydm.connect(user="SYSDBA", password="SYSDBA",
                         server="127.0.0.1", port=5236) as conn:
        cur = conn.cursor()

        # 查询今日订单汇总
        today = date.today().isoformat()
        cur.execute("""
            SELECT product, SUM(qty) AS total_qty, SUM(amount) AS total_amount
            FROM orders
            WHERE order_date = ?
            GROUP BY product
            ORDER BY total_amount DESC
        """, [today])

        results = cur.fetchall()
        print(f"=== {today} 订单报表 ===")
        print(f"{'产品':<20} {'数量':>10} {'金额':>15}")
        print("-" * 47)
        for row in results:
            print(f"{row[0]:<20} {row[1]:>10} {row[2]:>15}")

        return results

generate_daily_report()

案例4：BLOB 文件存储

import nspydm

def save_file(filename: str, data: bytes):
    """将文件保存到数据库"""
    with nspydm.connect(user="SYSDBA", password="SYSDBA",
                         server="127.0.0.1", port=5236) as conn:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO files (name, content) VALUES (?, ?)",
            [filename, nspydm.Binary(data)]
        )

def load_file(filename: str) -> bytes:
    """从数据库读取文件"""
    with nspydm.connect(user="SYSDBA", password="SYSDBA",
                         server="127.0.0.1", port=5236) as conn:
        cur = conn.cursor()
        cur.execute("SELECT content FROM files WHERE name = ?", [filename])
        row = cur.fetchone()
        return row[0] if row else None

# 使用
with open("report.pdf", "rb") as f:
    save_file("report.pdf", f.read())

data = load_file("report.pdf")
with open("report_copy.pdf", "wb") as f:
    f.write(data)

架构

┌─────────────────────────────────────────────┐
│                 nspydm                      │
│         (跨平台统一 API 入口)                │
├────────────────┬────────────────────────────┤
│  macOS         │  Linux / Windows          │
│                │                           │
│  dmclient      │  dmPython                 │
│  (原生可执行    │  (官方 C 原生驱动)         │
│   文件)         │                           │
│    ↓           │    ↓                      │
│  DM JDBC       │  DPI (C API)             │
│    ↓           │    ↓                      │
├────────────────┴────────────────────────────┤
│           DM Database Server               │
└─────────────────────────────────────────────┘

dmclient 源码及预编译二进制：https://gitee.com/navysummer/dm-client

环境变量

变量名	说明
`DMCLIENT_PATH`	指定 dmclient 可执行文件的完整路径（仅 macOS）

dmclient 原生客户端

macOS 后端所使用的 dmclient 原生可执行文件由独立项目维护：

👉 https://gitee.com/navysummer/dm-client

该仓库包含 GraalVM native-image 构建脚本、预编译二进制文件以及详细的编译说明。

许可证

本项目仅分发源代码。dmclient 二进制文件由 navysummer/dm-client 提供，其编译依赖达梦 JDBC 驱动，分发需遵循达梦数据库的许可协议。

从旧版迁移

旧版基于 JPype + JVM，新版为跨平台智能路由。API 完全兼容：

# 旧版（需要 Java）
# 新版（无需 Java，jars/jvm_args 参数保留但忽略）
conn = nspydm.connect(user="...", password="...", server="...")

Project details

Release history Release notifications | RSS feed

0.0.15

Jun 16, 2026

0.0.14

Jun 13, 2026

0.0.13

Jun 13, 2026

This version

0.0.12

Jun 13, 2026

0.0.11

Jun 13, 2026

0.0.10

Jun 11, 2026

0.0.9

Jun 11, 2026

0.0.8

Apr 26, 2026

0.0.7

Apr 26, 2026

0.0.6

Apr 25, 2026

0.0.5

Apr 25, 2026

0.0.4

Apr 25, 2026

0.0.3

Apr 25, 2026

0.0.2

Apr 25, 2026

0.0.1

Apr 25, 2026

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

ns_pydm-0.0.12.tar.gz (34.2 kB view details)

Uploaded Jun 13, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

ns_pydm-0.0.12-py3-none-any.whl (28.3 kB view details)

Uploaded Jun 13, 2026 Python 3

File details

Details for the file ns_pydm-0.0.12.tar.gz.

File metadata

Download URL: ns_pydm-0.0.12.tar.gz
Upload date: Jun 13, 2026
Size: 34.2 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for ns_pydm-0.0.12.tar.gz
Algorithm	Hash digest
SHA256	`928190ebba8c560f9de83019e367c89f492a5e777d6f4daa2eb504f93bb2cfdb`
MD5	`c22b1335497cf25dae87bdc01a9ec015`
BLAKE2b-256	`f7b3629400193cbb1b66b12760a3e40983df0aac079099f7458994b2f108ac06`

See more details on using hashes here.

File details

Details for the file ns_pydm-0.0.12-py3-none-any.whl.

File metadata

Download URL: ns_pydm-0.0.12-py3-none-any.whl
Upload date: Jun 13, 2026
Size: 28.3 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for ns_pydm-0.0.12-py3-none-any.whl
Algorithm	Hash digest
SHA256	`35d5cf2f391b22e9a47ef2720d981cc229b2ce5c9adb66093cab4a1988e090a3`
MD5	`e72104b8a4a07c920581fe520d91d28b`
BLAKE2b-256	`6e18f18990d7277f2fa29f7d192f596842773fd7fcdae815940cae6f5accfb6c`

See more details on using hashes here.

ns-pydm 0.0.12

Navigation

Verified details

Maintainers

Unverified details

Meta

Project description

ns-pydm

特性

安装

快速开始

基本连接与查询

使用上下文管理器（推荐）

连接参数

参数绑定

事务管理

自动提交模式

数据操作 CRUD

建表

插入数据

查询数据

更新数据

删除数据

删表

数据类型映射

连接池

创建连接池

使用连接池

连接池上下文管理器

监控连接池状态

关闭连接池

多线程使用

DB-API 2.0 兼容性

完整使用案例

案例1：Web 服务数据库层

案例2：数据迁移脚本

案例3：定时报表生成

案例4：BLOB 文件存储

架构

环境变量

dmclient 原生客户端

许可证

从旧版迁移

Project details

Verified details

Maintainers

Unverified details

Meta

Release history Release notifications | RSS feed

Download files

Source Distribution

Built Distribution

File details

File metadata

File hashes

File details

File metadata

File hashes