程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
 程式師世界 >> 數據庫知識 >> MYSQL數據庫 >> MySQL綜合教程 >> Python訪問MySQL數據庫並實現其增刪改查功能

Python訪問MySQL數據庫並實現其增刪改查功能

編輯:MySQL綜合教程

Python訪問MySQL數據庫並實現其增刪改查功能


概述:

對於訪問數據庫的操作,我想大家也都有一些了解。不過,因為最近在學習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()

注:請不要不假思索地使用他們。如果你想實現某一個功能點,請最好將其他的功能點注釋掉,這樣才符合單元測試的規范。

  1. 上一頁:
  2. 下一頁:
Copyright © 程式師世界 All Rights Reserved