概述：

? 對于訪問數(shù)據(jù)庫的操作，我想大家也都有一些了解。不過，因為最近在學(xué)習(xí)Python，以下就用Python來實現(xiàn)它。其中包括創(chuàng)建數(shù)據(jù)庫和數(shù)據(jù)表、插入記錄、刪除記錄、修改記錄數(shù)據(jù)、查詢數(shù)據(jù)、刪除數(shù)據(jù)表、刪除數(shù)據(jù)庫。還有一點就是我們最好使用一個新定義的類來處理這件事。因為這會使在以后的使用過程中更加的方便（只需要導(dǎo)入即可，避免了重復(fù)制造輪子）。

實現(xiàn)功能介紹：

1.封裝一個DB類

2.數(shù)據(jù)庫操作：創(chuàng)建數(shù)據(jù)庫和數(shù)據(jù)表

3.數(shù)據(jù)庫操作：插入記錄

4.數(shù)據(jù)庫操作：一次插入多條記錄

5.數(shù)據(jù)庫操作：刪除記錄

6.數(shù)據(jù)庫操作：修改記錄數(shù)據(jù)

7.數(shù)據(jù)庫操作：一次修改多條記錄數(shù)據(jù)

8.數(shù)據(jù)庫操作：查詢數(shù)據(jù)

9.數(shù)據(jù)庫操作：刪除數(shù)據(jù)表

10.數(shù)據(jù)庫操作：刪除數(shù)據(jù)庫

數(shù)據(jù)庫類的定義：

heroDB.py

#!/usr/bin/env pythonimport MySQLdbDATABASE_NAME = 'hero'class HeroDB:# init class and create a databasedef __init__(self, name, conn, cur):self.name = nameself.conn = connself.cur = curtry:cur.execute('create database if not exists ' + name)conn.select_db(name)conn.commit()except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# create a tabledef createTable(self, name):try:ex = self.cur.executeif ex('show tables') == 0:ex('create table ' + name + '(id int, name varchar(20), sex int, age int, info varchar(50))')self.conn.commit()except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# insert single recorddef insert(self, name, value):try:self.cur.execute('insert into ' + name + ' values(%s,%s,%s,%s,%s)', value)except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# insert more recordsdef insertMore(self, name, values):try:self.cur.executemany('insert into ' + name + ' values(%s,%s,%s,%s,%s)', values)except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# update single record from table# name: table name# values: waiting to update datadef updateSingle(self, name, value):try:# self.cur.execute('update ' + name + ' set name=' + str(values[1]) + ', sex=' + str(values[2]) + ', age=' + str(values[3]) + ', info=' + str(values[4]) + ' where id=' + str(values[0]) + ';')self.cur.execute('update ' + name + ' set name=%s, sex=%s, age=%s, info=%s where id=%s;', value)except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# update some record from tabledef update(self, name, values):try:self.cur.executemany('update ' + name + ' set name=%s, sex=%s, age=%s, info=%s where id=%s;', values)except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# get record count from db tabledef getCount(self, name):try:count = self.cur.execute('select * from ' + name)return countexcept MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# select first record from databasedef selectFirst(self, name):try:self.cur.execute('select * from ' + name + ';')result = self.cur.fetchone()return resultexcept MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# select last record from databasedef selectLast(self, name):try:self.cur.execute('SELECT * FROM ' + name + ' ORDER BY id DESC;')result = self.cur.fetchone()return resultexcept MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# select next n records from databasedef selectNRecord(self, name, n):try:self.cur.execute('select * from ' + name + ';')results = self.cur.fetchmany(n)return resultsexcept MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# select all recordsdef selectAll(self, name):try:self.cur.execute('select * from ' + name + ';')self.cur.scroll(0, mode='absolute') # reset cursor location (mode = absolute | relative)results = self.cur.fetchall()return resultsexcept MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# delete a recorddef deleteByID(self, name, id):try:self.cur.execute('delete from ' + name + ' where id=%s;', id)except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# delete some recorddef deleteSome(self, name):pass# drop the tabledef dropTable(self, name):try:self.cur.execute('drop table ' + name + ';')except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])# drop the databasedef dropDB(self, name):try:self.cur.execute('drop database ' + name + ';')except MySQLdb.Error, e:print "Mysql Error %d: %s" % (e.args[0], e.args[1])def __del__(self):if self.cur != None:self.cur.close()if self.conn != None:self.conn.close()

使用范例：

testHeroDB.py

