python使用 sqlalchemy连接数据库帮助类-编程知识

python使用 sqlalchemy连接数据库帮助类

在这里插入图片描述

import mysql.connectorclass MySqlHelper(object):"""操作数据库帮助类"""def __init__(self):#self.host = "localhost"#self.user = "root"#self.password = "xinshiyun@123"#self.database = "deliverunion_callcenter"self.host = "192.168.60.156"self.user = "root"self.password = "root"self.database = "deliverunion_callcenter"try:self.mydb = mysql.connector.connect(host=self.host,user=self.user,passwd=self.password,database=self.database,connect_timeout=10)#database=self.database,#auth_plugin='mysql_native_password')self.mycursor = self.mydb.cursor()except Exception as e:print('MySql Error : %d %s' % (e.args[0],e.args[1]))#不带参数的查询def select(self,mysql):try:self.mycursor.execute(mysql)values = self.mycursor.fetchall()return valuesexcept Exception as e:print ('select Error : %d %s' % (e.args[0],e.args[1]))return []finally:self.mycursor.close()self.mydb.close()#带参数的查询def select2(self,mysql,na):try:self.mycursor.execute(mysql,na)values = self.mycursor.fetchall()return valuesexcept Exception as e:print ('select2 Error : %d %s' % (e.args[0],e.args[1]))return []finally:self.mycursor.close()self.mydb.close()#更新def Update(self,mysql,na):try:self.mycursor.execute(mysql,na)self.mydb.commit()row = self.mycursor.rowcountif row > 0:return Trueelse:return Falseexcept Exception as e:print ('Update Error : %d %s' % (e.args[0],e.args[1]))return Falsefinally:self.mycursor.close()self.mydb.close()#插入数据def Insert(self,mysql,na):try:self.mycursor.execute(mysql,na)self.mydb.commit()row = self.mycursor.rowcountif row > 0:return Trueelse:return Falseexcept Exception as e:print('Insert Error : %d %s' % (e.args[0],e.args[1]))return Falsefinally:self.mycursor.close()self.mydb.close()

使用数据：

from DAL import MySqlHelper
from Entity import TaskPoolEntity
import datetimeclass TaskPoolDAL(object):"""操作数据库t_du_guiji_task"""sqlHelper = MySqlHelper.MySqlHelper()#查询所有的任务def selectTasks():TaskPoolDAL.sqlHelper = MySqlHelper.MySqlHelper()sql='select * from  t_du_guiji_task'values= TaskPoolDAL.sqlHelper.select(sql)data=[]for item in values:selectTask = TaskPoolEntity.TaskPoolEntity(item)data.append(selectTask)return data