math_dup_check/mysql_operate.py

import time
import pymysql
from config import mysqldb, log_msg

# 数据库查询函数
def mysql_operate(sql, insert_list=None, operate_flag=0):
    # 建立链接
    mysql_conn = pymysql.connect(
        host=mysqldb["host"],
        user=mysqldb["user"],
        password=mysqldb["password"],
        db=mysqldb["db"],
        charset=mysqldb["charset"],
        cursorclass=pymysql.cursors.DictCursor
    )
    # 查询返回字典
    res_dict = dict()
    # 创建游标对象cursor
    cursor = mysql_conn.cursor()
    # 判断mysql查询和插入/删除标志
    if operate_flag == 0:
        # 执行sql语句
        cursor.execute(sql)
        res_dict = cursor.fetchone()
    else:
        try:
            # 执行sql语句
            if insert_list is not None:
                cursor.executemany(sql, insert_list)
            else:
                cursor.execute(sql)
            # 提交事务
            mysql_conn.commit()
        except Exception as e:
            print(e)
            # 如果出现异常,回滚
            mysql_conn.rollback()
            print(log_msg.format(id=time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()),
                                 type="mysql操作",
                                 message="mysql数据操作失败"))
    # 关闭数据库连接
    cursor.close()
    mysql_conn.close()
    return res_dict