python 链接mysql数据库查询数据导出成表格，自动将数据库字段名写入表格列名-数据库层级真分页，多次查询，再写入表格（可支持多线程）

# -*- ecoding: utf-8 -*-
# @ModuleName: test
# @Funcation: 
# @Author: darling
# @Time: 2022-07-28 15:29
import datetime
import math
import os
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor

import openpyxl
import pymysql
from loguru import logger


def client_database(sql):
    # 打开数据库连接
    db = pymysql.connect(host="127.0.0.1", user="root", password="123", db="test",
                         port=3306)
    # 使用 cursor() 方法创建一个游标对象 cursor
    cursor = db.cursor()
    # 使用 execute()  方法执行 SQL 查询
    cursor.execute(sql)
    # 使用 fetchall() 方法获取s所有数据.
    datas = cursor.fetchall()
    # 表头字段名
    fileds = [filed[0] for filed in cursor.description]
    db.close()
    return fileds, list(datas)


def get_sql_count(sql) -> int:
    # 打开数据库连接
    db = pymysql.connect(host="127.0.0.1", user="root", password="123", db="test",
                         port=3306)

    # 使用 cursor() 方法创建一个游标对象 cursor
    cursor = db.cursor()
    # 使用 execute()  方法执行 SQL 查询
    count = cursor.execute(sql)
    db.close()
    return count


def check_file_type(f_path):
    txt = re.compile(r'\.xlsx$')
    if txt.search(f_path) is None:
        exit_sys('文件格式格式不正确，只允许xlsx文件')
    if fileIsOpen(f_path):
        logger.info('============={}：文件被占用，无法写入，请关闭文件=========', f_path)
        exit_sys('文件被占用，请关闭 ' + f_path)


def exit_sys(msg):
    logger.info(msg)
    logger.info('程序退出··· ···')
    time.sleep(3)
    sys.exit()


# 判定文件是否打开
def fileIsOpen(filepath):
    filef = os.path.split(filepath)  # 文件路径和文件名拆开
    excelname = filef[-1]
    excelpath = filef[0]
    hidefilename = excelpath + r"/~$" + excelname  # 拼接出隐藏文件的文件路径
    if os.path.exists(hidefilename):
        return True
    else:
        return False


def exists_file(file_name):
    return os.path.exists(file_name)


# 判断文件是否存在
def sheet_method(file_path, add_sheet, sheet_num=1):
    if not exists_file(file_path):
        wk = openpyxl.Workbook()
    else:
        wk = openpyxl.load_workbook(file_path)
    index = sheet_num - 1
    add_sheet = add_sheet + str(sheet_num)
    if add_sheet in wk.sheetnames:  # 表格如果已经存在，那就删除，在重新创建，清空历史数据
        del wk[add_sheet]
    if 'Sheet' in wk.sheetnames:  # 将默认的sheet表格删除
        del wk['Sheet']
    wk.create_sheet(title=add_sheet, index=index)
    wk.save(file_path)


# 保存数据到多个文件
def save_excel_to_file(fileds, datas, file_name, sheet_name, sheet_num=0):
    logger.info('我进来了save方法')
    file_name = file_name.replace(f'.xlsx', f'-{sheet_num}.xlsx')
    # 判断文件是否存在
    sheet_method(file_name, sheet_name, sheet_num)
    wb = openpyxl.load_workbook(file_name)  # 打开文件
    ws = wb.active
    logger.info('总数据：{} 行', len(datas))
    # 写入表格第一行表头
    i = 1
    leng = len(fileds)
    while i <= leng:
        ws.cell(1, i, fileds[i - 1])
        i = i + 1
    # 写入表格内容
    j = 1
    # 循环行
    for data in datas:
        di = 1
        j = j + 1
        # 循环列
        for k in range(len(data)):
            ws.cell(j, di, data[k])
            di = di + 1
    wb.save(file_name)
    logger.info('保存文件{}', file_name)


# 保存数据到一个文件，多个sheet页签
def save_excel_to_sheets(fileds, datas, file_name, sheet_name, sheet_num=0):
    logger.info('我进来了save方法')
    # 判断文件是否存在
    sheet_method(file_name, sheet_name, sheet_num)
    wb = openpyxl.load_workbook(file_name)  # 打开文件
    ws = wb[sheet_name + str(sheet_num)]
    logger.info('总数据：{} 行', len(datas))
    # 写入表格第一行表头
    i = 1
    leng = len(fileds)
    while i <= leng:
        ws.cell(1, i, fileds[i - 1])
        i = i + 1
    # 写入表格内容
    j = 1
    # 循环行
    for data in datas:
        di = 1
        j = j + 1
        # 循环列
        for k in range(len(data)):
            ws.cell(j, di, data[k])
            di = di + 1
    wb.save(file_name)
    logger.info('保存文件{}', file_name)


def exec_sql_query(sql_t, numb):
    logger.info('我进来了线程执行的方法{}', sql_t)
    logger.info(numb)
    sql_res = client_database(sql_t)
    save_excel_to_file(fileds=sql_res[0], datas=sql_res[1], file_name=fileName, sheet_name='Sheet', sheet_num=numb)
    # 执行保存多个sheet的方法
    # save_excel_to_sheets(fileds=sql_res[0], datas=sql_res[1], file_name=fileName, sheet_name='Sheet', sheet_num=numb)


def query_sql_to_excel_for_sql_page(sql):
    page_num2 = 1000
    check_file_type(f_path=fileName)
    # 获取总行数
    sql = 'SELECT * FROM ({0}) temp'.format(sql)
    count = get_sql_count(sql)
    # 页码向上取整
    page_size = math.ceil(count / page_num2)
    for page_num1 in range(page_size):
        sql_t = '{0} limit {1},{2}'.format(sql, page_num1 * page_num2, page_num2)
        exec_sql_query(sql_t, page_num1 + 1)


def query_sql_to_excel_for_thread(sql):
    page_num2 = 2000
    check_file_type(f_path=fileName)
    # 获取总行数
    sql = 'SELECT * FROM ({0}) temp'.format(sql)
    count = get_sql_count(sql)
    # 页码向上取整
    page_size = math.ceil(count / page_num2)
    with ThreadPoolExecutor(50) as threadPool:
        for page_num1 in range(page_size):
            sql_t = '{0} limit {1},{2}'.format(sql, page_num1 * page_num2, page_num2)
            threadPool.submit(lambda p: exec_sql_query(*p), [sql_t, page_num1 + 1])


if __name__ == '__main__':
    fileName = './{0}.xlsx'.format(str(datetime.datetime.now().strftime("%Y-%m-%d-%H%M%S")))
    sql = """ SELECT id as '唯一id',`name` as '名称',phone as '电话',address as '地址' FROM people_infomation limit 1000000"""
    query_sql_to_excel_for_sql_page(sql)
    # 多线程执行方法
    # query_sql_to_excel_for_thread(sql)
    # 多线程执行方法操作一个文件多个sheet页写入时，会有问题，在于无法同时读取 open同一个文件，相当于是占用无法写入吧。

标签：sheet,name,表格,数据库,写入,cursor,file,import,path
来源： https://www.cnblogs.com/darling331/p/16529609.html