快捷導(dǎo)航

Python使用sqlalchemy實現(xiàn)連接數(shù)據(jù)庫的幫助類

更新時間：2024年02月15日 07:26:28 作者：搬磚的詩人Z

這篇文章主要為大家詳細介紹了Python如何使用sqlalchemy實現(xiàn)連接數(shù)據(jù)庫的幫助類,文中的示例代碼講解詳細,具有一定的借鑒價值,需要的可以參考下

python使用 sqlalchemy連接數(shù)據(jù)庫幫助類

實現(xiàn)代碼

import mysql.connector

class MySqlHelper(object):
    """操作數(shù)據(jù)庫幫助類"""

    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]))


    #不帶參數(shù)的查詢
    def select(self,mysql):
        try:
            self.mycursor.execute(mysql)
            values = self.mycursor.fetchall()
            return values
        except Exception as e:
            print ('select Error : %d %s' % (e.args[0],e.args[1]))
            return []
        finally:
            self.mycursor.close()
            self.mydb.close()

    #帶參數(shù)的查詢
    def select2(self,mysql,na):
        try:
            self.mycursor.execute(mysql,na)
            values = self.mycursor.fetchall()
            return values
        except 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.rowcount
            if row > 0:
                return True
            else:
                return False
         except Exception as e:
            print ('Update Error : %d %s' % (e.args[0],e.args[1]))
            return False
          
         finally:
            self.mycursor.close()
            self.mydb.close()

    #插入數(shù)據(jù)
    def Insert(self,mysql,na):
         try:
            self.mycursor.execute(mysql,na)
            self.mydb.commit()
            row = self.mycursor.rowcount
            if row > 0:
                return True
            else:
                return False
         except Exception as e:
            print('Insert Error : %d %s' % (e.args[0],e.args[1]))
            return False
          
         finally:
            self.mycursor.close()
            self.mydb.close()

使用數(shù)據(jù)：

from DAL import MySqlHelper
from Entity import TaskPoolEntity
import datetime

class TaskPoolDAL(object):
    """操作數(shù)據(jù)庫t_du_guiji_task"""

    sqlHelper = MySqlHelper.MySqlHelper()

    #查詢所有的任務(wù)
    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

以上就是Python使用sqlalchemy實現(xiàn)連接數(shù)據(jù)庫的幫助類的詳細內(nèi)容，更多關(guān)于Python sqlalchemy連接數(shù)據(jù)庫的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: