對於訪問數據庫的操作,我想大家也都有一些了解。不過,因為最近在學習Python,以下就用Python來實現它。其中包括創建數據庫和數據表、插入記錄、刪除記錄、修改記錄數據、查詢數據、刪除數據表、刪除數據庫。還有一點就是我們最好使用一個新定義的類來處理這件事。因為這會使在以後的使用過程中更加的方便(只需要導入即可,避免了重復制造輪子)。
1.封裝一個DB類
2.數據庫操作:創建數據庫和數據表
3.數據庫操作:插入記錄
4.數據庫操作:一次插入多條記錄
5.數據庫操作:刪除記錄
6.數據庫操作:修改記錄數據
7.數據庫操作:一次修改多條記錄數據
8.數據庫操作:查詢數據
9.數據庫操作:刪除數據表
10.數據庫操作:刪除數據庫
heroDB.py
#!/usr/bin/env python import MySQLdb DATABASE_NAME = 'hero' class HeroDB: # init class and create a database def __init__(self, name, conn, cur): self.name = name self.conn = conn self.cur = cur try: 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 table def createTable(self, name): try: ex = self.cur.execute if 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 record def 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 records def 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 data def 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 table def 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 table def getCount(self, name): try: count = self.cur.execute('select * from ' + name) return count except MySQLdb.Error, e: print "Mysql Error %d: %s" % (e.args[0], e.args[1]) # select first record from database def selectFirst(self, name): try: self.cur.execute('select * from ' + name + ';') result = self.cur.fetchone() return result except MySQLdb.Error, e: print "Mysql Error %d: %s" % (e.args[0], e.args[1]) # select last record from database def selectLast(self, name): try: self.cur.execute('SELECT * FROM ' + name + ' ORDER BY id DESC;') result = self.cur.fetchone() return result except MySQLdb.Error, e: print "Mysql Error %d: %s" % (e.args[0], e.args[1]) # select next n records from database def selectNRecord(self, name, n): try: self.cur.execute('select * from ' + name + ';') results = self.cur.fetchmany(n) return results except MySQLdb.Error, e: print "Mysql Error %d: %s" % (e.args[0], e.args[1]) # select all records def 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 results except MySQLdb.Error, e: print "Mysql Error %d: %s" % (e.args[0], e.args[1]) # delete a record def 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 record def deleteSome(self, name): pass # drop the table def 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 database def 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 python import MySQLdb from heroDB import HeroDB def 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 '-' * 60 print 'first record' result = hero.selectFirst('heros') print result print '-' * 60 print 'last record' result = hero.selectLast('heros') print result print '-' * 60 print 'more record' results = hero.selectNRecord('heros', 3) for item in results: print item print '-' * 60 print '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()
注:請不要不假思索地使用他們。如果你想實現某一個功能點,請最好將其他的功能點注釋掉,這樣才符合單元測試的規范。