#!/usr/bin/env pythonimport MySQLdb from heroDB import HeroDBdef main():conn = MySQLdb.connect(host='localhost', user='root', passwd='260606', db='hero', port=3306, charset='utf8')cur = conn.cursor()# ------------------------------------------- create -----------------------------------------------------hero = HeroDB('hero', conn, cur)hero.createTable('heros')# ------------------------------------------- insert -----------------------------------------------------hero.insert('heros', [3, 'Prophet', 0, 2000, 'The hero who in fairy tale.'])# ------------------------------------------- select -----------------------------------------------------print '-' * 60print 'first record'result = hero.selectFirst('heros')print resultprint '-' * 60print 'last record'result = hero.selectLast('heros')print resultprint '-' * 60print 'more record'results = hero.selectNRecord('heros', 3)for item in results:print itemprint '-' * 60print 'all record'results = hero.selectAll('heros')for item in results:print item# ------------------------------------------- update -----------------------------------------------------hero.updateSingle('heros', ['Zeus', 1, 22000, 'The god.', 2])values = []values.append(['SunWukong', 1, 1300, 'The hero who in fairy tale.', 1])values.append(['Zeus', 1, 50000, 'The king who in The Quartet myth.', 2])values.append(['Prophet', 1, 20000, 'The hero who in fairy tale.3', 3])hero.update('heros', values)# ------------------------------------------- delete -----------------------------------------------------hero.deleteByID('heros', 1)hero.dropTable('heros')hero.dropDB('hero')if __name__ == '__main__':main()

注：請不要不假思索地使用他們。如果你想實現(xiàn)某一個功能點，請最好將其他的功能點注釋掉，這樣才符合單元測試的規(guī)范。

代碼優(yōu)化：

# encoding:utf-8 #!/usr/bin/env pythonimport MySQLdb import dbConfigclass HeroDB:# init class and create a databasedef __init__(self, dbname = None, dbhost = None):if dbname is None:self._dbname = dbConfig.DBNAMEelse:self._dbname = dbnameif dbhost is None:self._dbhost = dbConfig.DBHOSTelse:self._dbhost = dbhostself._dbuser = dbConfig.DBUSERself._dbpassword = dbConfig.DBPWDself._dbport = dbConfig.DBPORTself._dbcharset = dbConfig.DBCHARSETself._conn = self.connectMySQL()if(self._conn):self._cursor = self._conn.cursor()# 連接數(shù)據(jù)庫def connectMySQL(self):conn = Falsetry:conn = MySQLdb.connect(host=self._dbhost,user=self._dbuser,passwd=self._dbpassword,port=self._dbport,db=self._dbname,charset=self._dbcharset)except Exception, data:print "connect database failed, %s" % dataconn = Falsereturn conn# 創(chuàng)建數(shù)據(jù)表def createTable(self, sql):flag = Falsetry:ex = self._cursor.executeex(sql)self._conn.commit()flag = Trueexcept MySQLdb.Error, e:flag = Falseprint "Mysql Error %d: %s" % (e.args[0], e.args[1])return flag# 刪除數(shù)據(jù)表def dropTable(self, sql):flag = Falsetry:self._cursor.execute(sql)self._conn.commit()flag = Trueexcept MySQLdb.Error, e:flag = Falseprint "Mysql Error %d: %s" % (e.args[0], e.args[1])return flag# 添加一條記錄def insert(self, sql):flag = Falseif(self._conn):try:self._cursor.execute(sql)self._conn.commit()flag = Trueexcept Exception, data:flag = Falseprint "update database exception, %s" % datareturn flag# 獲取查詢結(jié)果集(全部)def fetch_all(self, sql):res = ''if(self._conn):try:self._cursor.execute(sql)res = self._cursor.fetchall()except Exception, data:res = Falseprint "query database exception, %s" % datareturn res# 更新數(shù)據(jù)庫def update(self, sql):flag = Falseif(self._conn):try:self._cursor.execute(sql)self._conn.commit()flag = Trueexcept Exception, data:flag = Falseprint "update database exception, %s" % datareturn flag# 獲得數(shù)據(jù)表中的記錄數(shù)def getCount(self, sql):count = 0try:count = self._cursor.execute(sql)self._conn.commit()except MySQLdb.Error, e:count = 0print "Mysql Error %d: %s" % (e.args[0], e.args[1])return count# 刪除記錄def delete(self, sql):flag = Falseif(self._conn):try:self._cursor.execute(sql)self._conn.commit()flag = Trueexcept Exception, data:flag = Falseprint "update database exception, %s" % datareturn flag# 關(guān)閉數(shù)據(jù)庫連接def close(self):if(self._conn):try:if(type(self._cursor)=='object'):self._cursor.close()if(type(self._conn)=='object'):self._conn.close()except Exception, data:print "close database exception, %s,%s,%s" % (data, type(self._cursor), type(self._conn))
? 上面的代碼做了一些優(yōu)化，使代碼看上去更簡潔、靈活以及對后期維護的考慮。

源碼下載：

http://download.csdn.net/detail/u013761665/8615981

http://download.csdn.net/detail/u013761665/8876101

總結(jié)

以上是生活随笔為你收集整理的Python访问MySQL数据库并实现其增删改查功能的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。