1. 项目背景与需求解析最近在整理公司业务数据时经常需要将数据库中的多张表数据导出到Excel文件。手动操作不仅效率低下还容易出错。于是我用Python开发了一个自动化脚本可以批量导出数据库数据到Excel文件支持MySQL、PostgreSQL等多种数据库。这个脚本特别适合需要定期备份数据、生成报表或者进行数据迁移的场景。比如财务部门每月需要导出销售数据运营团队要分析用户行为日志都可以用这个工具快速完成。2. 技术方案设计2.1 核心组件选择我选择了以下几个Python库来构建这个工具SQLAlchemy作为ORM框架统一不同数据库的操作接口pandas处理数据转换和Excel文件生成openpyxl增强Excel文件的格式化能力configparser读取数据库连接配置选择SQLAlchemy是因为它支持多种数据库后端只需要更换连接字符串就能切换数据库类型。pandas的DataFrame非常适合做数据转换而且内置了to_excel方法。2.2 程序架构设计脚本的主要流程分为四个步骤读取配置文件建立数据库连接查询指定表的数据将数据转换为Excel格式保存到指定路径为了提高灵活性我设计了以下功能支持导出单表或多表可自定义查询条件可设置Excel样式和格式支持增量导出3. 核心代码实现3.1 数据库连接管理from sqlalchemy import create_engine import configparser def get_db_engine(config_file): config configparser.ConfigParser() config.read(config_file) db_type config.get(database, type) if db_type mysql: conn_str fmysqlpymysql://{config.get(database, user)}:{config.get(database, password)}{config.get(database, host)}:{config.get(database, port)}/{config.get(database, name)} elif db_type postgresql: conn_str fpostgresql://{config.get(database, user)}:{config.get(database, password)}{config.get(database, host)}:{config.get(database, port)}/{config.get(database, name)} return create_engine(conn_str)3.2 数据查询与导出import pandas as pd from openpyxl import load_workbook from openpyxl.styles import Font, Alignment def export_to_excel(engine, table_name, output_file, queryNone): # 读取数据 if query: df pd.read_sql(query, engine) else: df pd.read_sql_table(table_name, engine) # 导出到Excel if not os.path.exists(output_file): df.to_excel(output_file, indexFalse) else: with pd.ExcelWriter(output_file, engineopenpyxl, modea) as writer: df.to_excel(writer, sheet_nametable_name, indexFalse) # 设置Excel样式 wb load_workbook(output_file) ws wb.active for col in ws.columns: max_length 0 column col[0].column_letter for cell in col: cell.font Font(nameArial, size10) cell.alignment Alignment(horizontalleft) try: if len(str(cell.value)) max_length: max_length len(str(cell.value)) except: pass adjusted_width (max_length 2) * 1.2 ws.column_dimensions[column].width adjusted_width wb.save(output_file)4. 使用示例4.1 基本用法# 配置文件db.ini内容 [database] type mysql host localhost port 3306 name test_db user root password 123456 # 导出单个表 engine get_db_engine(db.ini) export_to_excel(engine, users, output.xlsx) # 导出多个表 tables [products, orders, customers] for table in tables: export_to_excel(engine, table, output.xlsx)4.2 高级用法# 带条件查询导出 query SELECT * FROM orders WHERE order_date 2023-01-01 export_to_excel(engine, None, orders_2023.xlsx, queryquery) # 增量导出基于最后更新时间 last_export_time 2023-06-01 00:00:00 query fSELECT * FROM products WHERE update_time {last_export_time} export_to_excel(engine, None, products_update.xlsx, queryquery)5. 实用技巧与注意事项5.1 性能优化大数据量导出时建议分批查询chunk_size 10000 for chunk in pd.read_sql_table(table_name, engine, chunksizechunk_size): # 处理每个数据块关闭自动提交可以提升性能engine create_engine(conn_str, pool_pre_pingTrue, pool_recycle3600)5.2 常见问题解决中文乱码问题确保数据库连接字符串添加charsetutf8Excel文件保存时指定编码df.to_excel(output_file, indexFalse, encodingutf-8-sig)内存不足问题使用chunksize参数分批处理考虑使用dask替代pandas处理超大数据集日期格式问题# 在导出前统一格式化日期列 df[date_column] pd.to_datetime(df[date_column]).dt.strftime(%Y-%m-%d)5.3 扩展功能添加进度显示from tqdm import tqdm tables [table1, table2, table3] for table in tqdm(tables): export_to_excel(engine, table, output.xlsx)支持更多数据库elif db_type sqlite: conn_str fsqlite:///{config.get(database, path)} elif db_type oracle: conn_str foraclecx_oracle://{config.get(database, user)}:{config.get(database, password)}{config.get(database, host)}:{config.get(database, port)}/?service_name{config.get(database, service_name)}自动压缩输出文件import zipfile def compress_file(input_file, output_zip): with zipfile.ZipFile(output_zip, w, zipfile.ZIP_DEFLATED) as zipf: zipf.write(input_file, os.path.basename(input_file))6. 完整脚本示例import os import pandas as pd from sqlalchemy import create_engine from openpyxl import load_workbook from openpyxl.styles import Font, Alignment import configparser class DatabaseExporter: def __init__(self, config_file): self.config_file config_file self.engine self._get_db_engine() def _get_db_engine(self): config configparser.ConfigParser() config.read(self.config_file) db_type config.get(database, type) if db_type mysql: conn_str fmysqlpymysql://{config.get(database, user)}:{config.get(database, password)}{config.get(database, host)}:{config.get(database, port)}/{config.get(database, name)}?charsetutf8mb4 elif db_type postgresql: conn_str fpostgresql://{config.get(database, user)}:{config.get(database, password)}{config.get(database, host)}:{config.get(database, port)}/{config.get(database, name)} elif db_type sqlite: conn_str fsqlite:///{config.get(database, path)} return create_engine(conn_str, pool_pre_pingTrue, pool_recycle3600) def export_tables(self, tables, output_file, query_dictNone): if query_dict is None: query_dict {} for table in tables: query query_dict.get(table) self._export_single_table(table, output_file, query) def _export_single_table(self, table_name, output_file, queryNone): try: if query: df pd.read_sql(query, self.engine) else: df pd.read_sql_table(table_name, self.engine) if not os.path.exists(output_file): df.to_excel(output_file, indexFalse, encodingutf-8-sig) else: with pd.ExcelWriter(output_file, engineopenpyxl, modea) as writer: df.to_excel(writer, sheet_nametable_name, indexFalse) self._format_excel(output_file, table_name) print(f成功导出表 {table_name} 到 {output_file}) except Exception as e: print(f导出表 {table_name} 失败: {str(e)}) def _format_excel(self, file_path, sheet_name): wb load_workbook(file_path) ws wb[sheet_name] # 设置表头样式 for cell in ws[1]: cell.font Font(boldTrue) cell.alignment Alignment(horizontalcenter) # 自动调整列宽 for col in ws.columns: max_length 0 column col[0].column_letter for cell in col: try: if len(str(cell.value)) max_length: max_length len(str(cell.value)) except: pass adjusted_width (max_length 2) * 1.2 ws.column_dimensions[column].width adjusted_width wb.save(file_path) # 使用示例 if __name__ __main__: exporter DatabaseExporter(db.ini) tables [users, products, orders] exporter.export_tables(tables, output.xlsx)这个脚本在实际项目中已经稳定运行了半年多导出了超过100GB的业务数据。通过不断优化现在导出速度比最初版本提升了3倍以上。建议根据实际需求调整chunksize和连接池参数可以达到最佳性能。
