一、代码编程

连接数据库

  1. # !/usr/bin/python
  2. # coding:utf-8
  4. import dmPython
  6. try:
  7.     conn = dmPython.connect(user='SYSDBA', password='SYSDBA', server='localhost', port=5236)
  8.     cursor = conn.cursor()
  9.     print('python: conn success!')
  10.     conn.close()
  11. except (dmPython.Error, Exception) as err:
  12.     print(err)
  15. #PATH=PATH=D:\damba\dm8\drivers\\;D:\damba\dm8\drivers\logmnr\\;D:\damba\dm8\bin

绑定数据方式传参查询

  1. #!/usr/bin/python
  2. # coding:utf-8
  3. import dmPython
  4. ## 绑定数据的方式 ?
  5. try:
  6.     conn = dmPython.connect(user='SYSDBA', password='SYSDBA', server='localhost', port=5236, local_code='utf-8')
  7.     cursor = conn.cursor()
  8.     try:
  9.         # 清空表，初始化测试环境
  10.         cursor.execute('delete from PRODUCTION.PRODUCT_CATEGORY')
  11.     except (dmPython.Error, Exception) as err:
  12.         print (err)
  14.     try:
  15.         # 插入数据
  16.         values = ('稍等')
  17.         cursor.execute("insert into PRODUCTION.PRODUCT_CATEGORY(name) values(?)", values)
  18.         print('python: insert success!')
  20.         # 查询数据
  21.         cursor.execute("select name from PRODUCTION.PRODUCT_CATEGORY")
  22.         res = cursor.fetchall()
  23.         for tmp in res:
  24.             for c1 in tmp:
  25.                 print c1
  27.         print('python: select success!')
  28.     except (dmPython.Error, Exception) as err:
  29.         print(err)
  31.     conn.close()
  32. except (dmPython.Error, Exception) as err:
  33.     print(err)

调用DM的存储函授

  1. #!/usr/bin/python
  2. # coding:utf-8
  3. #coding:utf-8
  6. import dmPython
  8. try:
  9.     conn = dmPython.connect(user='SYSDBA', password='SYSDBA', server='localhost', port=5236, local_code=1)
  10.     cursor = conn.cursor()
  12.     try:
  13.         # 插入数据
  14.         print('python: start success!')
  16.         ## 调用无参数的存储函数
  17.         cursor.execute('create or replace function test_func_1() return int as begin return 123;end;')
  18.         print(cursor.callfunc('test_func_1'))
  20.         ## 调用带参数的存储函数
  21.         cursor.execute('create or replace function test_func_2(p1 int, p2 out int) return int as begin p2 = p1 + 1; return 456;end;')
  22.         print(cursor.callfunc('test_func_2', 10000, 0))
  24.         print('python: select success!')
  26.     except (dmPython.Error, Exception) as err1:
  27.         print (err1)
  28.     conn.close()
  29. except (dmPython.Error, Exception) as err:
  30.     print(err)

调用DM的存储过程

  1. #!/usr/bin/python
  2. # coding:utf-8
  3. #coding:utf-8
  6. import dmPython
  8. try:
  9.     conn = dmPython.connect(user='SYSDBA', password='SYSDBA', server='localhost', port=5236, local_code=1)
  10.     cursor = conn.cursor()
  12.     try:
  13.         # 插入数据
  14.         print('python: start success!')
  16.         # 调用无参的存储过程
  17.         cursor.execute('create or replace procedure test_proc_1() as begin print true;end;')
  18.         print(cursor.callproc('test_proc_1'))
  20.         ## 调用有参的存储过程
  21.         cursor.execute('create or replace procedure test_proc_2(p1 int, p2 out int) as begin p2 = p1 + 1;end;')
  22.         print(cursor.callproc('test_proc_2', 10000, 0))
  24.         paras_list = (10000, 0)
  25.         ## 元组 存储 的存储过程
  26.         cursor.execute('create or replace procedure test_proc_2(p1 int, p2 out int) as begin p2 = p1 + 1;end;')
  27.         print(cursor.callproc('test_proc_2', paras_list))
  29.         ## 调用其他模式有参的存储过程
  30.         cursor.execute('create or replace procedure dmhr.test_proc_2(p1 int, p2 out int) as begin p2 = p1 + 1;end;')
  31.         print(cursor.callproc('test_proc_2', 10000, 0))
  33.         print('python: select success!')
  35.     except (dmPython.Error, Exception) as err1:
  36.         print (err1)
  37.     conn.close()
  38. except (dmPython.Error, Exception) as err:
  39.     print(err)

