Python3.6 连接mysql 数据库，增删改查，及多线程简单运用

readme:

       导入 pymysql 连接数据库，完成数据处理后的增删改查操作。导入到其他Python文件就可以直接调用。后面一个文件是多线程操作， 另一个文件是处理曲线拟合和积分的然后数据和数据库交互的运用。

aliyunMySQL_test.py

import pymysql

class oilCheckMysql:
    def __init__(self,host,user,passwd,db,tableName):
        self.host = host
        self.user = user
        self.passwd =passwd
        self.db=db
        self.tableName =tableName

    def serachMySQL(self):
        #  db = pymysql.connect(host='**.**.**.**', user='root', passwd='*****', db="zabbix", charset="utf8") # lab420 校罐
        db = pymysql.connect(host=str(self.host),user=str(self.user), passwd = str(self.passwd), db=str(self.db), charset="utf8")
        cur = db.cursor()
        try:
            # 查询
            cur.execute("SELECT * FROM "+str(self.tableName))
            for row in cur.fetchall():
                #   print(row)
                id = row[0]

                startHeight= row[1]
                endHeight = row[2]
                oilV = row[3]
                calculateV = row[4]
                ratioV=row[5]
                print("id=%d ,startHeight=%f,endHeight =%f, oilV=%f, ratio=%f  ",
                    (id, startHeight, endHeight, oilV,calculateV,ratioV))
        except:
            print("MySQL Error: unable to fecth data of station_stock")
        cur.close()
        db.close()

    def insertMySQL(self,ID,CH1,CH2,CHV,calV,ratV):
        conn = pymysql.connect(host=str(self.host), user=str(self.user), passwd=str(self.passwd),
                               db=str(self.db),charset="utf8")
        cur = conn.cursor()
        try:
            sqli = "insert into "+str(self.tableName)+"(id,startHeight,endHeight,oilV,calculateV,ratioV) values(%s,%s,%s,%s,%s,%s)"
            param =(ID,CH1,CH2,CHV,calV,ratV)
            cur.execute(sqli, param)

            cur.close()
            conn.commit()
            conn.close()
        except:
            print("MySQL Error :unable to insert data")

    def updateMySQL(self,id,calV,ratV):
        conn = pymysql.connect(host=str(self.host), user=str(self.user), passwd=str(self.passwd),
                               db=str(self.db),charset="utf8")
        cur = conn.cursor()
        try:
         #   sqli = "update checkVolume set calculateV=23.34 ,ratioV=1.6  where id = 123"
          #  cur.execute(sqli)

            sqlit='update '+str(self.tableName)+' set calculateV='+str(calV)+',ratioV='+str(ratV)+'where id='+str(id)
            cur.execute(sqlit)
            cur.close()
            conn.commit()
            conn.close()
        except:
            print("MySQL Error :unable to update data")

    # delete a record
    def deleteByID(self, id):

        conn = pymysql.connect(host=str(self.host), user=str(self.user), passwd=str(self.passwd),
                               db=str(self.db),charset="utf8")
        cur = conn.cursor()
        try:
            sqlit='delete from '+ str(self.tableName) +' where id>'+ str(id)
            cur.execute(sqlit)
            cur.close()
            conn.commit()
            conn.close()
        except :
            print("MySQL Error :unable to delete data")

def main():
    CheckH1=[12.2,23.4]
    CheckH2=[10.8,23.3]
    CheckV=[13200.1,12500]
    CalculateV=[10020.4,13000]
    ratioV=[0.23,0.12]

    ckl =oilCheckMysql('***.**.***.***','root','123456',"oilCheck",'checkVolume')

    ckl.insertMySQL(39,CheckH1[0],CheckH2[0],CheckV[0],CalculateV[0],ratioV[0])
    ckl.updateMySQL(39,CalculateV[0],ratioV[0])
    ckl.serachMySQL()
    ckl.deleteByID(38)

if __name__ == '__main__':
    main()

import threading
from time import sleep,ctime

loops =(4,2)

class MyThread(threading.Thread):
    def __init__(self,func,args,name=''):
        threading.Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args
    def run(self):
        self.func(*self.args)


def loop(nloop,nsec):
    print('start loop',nloop, 'at:',ctime())
    sleep(nsec)
    print('loop',nloop,'don at',ctime())


