影刀RPA 数据库操作入门连接MySQL执行增删改查很多RPA初学者觉得数据库是程序员的事跟自己没关系。但等你开始做数据采集、报表汇总、多系统数据同步的时候就会发现Excel扛不住了——10万行数据Excel打开要半分钟筛选一下又卡半天。这时候数据库就是必经之路。这篇文章从零开始讲清楚在影刀RPA里怎么用Python节点操作MySQL数据库。不要求你有数据库基础跟着走就能跑通。什么情况需要用数据库三条判断标准中了任意一条就该切数据库了1. 单次处理数据超过5万行。Excel的上限不是104万行而是操作体验的上限。超过5万行的Excel筛选、排序、公式计算都会明显变慢。而数据库里的百万级数据查询还是秒级。2. 多个流程需要读写同一份数据。A流程采集数据往里写B流程读数据生成报表C流程根据数据状态触发动作。如果用Excel做中转文件锁、并发写入、数据覆盖这些问题全冒出来了。数据库天然支持并发和事务不用自己造轮子。3. 数据之间有复杂关联。“查询这个客户的最近10笔订单关联商品表和库存表算出总金额和库存状态”——这种查询在Excel里得VLOOKUP套N层在数据库里一条SQL就搞定。但反过来如果你只是处理几百行数据、做简单的加加减减、或者数据只在一个流程里用数据库就是杀鸡用牛刀。Excel完全够用。前置准备安装MySQL驱动在影刀的Python节点里操作MySQL需要一个叫pymysql的库。它是纯Python实现不需要安装额外的系统依赖在影刀环境中可以直接用。但影刀默认环境没有装这个库。你需要先安装第一步找到影刀的Python环境路径。影刀安装目录下有一个python文件夹。在影刀里新建一个Python节点输入以下代码查看Python路径importsysprint(sys.executable)运行后会输出类似C:\Program Files\yingdao\python\python.exe的路径。第二步用pip安装pymysql。在电脑的命令行CMD里执行C:\Program Files\yingdao\python\python.exe -m pip install pymysql看到Successfully installed pymysql-1.x.x就装好了。如果安装报权限错误以管理员身份运行CMD再试。如果公司网络限制了pip可以下载pymysql的.whl文件离线安装拼多多店群自动化上架方案C:\Program Files\yingdao\python\python.exe -m pip install C:\downloads\PyMySQL-1.1.0-py3-none-any.whl连接数据库安装完成后在影刀的Python节点里写连接代码importpymysql# 创建连接connpymysql.connect(host127.0.0.1,# 数据库IP。本地就是127.0.0.1port3306,# MySQL默认端口userroot,# 用户名passwordyour_password,# 密码databaserpa_data,# 要连接的数据库名charsetutf8mb4# 字符集)# 创建游标操作数据库的手柄cursorconn.cursor()print(数据库连接成功)连接参数说明参数说明示例host数据库服务器地址本地127.0.0.1远程192.168.1.100port端口号MySQL默认3306user用户名rootpassword密码你的数据库密码database数据库名要操作的具体数据库charset字符集utf8mb4支持emoji纯中文用utf8也行连接中的常见报错“Can’t connect to MySQL server on ‘127.0.0.1’”MySQL服务没启动。在Windows服务里找到MySQL服务启动它。或者用命令行net start mysql。“Access denied for user ‘root’‘localhost’”用户名或密码错了。确认一下是不是连接的是正确的数据库——有些人电脑上装了多个MySQL实例。“Unknown database ‘rpa_data’”数据库不存在。先用Navicat、DBeaver这类工具创建一个或者用代码创建# 先不指定database连接后创建connpymysql.connect(host127.0.0.1,userroot,passwordyour_password)cursorconn.cursor()cursor.execute(CREATE DATABASE IF NOT EXISTS rpa_data CHARACTER SET utf8mb4)cursor.execute(USE rpa_data)增删改查四件套数据库操作的精髓就是四个字增、删、改、查。来逐个击破。建表第一次用数据库先建一张表。表相当于Excel的一个Sheet定义了每列叫什么、什么类型。create_table_sql CREATE TABLE IF NOT EXISTS orders ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT 主键ID, order_no VARCHAR(50) NOT NULL COMMENT 订单号, customer_name VARCHAR(100) COMMENT 客户名称, amount DECIMAL(10,2) COMMENT 金额, status VARCHAR(20) DEFAULT 待处理 COMMENT 订单状态, created_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT订单表 cursor.execute(create_table_sql)conn.commit()print(表创建成功)几个常用字段类型INT整数比如IDVARCHAR(N)字符串最多N个字符DECIMAL(10,2)精确小数总共10位小数点后2位。金额字段必须用这个不要用FLOAT浮点数有精度问题DATETIME日期时间TEXT长文本插入数据增# 插入单条insert_sql INSERT INTO orders (order_no, customer_name, amount, status) VALUES (%s, %s, %s, %s) cursor.execute(insert_sql,(ORD001,张三贸易,3580.50,待处理))conn.commit()print(f插入成功新ID{cursor.lastrowid})# 批量插入从影刀变量读取数据# 假设 orders_data 是影刀传入的订单列表orders_data[(ORD002,李四科技,12000.00,待处理),(ORD003,王五百货,560.80,待处理),(ORD004,赵六建材,8920.00,待处理),]cursor.executemany(insert_sql,orders_data)conn.commit()print(f批量插入{cursor.rowcount}条)关键点%s是占位符不是Python的字符串格式化。绝对不要用 f-string 拼接SQL会有SQL注入风险# 错误不要这样写cursor.execute(fINSERT INTO orders VALUES ({order_no}))# 正确用参数化查询cursor.execute(INSERT INTO orders (order_no) VALUES (%s),(order_no,))查询数据查# 查询全部cursor.execute(SELECT * FROM orders)all_orderscursor.fetchall()print(f共{len(all_orders)}条记录)# 条件查询cursor.execute(SELECT order_no, customer_name, amount FROM orders WHERE status %s,(待处理,))pending_orderscursor.fetchall()forrowinpending_orders:print(f订单号{row[0]}客户{row[1]}金额¥{row[2]})# 带排序和分页cursor.execute( SELECT order_no, customer_name, amount, created_time FROM orders WHERE amount %s ORDER BY created_time DESC LIMIT %s OFFSET %s ,(1000,20,0))# 金额1000的订单前20条fetchall()返回所有结果fetchone()返回一条fetchmany(N)返回N条。更新数据改# 更新单条cursor.execute(UPDATE orders SET status %s WHERE order_no %s,(已完成,ORD001))conn.commit()print(f更新了{cursor.rowcount}条记录)# 批量更新循环中要小心fororderinto_update:cursor.execute(UPDATE orders SET status %s WHERE order_no %s,(order[new_status],order[order_no]))conn.commit()# 循环结束统一提交不要每条都commit注意UPDATE 不带 WHERE 条件会把整张表的数据全改了。这是数据库操作里最致命的翻车场景之一。写UPDATE之前先SELECT确认一下条件对不对。删除数据删cursor.execute(DELETE FROM orders WHERE order_no %s,(ORD001,))conn.commit()print(f删除了{cursor.rowcount}条记录)DELETE比UPDATE更危险。建议先用SELECT确认要删的数据然后把SELECT的WHERE条件原封不动复制到DELETE里# 步骤1先看看要删什么cursor.execute(SELECT * FROM orders WHERE status %s,(已取消,))to_deletecursor.fetchall()print(f将要删除{len(to_delete)}条记录)forrowinto_delete:print(row)# 步骤2确认无误后再执行删除cursor.execute(DELETE FROM orders WHERE status %s,(已取消,))conn.commit()print(f实际删除了{cursor.rowcount}条)数据库操作的正确流程框架影刀里操作数据库建议遵循这个框架Python节点 - 数据库操作的标准写法 1. 建立连接 2. try: 3. 执行业务操作增删改查 4. conn.commit() ← 提交事务 5. except Exception as e: 6. conn.rollback() ← 出错回滚 7. 记录错误信息 8. finally: 9. cursor.close() ← 关闭游标 10. conn.close() ← 关闭连接完整代码importpymysql connNonecursorNonetry:connpymysql.connect(host127.0.0.1,port3306,userroot,passwordyour_password,databaserpa_data,charsetutf8mb4)cursorconn.cursor()# 业务操作 cursor.execute(SELECT COUNT(*) FROM orders WHERE status 待处理)countcursor.fetchone()[0]print(f待处理订单数{count})# 如果count 0批量更新ifcount0:cursor.execute(UPDATE orders SET status 处理中 WHERE status 待处理)print(f更新了{cursor.rowcount}条订单状态)conn.commit()print(操作完成事务已提交)exceptpymysql.Errorase:ifconn:conn.rollback()print(f数据库操作失败{e})# 把错误信息传回影刀变量供后续判断error_msgstr(e)finally:ifcursor:cursor.close()ifconn:conn.close()print(数据库连接已关闭)为什么连接一定要关MySQL有最大连接数限制默认151。每个Python节点执行完不关连接连接数慢慢就耗光了。后续流程再想连数据库就会报Too many connections。影刀变量与数据库的交互数据在影刀变量和数据库之间来回传递这是最常见的需求。关键是做好格式转换。数据库查询结果 → 影刀变量影刀的Python节点可以定义输出变量。查询结果转成字典列表传回影刀# 查询结果转成字典列表方便影刀后续处理cursor.execute(SELECT id, order_no, customer_name, amount, status FROM orders)columns[id,order_no,customer_name,amount,status]result_list[]forrowincursor.fetchall():result_list.append(dict(zip(columns,row)))# result_list 是输出变量影刀可以直接用# result_list [# {id: 1, order_no: ORD001, customer_name: 张三, amount: 3580.50, status: 待处理},# {id: 2, order_no: ORD002, customer_name: 李四, amount: 12000.00, status: 待处理},# ]转过以后在影刀里result_list就是标准的列表字典结构可以ForEach遍历、可以取字段值跟处理Excel数据一模一样。影刀变量 → 数据库写入反过来影刀采集到的数据写入数据库TEMU店群如何管理运营# data_from_yingdao 是从影刀传入的变量列表字典data_from_yingdao[{order_no:ORD005,customer:新客户A,amount:5000.00},{order_no:ORD006,customer:新客户B,amount:7800.00},]# 转成元组列表用于executemanyinsert_data[(item[order_no],item[customer],item[amount])foritemindata_from_yingdao]cursor.executemany(INSERT INTO orders (order_no, customer_name, amount) VALUES (%s, %s, %s),insert_data)conn.commit()用json中转复杂数据如果数据结构太复杂嵌套很深可以用json序列化中转importjson# 复杂数据存为JSON字符串complex_data{order_no:ORD007,items:[{name:商品A,qty:2,price:100},{name:商品B,qty:1,price:500},],discount:0.9,total:630}cursor.execute(INSERT INTO orders (order_no, detail_json) VALUES (%s, %s),(ORD007,json.dumps(complex_data,ensure_asciiFalse)))conn.commit()# 读取时反序列化cursor.execute(SELECT detail_json FROM orders WHERE order_no ORD007)rowcursor.fetchone()detailjson.loads(row[0])print(detail[items][0][name])# 输出商品A这个技巧在存采集日志、复杂配置、嵌套业务数据时特别有用。四个踩坑实录坑1字符编码导致中文乱码数据写进去看起来正常读出来全是问号或乱码。根因是字符集没设对。MySQL默认字符集可能是latin1不支持中文。解决连接时指定charsetutf8mb4建表时指定DEFAULT CHARSETutf8mb4如果数据库已经建好了执行ALTER DATABASE rpa_data CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci坑2连接没关导致Too many connections流程跑了几轮后突然报这个错。排查后发现每次Python节点执行完没有cursor.close()和conn.close()。连接池不是你关不关闭都能自动回收的——MySQL服务端要等到超时才会释放闲置连接默认8小时。如果你一小时跑100次流程每次开一个新连接不关100个连接就占满了。铁律用 try-finally 确保连接一定关闭写成肌肉记忆。坑3SQL注入风险从影刀变量里拼SQL字符串遇上用户输入里带单引号就直接崩nameOBrien# 名字里有个单引号# 错误写法sqlfSELECT * FROM orders WHERE customer_name {name}# 生成SELECT * FROM orders WHERE customer_name OBrien# SQL语法错误单引号提前闭合了永远用参数化查询%s占位符不要字符串拼接。pymysql会自动处理特殊字符的转义。坑4Decimal类型在JSON序列化时报错amount字段是DECIMAL(10,2)pymysql读出来是Decimal(3580.50)类型。如果直接json.dumps()会报Object of type Decimal is not JSON serializable。fromdecimalimportDecimalimportjsonclassDecimalEncoder(json.JSONEncoder):defdefault(self,obj):ifisinstance(obj,Decimal):returnfloat(obj)returnsuper().default(obj)# 使用自定义编码器json.dumps(data,clsDecimalEncoder)或者更简单——查询时在SQL里转换SELECT CAST(amount AS DOUBLE) as amount FROM orders返回的就是普通浮点数了。快速参考卡片连接pymysql.connect(host, port, user, password, database, charset) 查询cursor.execute(SQL, params) → cursor.fetchall() 插入cursor.execute(INSERT_SQL, values) → conn.commit() 更新cursor.execute(UPDATE_SQL, values) → conn.commit() 删除cursor.execute(DELETE_SQL, values) → conn.commit() 回滚conn.rollback() 关闭cursor.close() → conn.close() 占位符%s不是字符串格式化是参数化查询 金额字段DECIMAL(10,2)不要用FLOAT 每次操作完要commit出异常要rollback finally里必须close连接最后说一句数据库是RPA进阶的分水岭。掌握了数据库你就能处理百万级数据、做多系统数据同步、建自己的数据中台。影刀的Python节点让这件事的门槛降低了很多——你不需要成为DBA会用增删改查就够了。作者林焱
影刀RPA 数据库操作入门连接MySQL执行增删改查很多RPA初学者觉得数据库是程序员的事跟自己没关系。但等你开始做数据采集、报表汇总、多系统数据同步的时候就会发现Excel扛不住了——10万行数据Excel打开要半分钟筛选一下又卡半天。这时候数据库就是必经之路。这篇文章从零开始讲清楚在影刀RPA里怎么用Python节点操作MySQL数据库。不要求你有数据库基础跟着走就能跑通。什么情况需要用数据库三条判断标准中了任意一条就该切数据库了1. 单次处理数据超过5万行。Excel的上限不是104万行而是操作体验的上限。超过5万行的Excel筛选、排序、公式计算都会明显变慢。而数据库里的百万级数据查询还是秒级。2. 多个流程需要读写同一份数据。A流程采集数据往里写B流程读数据生成报表C流程根据数据状态触发动作。如果用Excel做中转文件锁、并发写入、数据覆盖这些问题全冒出来了。数据库天然支持并发和事务不用自己造轮子。3. 数据之间有复杂关联。“查询这个客户的最近10笔订单关联商品表和库存表算出总金额和库存状态”——这种查询在Excel里得VLOOKUP套N层在数据库里一条SQL就搞定。但反过来如果你只是处理几百行数据、做简单的加加减减、或者数据只在一个流程里用数据库就是杀鸡用牛刀。Excel完全够用。前置准备安装MySQL驱动在影刀的Python节点里操作MySQL需要一个叫pymysql的库。它是纯Python实现不需要安装额外的系统依赖在影刀环境中可以直接用。但影刀默认环境没有装这个库。你需要先安装第一步找到影刀的Python环境路径。影刀安装目录下有一个python文件夹。在影刀里新建一个Python节点输入以下代码查看Python路径importsysprint(sys.executable)运行后会输出类似C:\Program Files\yingdao\python\python.exe的路径。第二步用pip安装pymysql。在电脑的命令行CMD里执行C:\Program Files\yingdao\python\python.exe -m pip install pymysql看到Successfully installed pymysql-1.x.x就装好了。如果安装报权限错误以管理员身份运行CMD再试。如果公司网络限制了pip可以下载pymysql的.whl文件离线安装拼多多店群自动化上架方案C:\Program Files\yingdao\python\python.exe -m pip install C:\downloads\PyMySQL-1.1.0-py3-none-any.whl连接数据库安装完成后在影刀的Python节点里写连接代码importpymysql# 创建连接connpymysql.connect(host127.0.0.1,# 数据库IP。本地就是127.0.0.1port3306,# MySQL默认端口userroot,# 用户名passwordyour_password,# 密码databaserpa_data,# 要连接的数据库名charsetutf8mb4# 字符集)# 创建游标操作数据库的手柄cursorconn.cursor()print(数据库连接成功)连接参数说明参数说明示例host数据库服务器地址本地127.0.0.1远程192.168.1.100port端口号MySQL默认3306user用户名rootpassword密码你的数据库密码database数据库名要操作的具体数据库charset字符集utf8mb4支持emoji纯中文用utf8也行连接中的常见报错“Can’t connect to MySQL server on ‘127.0.0.1’”MySQL服务没启动。在Windows服务里找到MySQL服务启动它。或者用命令行net start mysql。“Access denied for user ‘root’‘localhost’”用户名或密码错了。确认一下是不是连接的是正确的数据库——有些人电脑上装了多个MySQL实例。“Unknown database ‘rpa_data’”数据库不存在。先用Navicat、DBeaver这类工具创建一个或者用代码创建# 先不指定database连接后创建connpymysql.connect(host127.0.0.1,userroot,passwordyour_password)cursorconn.cursor()cursor.execute(CREATE DATABASE IF NOT EXISTS rpa_data CHARACTER SET utf8mb4)cursor.execute(USE rpa_data)增删改查四件套数据库操作的精髓就是四个字增、删、改、查。来逐个击破。建表第一次用数据库先建一张表。表相当于Excel的一个Sheet定义了每列叫什么、什么类型。create_table_sql CREATE TABLE IF NOT EXISTS orders ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT 主键ID, order_no VARCHAR(50) NOT NULL COMMENT 订单号, customer_name VARCHAR(100) COMMENT 客户名称, amount DECIMAL(10,2) COMMENT 金额, status VARCHAR(20) DEFAULT 待处理 COMMENT 订单状态, created_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT订单表 cursor.execute(create_table_sql)conn.commit()print(表创建成功)几个常用字段类型INT整数比如IDVARCHAR(N)字符串最多N个字符DECIMAL(10,2)精确小数总共10位小数点后2位。金额字段必须用这个不要用FLOAT浮点数有精度问题DATETIME日期时间TEXT长文本插入数据增# 插入单条insert_sql INSERT INTO orders (order_no, customer_name, amount, status) VALUES (%s, %s, %s, %s) cursor.execute(insert_sql,(ORD001,张三贸易,3580.50,待处理))conn.commit()print(f插入成功新ID{cursor.lastrowid})# 批量插入从影刀变量读取数据# 假设 orders_data 是影刀传入的订单列表orders_data[(ORD002,李四科技,12000.00,待处理),(ORD003,王五百货,560.80,待处理),(ORD004,赵六建材,8920.00,待处理),]cursor.executemany(insert_sql,orders_data)conn.commit()print(f批量插入{cursor.rowcount}条)关键点%s是占位符不是Python的字符串格式化。绝对不要用 f-string 拼接SQL会有SQL注入风险# 错误不要这样写cursor.execute(fINSERT INTO orders VALUES ({order_no}))# 正确用参数化查询cursor.execute(INSERT INTO orders (order_no) VALUES (%s),(order_no,))查询数据查# 查询全部cursor.execute(SELECT * FROM orders)all_orderscursor.fetchall()print(f共{len(all_orders)}条记录)# 条件查询cursor.execute(SELECT order_no, customer_name, amount FROM orders WHERE status %s,(待处理,))pending_orderscursor.fetchall()forrowinpending_orders:print(f订单号{row[0]}客户{row[1]}金额¥{row[2]})# 带排序和分页cursor.execute( SELECT order_no, customer_name, amount, created_time FROM orders WHERE amount %s ORDER BY created_time DESC LIMIT %s OFFSET %s ,(1000,20,0))# 金额1000的订单前20条fetchall()返回所有结果fetchone()返回一条fetchmany(N)返回N条。更新数据改# 更新单条cursor.execute(UPDATE orders SET status %s WHERE order_no %s,(已完成,ORD001))conn.commit()print(f更新了{cursor.rowcount}条记录)# 批量更新循环中要小心fororderinto_update:cursor.execute(UPDATE orders SET status %s WHERE order_no %s,(order[new_status],order[order_no]))conn.commit()# 循环结束统一提交不要每条都commit注意UPDATE 不带 WHERE 条件会把整张表的数据全改了。这是数据库操作里最致命的翻车场景之一。写UPDATE之前先SELECT确认一下条件对不对。删除数据删cursor.execute(DELETE FROM orders WHERE order_no %s,(ORD001,))conn.commit()print(f删除了{cursor.rowcount}条记录)DELETE比UPDATE更危险。建议先用SELECT确认要删的数据然后把SELECT的WHERE条件原封不动复制到DELETE里# 步骤1先看看要删什么cursor.execute(SELECT * FROM orders WHERE status %s,(已取消,))to_deletecursor.fetchall()print(f将要删除{len(to_delete)}条记录)forrowinto_delete:print(row)# 步骤2确认无误后再执行删除cursor.execute(DELETE FROM orders WHERE status %s,(已取消,))conn.commit()print(f实际删除了{cursor.rowcount}条)数据库操作的正确流程框架影刀里操作数据库建议遵循这个框架Python节点 - 数据库操作的标准写法 1. 建立连接 2. try: 3. 执行业务操作增删改查 4. conn.commit() ← 提交事务 5. except Exception as e: 6. conn.rollback() ← 出错回滚 7. 记录错误信息 8. finally: 9. cursor.close() ← 关闭游标 10. conn.close() ← 关闭连接完整代码importpymysql connNonecursorNonetry:connpymysql.connect(host127.0.0.1,port3306,userroot,passwordyour_password,databaserpa_data,charsetutf8mb4)cursorconn.cursor()# 业务操作 cursor.execute(SELECT COUNT(*) FROM orders WHERE status 待处理)countcursor.fetchone()[0]print(f待处理订单数{count})# 如果count 0批量更新ifcount0:cursor.execute(UPDATE orders SET status 处理中 WHERE status 待处理)print(f更新了{cursor.rowcount}条订单状态)conn.commit()print(操作完成事务已提交)exceptpymysql.Errorase:ifconn:conn.rollback()print(f数据库操作失败{e})# 把错误信息传回影刀变量供后续判断error_msgstr(e)finally:ifcursor:cursor.close()ifconn:conn.close()print(数据库连接已关闭)为什么连接一定要关MySQL有最大连接数限制默认151。每个Python节点执行完不关连接连接数慢慢就耗光了。后续流程再想连数据库就会报Too many connections。影刀变量与数据库的交互数据在影刀变量和数据库之间来回传递这是最常见的需求。关键是做好格式转换。数据库查询结果 → 影刀变量影刀的Python节点可以定义输出变量。查询结果转成字典列表传回影刀# 查询结果转成字典列表方便影刀后续处理cursor.execute(SELECT id, order_no, customer_name, amount, status FROM orders)columns[id,order_no,customer_name,amount,status]result_list[]forrowincursor.fetchall():result_list.append(dict(zip(columns,row)))# result_list 是输出变量影刀可以直接用# result_list [# {id: 1, order_no: ORD001, customer_name: 张三, amount: 3580.50, status: 待处理},# {id: 2, order_no: ORD002, customer_name: 李四, amount: 12000.00, status: 待处理},# ]转过以后在影刀里result_list就是标准的列表字典结构可以ForEach遍历、可以取字段值跟处理Excel数据一模一样。影刀变量 → 数据库写入反过来影刀采集到的数据写入数据库TEMU店群如何管理运营# data_from_yingdao 是从影刀传入的变量列表字典data_from_yingdao[{order_no:ORD005,customer:新客户A,amount:5000.00},{order_no:ORD006,customer:新客户B,amount:7800.00},]# 转成元组列表用于executemanyinsert_data[(item[order_no],item[customer],item[amount])foritemindata_from_yingdao]cursor.executemany(INSERT INTO orders (order_no, customer_name, amount) VALUES (%s, %s, %s),insert_data)conn.commit()用json中转复杂数据如果数据结构太复杂嵌套很深可以用json序列化中转importjson# 复杂数据存为JSON字符串complex_data{order_no:ORD007,items:[{name:商品A,qty:2,price:100},{name:商品B,qty:1,price:500},],discount:0.9,total:630}cursor.execute(INSERT INTO orders (order_no, detail_json) VALUES (%s, %s),(ORD007,json.dumps(complex_data,ensure_asciiFalse)))conn.commit()# 读取时反序列化cursor.execute(SELECT detail_json FROM orders WHERE order_no ORD007)rowcursor.fetchone()detailjson.loads(row[0])print(detail[items][0][name])# 输出商品A这个技巧在存采集日志、复杂配置、嵌套业务数据时特别有用。四个踩坑实录坑1字符编码导致中文乱码数据写进去看起来正常读出来全是问号或乱码。根因是字符集没设对。MySQL默认字符集可能是latin1不支持中文。解决连接时指定charsetutf8mb4建表时指定DEFAULT CHARSETutf8mb4如果数据库已经建好了执行ALTER DATABASE rpa_data CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci坑2连接没关导致Too many connections流程跑了几轮后突然报这个错。排查后发现每次Python节点执行完没有cursor.close()和conn.close()。连接池不是你关不关闭都能自动回收的——MySQL服务端要等到超时才会释放闲置连接默认8小时。如果你一小时跑100次流程每次开一个新连接不关100个连接就占满了。铁律用 try-finally 确保连接一定关闭写成肌肉记忆。坑3SQL注入风险从影刀变量里拼SQL字符串遇上用户输入里带单引号就直接崩nameOBrien# 名字里有个单引号# 错误写法sqlfSELECT * FROM orders WHERE customer_name {name}# 生成SELECT * FROM orders WHERE customer_name OBrien# SQL语法错误单引号提前闭合了永远用参数化查询%s占位符不要字符串拼接。pymysql会自动处理特殊字符的转义。坑4Decimal类型在JSON序列化时报错amount字段是DECIMAL(10,2)pymysql读出来是Decimal(3580.50)类型。如果直接json.dumps()会报Object of type Decimal is not JSON serializable。fromdecimalimportDecimalimportjsonclassDecimalEncoder(json.JSONEncoder):defdefault(self,obj):ifisinstance(obj,Decimal):returnfloat(obj)returnsuper().default(obj)# 使用自定义编码器json.dumps(data,clsDecimalEncoder)或者更简单——查询时在SQL里转换SELECT CAST(amount AS DOUBLE) as amount FROM orders返回的就是普通浮点数了。快速参考卡片连接pymysql.connect(host, port, user, password, database, charset) 查询cursor.execute(SQL, params) → cursor.fetchall() 插入cursor.execute(INSERT_SQL, values) → conn.commit() 更新cursor.execute(UPDATE_SQL, values) → conn.commit() 删除cursor.execute(DELETE_SQL, values) → conn.commit() 回滚conn.rollback() 关闭cursor.close() → conn.close() 占位符%s不是字符串格式化是参数化查询 金额字段DECIMAL(10,2)不要用FLOAT 每次操作完要commit出异常要rollback finally里必须close连接最后说一句数据库是RPA进阶的分水岭。掌握了数据库你就能处理百万级数据、做多系统数据同步、建自己的数据中台。影刀的Python节点让这件事的门槛降低了很多——你不需要成为DBA会用增删改查就够了。作者林焱
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