DM的CRUD

  1. #!/usr/bin/python
  2. # coding:utf-8
  3. import dmPython
  5. try:
  6.     conn = dmPython.connect(user='SYSDBA', password='SYSDBA', server='localhost', port=5236, local_code=1)
  7.     cursor = conn.cursor()
  8.     try:
  9.         # 清空表，初始化测试环境
  10.         cursor.execute('delete from PRODUCTION.PRODUCT_CATEGORY')
  11.     except (dmPython.Error, Exception) as err:
  12.         print(err)
  13.     try:
  14.         # 插入数据
  15.         cursor.execute("insert into PRODUCTION.PRODUCT_CATEGORY(NAME) values('sd'), ('单独'), ('3'), ('4')")
  16.         print('python: insert success!')
  17.         # 删除数据
  18.         cursor.execute("delete from PRODUCTION.PRODUCT_CATEGORY where name='1'")
  19.         print('python: delete success!')
  21.         # 更新数据
  22.         cursor.execute('update PRODUCTION.PRODUCT_CATEGORY set name = \'222\' where name=\'2\'')
  23.         print('python: update success!')
  25.         # 查询数据
  26.         cursor.execute("select name from PRODUCTION.PRODUCT_CATEGORY")
  27.         res = cursor.fetchall()
  28.         for tmp in res:
  29.             for c1 in tmp:
  30.                 print c1
  32.         print('python: select success!')
  33.     except (dmPython.Error, Exception) as err1:
  34.         print (err1)
  35.     conn.close()
  36. except (dmPython.Error, Exception) as err:
  37.     print(err)

执行命令操作fldr

fldr_export.py

  1. #encoding:utf-8
  3. import subprocess
  4. cmd = r"D:\damba\dm8\bin\dmfldr.exe userid=SYSDBA/SYSDBA@localhost:5236 control='E:\export.ctl' mode = 'out'"
  6. p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
  7. for line in p.stdout.readlines():  # 逐行输出
  8.     print(line.decode('GBK'))
  9. retval = p.wait()  # 等待进程结束返回结果
  10. print(retval)
  12. ### -- 创建表sql
  13. # CREATE TABLE "SYSDBA"."TEST"
  14. # (
  15. # "C1" INT,
  16. # "C2" INT,
  17. # "C3" DATE) STORAGE(ON "MAIN", CLUSTERBTR) ;
  18. # -- 插入数据
  19. # INSERT INTO test VALUES(1, 1 , '2021-09-15');
  20. # INSERT INTO test VALUES(2, 2 , '2021-09-16');
  21. # INSERT INTO test VALUES(3, 3 , '2021-09-17');
  22. # INSERT INTO test VALUES(4, null , '2021-09-18');
  24. ###

fldr_import.py

  1. #encoding:utf-8
  3. import subprocess
  4. cmd = r"D:\damba\dm8\bin\dmfldr.exe userid=SYSDBA/SYSDBA@localhost:5236 control='E:\import.ctl'"
  6. p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
  7. for line in p.stdout.readlines():  # 逐行输出
  8.     print(line.decode('GBK'))
  9. retval = p.wait()  # 等待进程结束返回结果
  10. print(retval)
  14. # CREATE TABLE "SYSDBA"."TEST4"
  15. # (
  16. # "C1" INT,
  17. # "C2" VARCHAR(8188)) STORAGE(ON "MAIN", CLUSTERBTR) ;

二、测试案例

1. 报文测试

  1. #!/usr/bin/python
  2. #coding:utf-8
  3. import codecs
  4. import socket
  5. import time
  6. payload="00000000c800520000000000000000000000009a000000000000000001010200000000000000000000000000000000000000000000000000000000000000000009000000382e312e302e313437004000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
  7. s=socket.socket()
  8. s.connect(('120.53.103.235',5236))
  9. shakehand=codecs.decode(payload,'hex')
  10. s.send(shakehand)
  11. time.sleep(1)
  12. print(s.recv(1024))
  13. time.sleep(1)

参考连接

[Python 封装 DM 达梦 数据库操作（使用类封装基本的增删改查）]https://www.cndba.cn/dave/article/3827 [Python 数据库接口]https://eco.dameng.com/docs/zh-cn/app-dev/python-python.html

代码包

pythonCode.zip