影刀RPA 数据库连接与操作：MySQL、SQLite基础

影刀RPA 数据库连接与操作MySQL、SQLite基础作者林焱很多RPA场景需要和数据库打交道采集的数据要存入数据库、从数据库读取配置、把Excel数据导入数据库、生成数据报表。这篇文章不讲数据库理论只讲实操——影刀RPA里怎么连MySQL和SQLite怎么增删改查怎么处理常见报错。一、什么情况用什么场景推荐数据库说明单机流程数据量小SQLite零配置一个文件就是数据库团队共享数据MySQL支持多用户并发访问采集结果存储MySQL/SQLite看数据量和是否需要共享配置管理SQLite轻量不需要单独部署大数据量分析MySQL pandas数据库存储pandas分析历史数据归档SQLite压缩成单个文件方便备份二、怎么做2.1 SQLite操作零配置SQLite是Python内置的不需要安装任何东西一个.db文件就是一个完整数据库。连接和建表importsqlite3importosdefinit_sqlite_db(db_path):初始化SQLite数据库# 确保目录存在os.makedirs(os.path.dirname(db_path),exist_okTrue)connsqlite3.connect(db_path)cursorconn.cursor()# 创建商品表cursor.execute( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, price REAL, stock INTEGER DEFAULT 0, url TEXT, collect_time TEXT, UNIQUE(url) -- URL唯一防止重复插入 ) )# 创建采集日志表cursor.execute( CREATE TABLE IF NOT EXISTS collect_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, flow_name TEXT, start_time TEXT, end_time TEXT, status TEXT, record_count INTEGER, error_msg TEXT ) )conn.commit()conn.close()print(f数据库初始化完成:{db_path})# 初始化# init_sqlite_db(rD:\data\rpa.db)插入数据importsqlite3fromdatetimeimportdatetimedefinsert_products(db_path,products):批量插入商品数据connsqlite3.connect(db_path)cursorconn.cursor()nowdatetime.now().strftime(%Y-%m-%d %H:%M:%S)# 使用INSERT OR IGNORE遇到重复URL跳过sql INSERT OR IGNORE INTO products (name, price, stock, url, collect_time) VALUES (?, ?, ?, ?, ?) data[(p[name],p[price],p[stock],p[url],now)forpinproducts[video(video-F8vXzCHC-1784265031841)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限)]]cursor.executemany(sql,data)conn.commit()insertedcursor.rowcountprint(f插入{inserted}条记录共{len(data)}条跳过{len(data)-inserted}条重复)conn.close()returninserted# 使用示例# products [# {name: 商品A, price: 99.5, stock: 100, url: https://example.com/1},# {name: 商品B, price: 199.0, stock: 50, url: https://example.com/2},# ]# insert_products(rD:\data\rpa.db, products)查询数据importsqlite3defquery_products(db_path,min_price0,limit100):查询商品数据connsqlite3.connect(db_path)conn.row_factorysqlite3.Row# 让查询结果可以用列名访问cursorconn.cursor()cursor.execute(SELECT * FROM products WHERE price ? ORDER BY price DESC LIMIT ?,(min_price,limit))rowscursor.fetchall()# 转为字典列表results[dict(row)forrowinrows]conn.close()print(f查询到{len(results)}条记录)returnresults# data query_products(rD:\data\rpa.db, min_price50)# for item in data:# print(f{item[name]}: ¥{item[price]} (库存: {item[stock]}))更新和删除importsqlite3defupdate_product(db_path,url,new_price,new_stock):更新商品信息connsqlite3.connect(db_path)cursorconn.cursor()cursor.execute(UPDATE products SET price ?, stock ? WHERE url ?,(new_price,new_stock,url))conn.commit()affectedcursor.rowcount conn.close()print(f更新{affected}条记录)returnaffecteddefdelete_old_data(db_path,days30):删除N天前的旧数据connsqlite3.connect(db_path)cursorconn.cursor()cursor.execute(DELETE FROM products WHERE collect_time datetime(now, ?),(f-{days}days,))conn.commit()deletedcursor.rowcount conn.close()print(f删除{deleted}条过期记录)returndeleted2.2 MySQL操作MySQL需要安装mysql-connector-python或pymysql# 安装pip install mysql-connector-python连接MySQLimportmysql.connectorfrommysql.connectorimportErrordefcreate_mysql_connection(host,port,database,user,password):创建MySQL连接try:connmysql.connector.connect(hosthost,portport,databasedatabase,useruser,passwordpassword,charsetutf8mb4,# 支持emoji等4字节字符connect_timeout10)ifconn.is_connected():print(fMySQL连接成功:{host}:{port}/{database})returnconnexceptErrorase:print(fMySQL连接失败:{e})returnNone# conn create_mysql_connection(# host192.168.1.100,# port3306,# databaserpa_data,# userrpa_user,# passwordyour_password# )MySQL增删改查importmysql.connectordefmysql_insert_batch(conn,table,columns,data):批量插入MySQLcursorconn.cursor()# 构建SQLplaceholders, .join([%s]*len(columns))columns_str, .join(columns)sqlfINSERT IGNORE INTO{table}({columns_str}) VALUES ({placeholders})try:cursor.executemany(sql,data)conn.commit()print(f插入{cursor.rowcount}条记录)returncursor.rowcountexceptExceptionase:conn.rollback()print(f插入失败:{e})return0finally:cursor.close()defmysql_query_to_list(conn,sql,paramsNone):查询MySQL并返回字典列表cursorconn.cursor(dictionaryTrue)try:cursor.execute(sql,paramsor())resultscursor.fetchall()print(f查询到{len(results)}条记录)returnresultsexceptExceptionase:print(f查询失败:{e})return[]finally:cursor.close()# 使用示例# conn create_mysql_connection(...)## # 插入# mysql_insert_batch(conn, products,# [name, price, stock, url],# [(商品A, 99.5, 100, url1), (商品B, 199.0, 50, url2)]# )## # 查询# results mysql_query_to_list(conn,# SELECT * FROM products WHERE price %s ORDER BY price DESC LIMIT 10,# (50,)# )## conn.close()2.3 数据库连接管理频繁开关数据库连接效率低而且可能导致连接数耗尽。用连接池importsqlite3importthreadingclassSQLiteConnectionPool:SQLite连接池简化版def__init__(self,db_path,pool_size5):self.db_pathdb_path self.pool_sizepool_size self.pool[]self.lockthreading.Lock()# 初始化连接池for_inrange(pool_size):connsqlite3.connect(db_path,check_same_threadFalse)conn.row_factorysqlite3.Row self.pool.append(conn)defget_connection(self):获取连接withself.lock:ifself.pool:returnself.pool.pop()# 池空了新建连接connsqlite3.connect(self.db_path,check_same_threadFalse)conn.row_factorysqlite3.Rowreturnconndefreturn_connection(self,conn):归还连接withself.lock:iflen(self.pool)self.pool_size:self.pool.append(conn)else:conn.close()defclose_all(self):关闭所有连接withself.lock:forconninself.pool:conn.close()self.pool.clear()print(所有数据库连接已关闭)# 使用示例# pool SQLiteConnectionPool(rD:\data\rpa.db)# conn pool.get_connection()# try:# cursor conn.cursor()# cursor.execute(SELECT COUNT(*) FROM products)# count cursor.fetchone()[0]# print(f总记录数: {count})# finally:# pool.return_connection(conn)# pool.close_all()2.4 Excel数据导入数据库importsqlite3importopenpyxlfromdatetimeimportdatetimedefexcel_to_sqlite(excel_path,db_path,table_name,sheet_nameNone):将Excel数据导入SQLite# 读取Excelwbopenpyxl.load_workbook(excel_path,read_onlyTrue)wswb[sheet_name]ifsheet_nameelsewb.active# 获取表头headers[]forcellinnext(ws.iter_rows(min_row1,max_row1)):headers.append(str(cell.value).strip()ifcell.valueelsefcol_{cell.column})# 获取数据data[]forrowinws.iter_rows(min_row2,values_onlyTrue):ifany(cellisnotNoneforcellinrow):data.append(tuple(row[:len(headers)]))wb.close()print(fExcel读取完成:{len(headers)}列,{len(data)}行)# 写入数据库connsqlite3.connect(db_path)cursorconn.cursor()# 自动建表columns_def, .join([f{h} TEXTforhinheaders])cursor.execute(fCREATE TABLE IF NOT EXISTS{table_name}({columns_def}))# 批量插入placeholders, .join([?]*len(headers))columns_str, .join([f{h}forhinheaders])sqlfINSERT INTO{table_name}({columns_str}) VALUES ({placeholders})cursor.executemany(sql,data)conn.commit()print(f导入成功:{cursor.rowcount}条记录 →{table_name})conn.close()returncursor.rowcount# excel_to_sqlite(rD:\data\products.xlsx, rD:\data\rpa.db, imported_products)2.5 数据库导出到Excelimportsqlite3importopenpyxlfromdatetimeimportdatetimedefsqlite_to_excel(db_path,sql,excel_path,sheet_name数据):将数据库查询结果导出到Excelconnsqlite3.connect(db_path)conn.row_factorysqlite3.Row cursorconn.cursor()cursor.execute(sql)rowscursor.fetchall()ifnotrows:print(查询结果为空)conn.close()returnFalse# 获取列名columns[desc[0]fordescincursor.description]# 写入Excelwbopenpyxl.Workbook()wswb.active ws.titlesheet_name# 写表头forcol_idx,col_nameinenumerate(columns,1):ws.cell(row1,columncol_idx,valuecol_name)ws.cell(row1,columncol_idx).fontopenpyxl.styles.Font(boldTrue)# 写数据forrow_idx,rowinenumerate(rows,2):forcol_idx,col_nameinenumerate(columns,1):ws.cell(rowrow_idx,columncol_idx,valuerow[col_name])# 自适应列宽forcolinws.columns:max_lengthmax(len(str(cell.valueor))forcellincol)ws.column_dimensions[col[0].column_letter].widthmin(max_length2,50)wb.save(excel_path)conn.close()print(f导出成功:{len(rows)}行 →{excel_path})returnTrue# sqlite_to_excel(# rD:\data\rpa.db,# SELECT * FROM products WHERE price 50 ORDER BY price DESC,# rD:\report\products_export.xlsx# )三、有什么坑坑1忘记关闭数据库连接连接打开后忘记关闭长期运行会导致连接数耗尽。解决用try...finally确保连接关闭connNonetry:connsqlite3.connect(db_path)# 执行操作passfinally:ifconn:conn.close()坑2SQL注入风险直接拼接SQL字符串有注入风险。解决永远用参数化查询# 错误字符串拼接sqlfSELECT * FROM users WHERE name {user_input}# 正确参数化查询sqlSELECT * FROM users WHERE name ?cursor.execute(sql,(user_input,))坑3SQLite并发写入冲突SQLite不支持多进程同时写入。多进程同时写会报 “database is locked”。temu店群自动化报活动案例解决单进程写多进程读或者用WAL模式增加并发性connsqlite3.connect(db_path)conn.execute(PRAGMA journal_modeWAL)# 启用WAL模式坑4MySQL中文乱码MySQL连接时没指定charset中文数据写入后变成问号。解决连接时指定charsetutf8mb4不是utf8utf8只支持3字节不支持emojiconnmysql.connector.connect(hosthost,databasedatabase,useruser,passwordpassword,charsetutf8mb4# 必须设置)坑5大批量插入太慢一条一条插入1万条数据要好几分钟。解决用executemany批量插入或者在事务中插入# 批量插入推荐cursor.executemany(sql,data_list)conn.commit()# 或者分批提交batch_size1000foriinrange(0,len(data_list),batch_size):cursor.executemany(sql,data_list[i:ibatch_size])conn.commit()print(f已插入{ibatch_size}/{len(data_list)})总结数据库操作的核心要点小数据用SQLite——零配置一个文件搞定参数化查询防注入——永远不要拼接SQL字符串连接用完要关闭——用try…finally确保批量插入用executemany——比逐条快几十倍MySQL指定utf8mb4——防止中文和emoji乱码