用明道云构建并扩展依赖库，定时读取外部数据

文/张春泉

项目需要复杂数据操作，但手头只有数据库；技术没有时间开发，又不想专门采购RPA，产生额外费用，如何进行数据对接呢？今天这套方案就能为你解忧！用明道云链接企业自有的数据库，读取数据再写入明道云。下面进入手把手教学模式，逐一拆解步骤。

第一步：安装明道私有部署

安装步骤参考：https://docs.pd.mingdao.com/deployment/docker-compose/standalone/quickstart.html

第二步：依赖库持久化并扩展依赖库

参考此处扩展代码块依赖库：https://docs.pd.mingdao.com/faq/codeextension.html

链接MySQL请安装 pymysql 扩展库：

docker exec -t $(docker ps | grep community | awk '{print $1}') bash -c 'pip3 install --target=/usr/local/lib/python3.6/site-packages/ pymysql'

1	docker exec -t $(docker ps | grep community | awk '{print $1}') bash -c 'pip3 install --target=/usr/local/lib/python3.6/site-packages/ pymysql'

链接SQL Service请安装 pymssql 扩展库：

docker exec -t $(docker ps | grep community | awk '{print $1}') bash -c 'pip3 install --target=/usr/local/lib/python3.6/site-packages/ pymssql'

1	docker exec -t $(docker ps | grep community | awk '{print $1}') bash -c 'pip3 install --target=/usr/local/lib/python3.6/site-packages/ pymssql'

Oracle linux连接配置参考： https://www.likecs.com/show-495081.html

验证安装完成后，是否可在宿主机目录 /data/mingdao/script/volume/command/package/python-3.6/site-packages/ 中进行查看。

第三步：应用设计

1. 创建《数据库配置》工作表，并设置自定义动作，获取工作表信息。

2. 配置《数据库表》和自定义动作，系统自动获取字段并创建任务。

3. 创建《字段表》

 

数据库连接的逻辑流程是统一的，只是中间执行不同的脚本。所以，我们要将统一的流程抽出来，做成PBP业务流，以提高重复流程的复用率。

MySQL连接脚本如下

import pymysql import json output = {} output['result'] = '' # 打开数据库连接 try: db = pymysql.connect(host=input['host'], user=input['username'], password=input['passwd'], port=int(input['port'])) # 使用 cursor() 方法创建一个游标对象 cursor db.select_db(input['db']) cursor = db.cursor() # 使用 execute() 方法执行 SQL 查询 output['sql'] = input['sqltext'] cursor.execute(input['sqltext']) cols=cursor.description # 使用 fetchone() 方法获取单条数据. datas = cursor.fetchall() allop=[] for data in datas: i = 0 op= {} for col in cols: op[cols[cols.index(col)][i]] =str(data[cols.index(col)]) i=i+1 allop.append(op) output['data'] = json.dumps(allop, ensure_ascii=False) #print(output) # 关闭数据库连接 db.close() output['code'] = 0 output['msg'] = '连接成功' except Exception as e: output['code'] = 1 output['msg'] = '连接失败：{}'.format(e)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

import pymysql import json output = {} output['result'] = '' # 打开数据库连接 try:     db = pymysql.connect(host=input['host'], user=input['username'], password=input['passwd'], port=int(input['port']))       # 使用 cursor() 方法创建一个游标对象 cursor     db.select_db(input['db'])     cursor = db.cursor()     # 使用 execute()  方法执行 SQL 查询     output['sql'] = input['sqltext']     cursor.execute(input['sqltext'])     cols=cursor.description     # 使用 fetchone() 方法获取单条数据.     datas = cursor.fetchall()     allop=[]     for data in datas:         i = 0         op= {}         for col in cols:            op[cols[cols.index(col)][i]] =str(data[cols.index(col)])         i=i+1         allop.append(op)       output['data'] = json.dumps(allop, ensure_ascii=False)     #print(output)     # 关闭数据库连接     db.close()     output['code'] = 0     output['msg'] = '连接成功' except Exception as e:     output['code'] = 1     output['msg'] = '连接失败：{}'.format(e)

获取库下面所有的表

select table_name tableName ,ifnull(TABLE_COMMENT,'') as tabledesc,create_time createTime from information_schema.tables where table_schema ='数据库库名' order by create_time desc

1	select table_name tableName ,ifnull(TABLE_COMMENT,'') as tabledesc,create_time createTime from information_schema.tables where table_schema ='数据库库名' order by create_time desc

获取表的描述信息

select column_name as '列名',data_type as '字段类型',column_comment as '字段注释' ,column_key as '是否主键',is_nullable as '是否允许为空',column_default as '默认值' from information_schema.columns where table_schema='数据库库名' and table_name='表名'

1	select column_name as '列名',data_type as '字段类型',column_comment as '字段注释' ,column_key as '是否主键',is_nullable as '是否允许为空',column_default as '默认值' from information_schema.columns  where table_schema='数据库库名' and table_name='表名'

「获取字段–创建自动任务」的自定义按钮工作流流程图

对接其他系统或者项目对接的时候，日志是不可缺少的，因为它能够帮助你快速定位错误。所以最后在应用里也记得加上哦。

最后，业务部门测试数据也已经写入工作表了。整体对接已完成，小伙伴们快快尝试吧！

 

封装业务流程