def main():
    print('starting at:',ctime())
    threads =[]
    nloops=range(len(loops))

    for i in nloops:
        t =MyThread(loop,(i,loops[i]), loop.__name__)
        threads.append(t)

    for i in nloops:
        threads[i].start()

    for i in nloops:
        threads[i].join()
    print ('All Done at:',ctime())

if __name__=='__main__':
    main()

###最小二乘法试验###
import numpy as np
import pymysql
from scipy.optimize import leastsq
from scipy import integrate
###绘图，看拟合效果###
import matplotlib.pyplot as plt
from sympy import *

from aliyunMySQL_test import oilCheckMysql
from myThread import MyThread
from time import ctime,sleep



path='E:PythonProgramoilCheckingoildata.txt'

h_median =[]      # 高度相邻中位数
h_subtract =[]          #相邻高度作差
v_subtract =[]          #相邻体积作差
select_h_subtr =[]           #筛选的高度作差    ########
select_v_subtr =[]           #筛选的体积作差

VdtH=[]                     #筛选的V 和 t 的 倒数。

CheckH1 = []
CheckH2 = []
CheckH1_H2=[]
CheckV=[]
vCalcute = []
ratioV=[]

def loadData(Spath):
    try:
        with open(Spath,'r') as f0:
             for i in f0:
                tmp=i.split()
                h_median.append(float(tmp[0]))
                h_subtract.append(float(tmp[1]))
                v_subtract.append(float(tmp[2]))
        print ("source lie0",len(h_median))
    except:
        print("cannot open file!")


def removeBadPoint(h_med,h_sub,v_sub):
    for val in h_sub:
        position=h_sub.index(val)
        if 0.01 > val > -0.01:
            del h_sub[position]
            del h_med[position]
            del v_sub[position]
    for  vmin in v_sub:
         pos =v_sub.index(vmin)
         if 18 > vmin > -18:
             del h_med[pos]
             del h_sub[pos]
             del v_sub[pos]
         elif vmin > 85:
             del h_med[pos]
             del h_sub[pos]
             del v_sub[pos]
         else :
             pass
    v_dt_h_ay = [(y/x) for x, y in zip(h_sub, v_sub)]
    return v_dt_h_ay



def selectRightPoint(h_med,h_subtr,v_dt_h_ay):

    for val in v_dt_h_ay:
        pos = v_dt_h_ay.index(val)
        if val > 20 :
            del v_dt_h_ay[pos]
            del h_med[pos]
            del h_subtr[pos]
    for val in v_dt_h_ay:
        ptr = v_dt_h_ay.index(val)
        if val < 14:
            del v_dt_h_ay[ptr]
            del h_med[ptr]
            del h_subtr[ptr]


def writeFile(h_mp, v_dt_h):
    s='n'.join(str(num)[1:-1] for num in h_mp)
    v='n'.join(str(vdt)[1:-1] for vdt in v_dt_h)
    open(r'h_2.txt','w').write(s)
    open(r'v_dt_h.txt','w').write(v)
    print("write h_median: ",len(h_mp))
    # print("V_dt also is (y-x) : ",v_dt_h,end="n")
    print("Write V_dt_h : ",len(v_dt_h))

def readFileCheck():
    path2 = 'E:PythonProgramoilCheckingCheckingData1.txt'
    with open(path2, 'r') as f0:
        for i in f0:
            tmp = i.split()
            CheckH1.append(float(tmp[0]))
            CheckH2.append(float(tmp[1]))
            CheckV.append(float(tmp[2]))
    CheckH12 = [(y-x) for x, y in zip(CheckH1, CheckH2)]
    return CheckH12


def integralCalculate(coeff,CH1,CH2,chkv):
    x = Symbol('x')
    a, b, c, d = coeff[0]
    y = a * x ** 3 + b * x ** 2 + c * x + d
    i=0
    while (i< len(CH1)) :
        m = integrate(y, (x, CH1[i], CH2[i]))-80
        vCalcute.append(abs(m))
        i=i+1
    ratioV[:] = [(((y-x)/x)*100) for x,y in zip(chkv,vCalcute)]
    print("求导结果：",vCalcute)
    print("求导长度 len(VCalcute): ",len(vCalcute))
    print("V 误差百分比大小：",ratioV)
    file=open('ratio.txt','w')
    file.write(str(ratioV))
    file.close
    return vCalcute

 ###需要拟合的函数func及误差error###

