MYSQL 从项目经理的一次查询, 到PYTHON 解决问题(2) --传统企业使用MYSQL的问题

import configparserimport pymysql
import time
import asyncio  #标准库异步线程的库,协程,与多并发不同的是这个是单线程的
from aiomysql import create_pool  #第三方库,为MYSQL 来进行的异步线程操作库




class Solution:  #定义类
    def __init__(self, sql1, sql2):  #初始化参数
        config = configparser.ConfigParser()
        config.read('config.ini')   #读取配置文件
        self.host = config.get('config', 'host')
        self.user = config.get('config', 'user')
        self.password = config.get('config', 'password')
        self.database = config.get('config', 'database')
        self.save_database = config.get('config', 'save_database')
        self.save_user = config.get('config', 'save_user')
        self.save_password = config.get('config', 'save_password')

        self.conn = pymysql.connect(host=self.host, user=self.user, password=self.password, database=self.database,
                                    charset='utf8mb4')  #
        self.sql1 = sql1  #定义两个SQL
        self.sql2 = sql2

        self.task_num = 300 #异步并发数量, 一次可以干300个事务
              self.pool_size = 100 #连接池size读写各100
    def get_ids(self, ):
        cursor = self.conn.cursor()
        cursor.execute(self.sql1)
        self.data_a_full = cursor.fetchall()
        self.data_a_full_len = len(self.data_a_full)    # 获取查询的COUNT

    def run(self, ):

        self.get_ids()
        cur_loop = asyncio.get_event_loop()  #获取一个事件的循环
        cur_loop.run_until_complete(self.start(cur_loop))  

    async def start(self, loop):
        pool = await create_pool(host=self.host, port=3306,
                                 user=self.user, password=self.password,
                                 db=self.database, loop=loop, maxsize=self.pool_size)
        save_pool = await create_pool(host=self.host, port=3306,
                                      user=self.save_user, password=self.save_password,
                                      db=self.save_database, loop=loop, maxsize=self.pool_size)
        i = 0
        while True:
            start = time.time()
            m = self.data_a_full[i:i + self.task_num]   
            i += self.task_num
            tasks = []
            if len(m) == 0:    #每次给300个事务,直到这个池里面的没有数据取出
                break   
            for s in m:
                # print(s)
                sql = f"{sql2}'{s[0]}'"   #PYTHON3.X新特性简写,拼接SQL
                c1 = self.select(loop=loop, sql=sql, pool=pool, save_pool=save_pool, data=s)
                tasks.append(c1)   #执行SQL 语句
            await asyncio.gather(*tasks)
            print(time.time() - start,'----------',i,'/',self.data_a_full_len)

    async def select(self, loop, sql, pool, save_pool, data):  #处理的业务逻辑
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute(sql)
                r = await cur.fetchall()
                table_name_dict = {
                    'F': 'tb_f',
                    'T': 'tb_t',
                    'D': 'tb_d',
                }
                sum_dict = {
                    'F': 0, 'T': 0, 'D': 0
                }
                for item in r:
                    if item[1] is not None:
                        sum_dict[item[1]] = sum_dict.get(item[1]) + item[0]
                data = list(data)
                sql_list = []
                for key,value in sum_dict.items():
                    table_name = table_name_dict.get(key)
                    new_data = data+[value]
                    new_data[2] = str(new_data[2])
                    if new_data[4] is not None:
                        new_data[4] = str(float(new_data[4]))
                    else:
                        new_data[4] = 0
                    new_data[3] = str(new_data[3])
                    if table_name:   #结果插入到MYSQL数据库中
                        insert_sql = "insert into {} (CONTRACTNO,APPLYNO,ACTIVEDATE,term,AMORTIZEAMT) values ('{}','{}','{}',{},{})".format(
                            table_name, *new_data)
                        sql_list.append(insert_sql)
                    sql_do = ';'.join(sql_list)
                    # print(sql_do)
                async with save_pool.acquire() as save_conn:
                    async with save_conn.cursor() as save_cur:
                        await save_cur.execute(sql_do)
                        await save_conn.commit()



if __name__ == '__main__':
    s = time.time()
    sql1 = "select CONTRACTNO,APPLYNO,ACTIVEDATE,term from tb_sync_contract where ACTIVEDATE>='2020-07-01'"
    sql2 = "select AMORTIZEAMT,SAP_POSTING_IND from tb_amortize where CAMAINID="
    solution = Solution(sql1, sql2)
    solution.run()
print(time.time() - s)