DM8-ODBC连接使用
一 ODBC介绍
ODBC（Open Database Connectivity，开放数据库互连）是一种数据库访问标准接口，通过统一的API屏蔽不同数据库之间的差异，使应用程序无需关心底层数据库类型，只需要通过ODBC Driver Manager调用对应数据库驱动即可完成连接和SQL操作。ODBC主要由应用程序、ODBC驱动管理器和数据库驱动组成，其中数据库厂商提供对应的ODBC Driver，用于实现应用程序与数据库之间的通信、数据类型转换和SQL执行。
在DM中，ODBC主要用于实现外部应用程序或异构数据库工具对DM数据库的访问。例如，Windows或Linux上的报表工具、ETL工具、C/C++应用程序可以通过DM提供的ODBC驱动连接DM数据库，而无需直接调用DM底层接口。在Linux环境中，通常通过unixODBC作为ODBC驱动管理器，配置odbc.ini定义数据源（DSN），在odbcinst.ini中指定DM ODBC驱动（如libdodbc.so）路径，应用程序通过DSN连接DM实例，实现数据查询、写入和事务操作。在数据库迁移场景中，ODBC还常用于Oracle、MySQL等异构数据库与DM之间的数据交换，例如通过ETL工具调用源库和目标库的ODBC接口完成数据迁移。
二 配置使用
2.1 WINDOWS
如果已经在本机上部署了DM8数据库并创建过实例，检查达梦安装目录下的bin文件夹中是否有dodbc.dll文件，确认驱动是否存在。
执行win+R，输入odbcad32调用odbc配置程序
如果电脑是64位，默认打开就是64位的。找到右侧添加选项。
选择需要添加的驱动，我这里是要添加DM8的驱动，选择DM8 ODBC DRIVER。
调整要添加的DSN配置的名称，不要重复使用否则会冲突，服务器本机上填localhost，远端可以填具体的IP，然后输入用户密码。之后点击右侧的测试键，进入之后点击测试连接，连接通过之后确定保存。
没问题之后点击确认就可以看到刚才添加的数据源。
2.2 Windows使用示例
编写python程序使用odbc连接DM8数据库做一些操作，这里做一个简单的连接查询。
from __future__ import annotations
import sys
import time
import traceback
import pyodbc
# ---------------------------------------------------------------- 连接配置
DSN = "DM8"
UID = "SYSDBA"
PWD = "DMdba123"
CONN_STR = f"DSN={DSN};UID={UID};PWD={PWD}"
DRIVER_NAME = "DM8 ODBC DRIVER"
QUERY_T1_TOP = "SELECT TOP 5 * FROM T1"
QUERY_T1_LIMIT = "SELECT * FROM T1 LIMIT 5"
SEP = "=" * 62
def _run(cur: pyodbc.Cursor, sql: str):
"""执行 SQL，返回 cursor；失败时由调用方决定是否降级。"""
cur.execute(sql)
return cur
def print_resultset(cur: pyodbc.Cursor, limit: int = 50) -> int:
"""打印结果集的列名与数据行，返回实际数据行数。"""
if cur.description is None:
print(" (无结果集)")
return 0
columns = [c[0] for c in cur.description]
rows = cur.fetchmany(limit)
if not rows:
print(f" 查询成功，但 0 行返回。列: {', '.join(columns)}")
return 0
# 按内容宽度做对齐，CJK 字符按 2 个宽度计算
def width(text: str) -> int:
return sum(2 if ord(ch) > 0x2E80 else 1 for ch in text)
def pad(text: str, target: int) -> str:
return text + " " * max(0, target - width(text))
widths: list[int] = []
for idx, col in enumerate(columns):
col_width = max(width(col), *(width(str(r[idx])) for r in rows), 4)
widths.append(col_width)
print(" " + " | ".join(pad(c, w) for c, w in zip(columns, widths)))
print(" " + "-+-".join("-" * w for w in widths))
for row in rows:
print(" " + " | ".join(pad(str(v), w) for v, w in zip(row, widths)))
return len(rows)
def main() -> int:
print(SEP)
print("达梦 DM8 连接测试")
print(SEP)
print(f" DSN : {DSN}")
print(f" DRIVER : {DRIVER_NAME}")
print(f" UID : {UID}")
print(f" PWD : {'*' * len(PWD)}")
print()
started = time.perf_counter()
try:
conn = pyodbc.connect(CONN_STR, timeout=10, autocommit=True)
except pyodbc.Error as exc:
print("[FAIL] 连接失败")
print(f" {exc}")
print()
print("排查建议：")
print(" 1. 确认 DM8 服务已启动（dmserver 进程）")
print(" 2. 确认 ODBC 数据源存在：odbcad32.exe -> 配置 -> 系统 DSN")
print(" 3. 确认账号密码与监听端口正确")
return 1
connected_at = time.perf_counter()
print(f"[ OK ] 连接成功，耗时 {(connected_at - started) * 1000:.1f} ms")
print()
try:
with conn.cursor() as cur:
# 1) 基础查询
try:
_run(cur, "SELECT 1")
print(f"[ OK ] 基础查询可用 SELECT 1 -> {cur.fetchone()[0]}")
except pyodbc.Error as exc:
print(f"[FAIL] 基础查询失败: {exc}")
return 1
# 2) 服务端版本
try:
cur.execute("SELECT BANNER FROM V$VERSION")
banner = cur.fetchone()
if banner:
print(f"[ OK ] 服务端版本 {banner[0]}")
except pyodbc.Error:
print("[SKIP] V$VERSION 不可读（权限或兼容模式限制），跳过版本检查")
# 3) 当前连接用户
try:
cur.execute("SELECT USER FROM DUAL")
user = cur.fetchone()
if user:
print(f"[ OK ] 当前连接用户 {user[0]}")
except pyodbc.Error:
print("[SKIP] 无法读取当前用户")
# 4) 业务表 T1
print()
print(SEP)
print("业务表 T1 数据读取")
print(SEP)
for sql in (QUERY_T1_TOP, QUERY_T1_LIMIT):
try:
_run(cur, sql)
except pyodbc.Error as exc:
print(f"[FAIL] 查询失败 ({sql})")
print(f" {exc}")
if sql is QUERY_T1_TOP:
print(" 尝试改用 LIMIT 写法...")
continue
return 1
print(f"[ OK ] {sql}")
try:
n = print_resultset(cur)
print(f" -> 返回 {n} 行")
except pyodbc.Error as exc:
print(f"[FAIL] 读取结果集失败: {exc}")
return 1
break
else:
return 1
except pyodbc.Error as exc:
print(f"[FAIL] 执行过程中出错: {exc}")
return 1
except Exception: # noqa: BLE001
print("[FAIL] 未预期异常")
traceback.print_exc()
return 1
finally:
conn.close()
print()
print(f"[ OK ] 连接已关闭，总耗时 {(time.perf_counter() - started) * 1000:.1f} ms")
print()
print("测试通过：DM8 连通性正常。")
return 0
if __name__ == "__main__":
sys.exit(main())
这里的DM8是上面添加的DSN，DRIVER_NAME填写具体的驱动名。
查看程序调用结果，这里我查询的T1表前5行数据，数据没有问题。
在数据库中查看比对，确认两边查出的数据一致。
2.3 Linux
首先安装ODBC依赖，没有的话需要下载。
yum install unixODBC unixODBC-devel -y
查看odbc安装路径
odbcinst -j
/etc/odbcinst.ini —— 驱动注册表
定义系统里有哪些ODBC驱动可用,每个数据库驱动在这里注册一次。libdodbc.so这里使用的是数据库自带的，默认在bin目录下。
/etc/odbc.ini —— 系统级 DSN 数据源
定义"连哪个库、用哪个驱动、以什么参数连",每个目标数据库一个 粗体DSN(Data Source Name)条目。
配置驱动信息
在odbcinst.ini文件中添加以下内容，注：libdodbc.so文件在安装数据库的bin目录下
vi /etc/odbcinst.ini
#添加以下内容
Driver for Dameng Database
[DM8 ODBC DRIVER]
Description = DM8 ODBC Driver
Driver = /home/dmdba/dm/dmdbms/bin/libdodbc.so
配置DSN
编辑/etc/odbc.ini文件，添加以下内容
[系统未激活][root@zhft ~]# more /etc/odbc.ini
[dm8]
Description = DM8 ODBC DSN
Driver = DM8 ODBC DRIVER
SERVER = 192.168.159.124
UID = SYSDBA
PWD = DMdba123
TCP_PORT = 5236
使用isql测试连接
isql -v DSN 用户名 密码
# isql 是做数据库连通性测试的常用命令行工具。省略 -v 则不显示连接过程信息,出错时不报错,建议测试时一定加 -v。
#这里由于用户密码已经配置在odbc.ini文件中，因此不需要再输入用户密码。
[root@zhft ~]# isql dm8 -v
查询一下源端的测试表。
SQL>select * from test;
程序中使用odbc调用远程数据库，在本机中添加Linux中的DSN。
驱动添加过程同上，名称与windows的做区分，否则会冲突，服务器填写具体的IP信息，用户名以及密码。确保网络是相通的，填写完成之后点击测试-->测试数据源。出现测试成功的提示之后点击确认即可保存。
添加完成的话会有以下显示。
2.4 Linux使用示例
使用程序调用DSN进行简单查询，这里的DSN修改为Linux配置的名字。
使用Linux的ODBC做一个简单的查询。
连接数据库创建一张新表new_T，并插入100行数据。最后删除表test。
import pyodbc
import time
# 连接配置
DSN = "dm8linux"
UID = "SYSDBA"
PWD = "DMdba123"
CONN_STR = f"DSN={DSN};UID={UID};PWD={PWD}"
def main():
print("=" * 50)
print("DM8 ODBC 测试")
print("=" * 50)
start = time.time()
try:
conn = pyodbc.connect(
CONN_STR,
timeout=10,
autocommit=True
)
print("[OK] DM8连接成功")
cur = conn.cursor()
# 创建新表
cur.execute("""
CREATE TABLE NEW_T(
ID INT,
NAME VARCHAR(50)
)
""")
print("[OK] 创建表 NEW_T")
# 插入100条数据
for i in range(1, 101):
cur.execute(
"INSERT INTO NEW_T VALUES (?, ?)",
(i, f"TEST_{i}")
)
print("[OK] 插入100条数据")
# 查询NEW_T
cur.execute("""
SELECT COUNT(*) FROM NEW_T
""")
count = cur.fetchone()[0]
print(f"[OK] NEW_T数据量: {count} 行")
# 查询前5行
cur.execute("""
SELECT TOP 5 * FROM NEW_T ORDER BY ID
""")
print("NEW_T前5行:")
for row in cur.fetchall():
print(row)
# 删除旧表TEST
try:
cur.execute("DROP TABLE TEST")
print("[OK] 删除旧表 TEST")
except Exception as e:
print("[WARN] 删除TEST失败或表不存在")
print(e)
# 删除后查询TEST
try:
cur.execute("SELECT COUNT(*) FROM TEST")
print("[FAIL] TEST表仍然存在")
except Exception as e:
print("[OK] TEST表删除验证成功")
print(f" 查询失败: {e}")
cur.close()
conn.close()
print()
print(
f"测试完成，耗时 {(time.time()-start)*1000:.1f} ms"
)
except Exception as e:
print("[FAIL] 测试失败")
print(e)
if __name__ == "__main__":
main()
第一步显示DM8连接成功，说明ODBC连接没有问题。
下一步创建表NEW_T。
CREATE TABLE NEW_T(
ID INT,
NAME VARCHAR(50)
接着插入100行数据
for i in range(1, 101):
cur.execute(
"INSERT INTO NEW_T VALUES (?, ?)",
(i, f"TEST_{i}")
)
然后打印前5行，数据没有问题。
SELECT TOP 5 * FROM NEW_T ORDER BY ID
最后删除测试表在查询
DROP TABLE TEST
SELECT COUNT(*) FROM TEST
最后去数据库中查询一下看是否已删除。
作者
文章
阅读量
获赞
联系客服
达梦在线服务平台社区用户管理规范
鄂ICP备18017926号-1