def func(p,x):
    a,b,c,d=p
    return a*x**3+b*x**2+c*x+d #指明待拟合的函数的形状，设定的拟合函数。

def error(p,x,y):
    return func(p,x)-y #x、y都是列表，故返回值也是个列表

def leastSquareFitting(h_mp,v_dt_hl):
    p0=[1,2,6,10]     #a,b,c 的假设初始值，随着迭代会越来越小
    #print(error(p0,h_mp,v_dt_h,"cishu"))  #目标是让error 不断减小
    #s="Test the number of iteration" #试验最小二乘法函数leastsq得调用几次error函数才能找到使得均方误差之和最小的a~c
    Para=leastsq(error,p0,args=(h_mp,v_dt_hl)) #把error函数中除了p以外的参数打包到args中
    a,b,c,d=Para[0]           #leastsq 返回的第一个值是a,b,c 的求解结果，leastsq返回类型相当于c++ 中的tuple
    print(" a=",a," b=",b,"  c=",c," d=",d)
    plt.figure(figsize=(8,6))
    plt.scatter(h_mp,v_dt_hl,color="red",label="Sample Point",linewidth=3) #画样本点
    x=np.linspace(200,2200,1000)
    y=a*x**3+b*x**2+c*x+d
    integralCalculate(Para,CheckH1,CheckH2,CheckV)
    plt.plot(x,y,color="orange",label="Fitting Curve",linewidth=2) #画拟合曲线
  #  plt.plot(h_mp, v_dt_hl,color="blue",  label='Origin Line',linewidth=1)  #画连接线
    plt.legend()
    plt.show()

def freeParameterFitting(h_mp,v_dt_hl):
    z1 = np.polyfit(h_mp, v_dt_hl, 6)   # 第一个拟合，自由度为6
        # 生成多项式对象
    p1 = np.poly1d(z1)
    print("Z1:")
    print(z1)
    print("P1:")
    print(p1)
    print("n")
    x = np.linspace(400, 1700, 1000)
    plt.plot(h_mp, v_dt_hl, color="blue", label='Origin Line', linewidth=1)  # 画连接线
    plt.plot(x, p1(x), 'gv--', color="black", label='Poly Fitting Line(deg=6)', linewidth=1)
    plt.legend()
    plt.show()

def operatMysql(h_mp,v_dt_h):
    print('operatMysql start at : ',ctime())
    sql = oilCheckMysql('***.**.***.***','root','*****',"oilCheck",'checkVolume')
    i=0
    while(i<len(CheckV)):
       t=i+1
      # sql.insertMySQL(t,float(CheckH1[i]),float(CheckH2[i]),float(CheckV[i]),float(vCalcute[i]),float(ratioV[i]))
       sql.updateMySQL(t,float(vCalcute[i]),float(ratioV[i]))
       i=i+1
    sql.serachMySQL()
    print('operatMysql finished at : ', ctime())




def main():
    loadData(path)
    CheckH1_H2[:]=readFileCheck()
    print("CheckH1_H2 length: ",len(CheckH1_H2))

    # 去除被除数为0对应的点，并得到v 和 h 求导 值的列表
    VdtH[:] = removeBadPoint(h_median, h_subtract, v_subtract)
    print("h_median1：", len(h_median))

    print("VdtH1 : ", len(VdtH))

    #  赛选数据，去除异常点
    selectRightPoint(h_median, h_subtract, VdtH)
    print("h_median2：", len(h_median))
    print("h_subtract: ", len(h_subtract))
    print("VdtH2 : ", len(VdtH))
    h_mp = np.array(h_median)
    v_dt_h = np.array(VdtH)


    # 最小二乘法作图
    leastSquareFitting(h_mp, v_dt_h)
    # 多项式自由参数法作图
    freeParameterFitting(h_mp, v_dt_h)

    threads = []
    funcs = [writeFile,operatMysql]
    nfuncs = range(len(funcs))
    for i in nfuncs:
        t= MyThread(funcs[i],(h_mp,v_dt_h),funcs[i].__name__)
        threads.append(t)
    for i in nfuncs:
        threads[i].start()
    for i in nfuncs:
        threads[i].join()



if __name__ == '__main__':
    main()

最后

以上就是火星上灯泡为你收集整理的Python3.6 连接mysql 数据库，增删改查，及多线程简单运用的全部内容，希望文章能够帮你解决Python3.6 连接mysql 数据库，增删改查，及多线程简单运用所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。