Python之Mysql数据库

在实际工作中经常会需要操作数据库，此处以链接Mysql数据库为例，进行说明

代码如下：

# __author__ ='dbl'
# -*- coding: utf-8 -*-

'''
对Mysql数据库基本操作的封装
1、单条增删改查操作
2、多条增删改查操作
'''
from sqlalchemy.sql.functions import now
import MySQLdb
from Configs import mylogger


class OperationDbInterface(object):
    def __init__(self, host, user, passwd, db, port=3306, charset='utf8', link_type=0):
        '''
        初始化操作，通过link_type 来确定返回结果的类型
        :param host:
        :param user:
        :param passwd:
        :param db:
        :param port:
        :param charset:
        :param link_type:
        '''

        self.host = host
        self.user = user
        self.passwd = passwd
        self.db = db
        self.port = port
        self.charset = charset
        try:
            self.conn = MySQLdb.connect(host=self.host, user=self.user, passwd=self.passwd, db=self.db, port=self.port,
                                        charset=self.charset)
            if link_type == 0:
                self.cur = self.conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
            else:
                self.cur = self.conn.cursor()

        except MySQLdb.Error, e:
            print "Mysql Error %d:%s" % (e.args[0], e.args[1])
            mylogger.info(u'Mysql数据库连接失败！')
            mylogger.info(e)

    def selectone(self, params):
        '''
        :param params:
        :return:
        '''
        try:
            self.cur.execute(params)
            results = self.cur.fetchone()
            result = {'code': '0000', 'message': u'执行单条查询操作成功', 'data': results}
        except MySQLdb.Error, e:
            result = {'code': '9999', 'message': u'执行单条查询异常', 'data': []}
            print "Mysql Error %d:%s" % (e.args[0], e.args[1])
            mylogger.info(u'Mysql查询失败！')
            mylogger.info(e)
        finally:
            return result

    def selectall(self, params):
        '''
        :param params:
        :return:
        '''
        try:
            rows = self.cur.execute(params)
            if rows > 0:
                self.cur.scroll(0, mode='absolute')
                results = self.cur.fetchall()
                result = {'code': '0000', 'message': u'执行批量查询操作成功', 'data': results}
            else:
                result = result = {'code': '0000', 'message': u'执行批量查询操作成功', 'data': []}
        except MySQLdb.Error, e:
            result = {'code': '9999', 'message': u'执行批量查询异常', 'data': []}
            print 'Mysql Error %d:%s' % (e.args[0], e.args[1])
            mylogger.info(e, u'Mysql查询所有数据失败！')
            mylogger.info(e)
        finally:
            return result

    def opone(self, params):
        '''
        :param params:
        :return:
        '''
        try:
            self.cur.execute(params)
            self.conn.commit()
            result = {'code': '0000', 'message': u'执行通用操作成功', 'data': []}
        except MySQLdb.Error, e:
            result = {'code': '9999', 'message': u'执行通用操作异常', 'data': []}
            print "Mysql Error %d:%s" % (e.args[0], e.args[1])
            mylogger.info(e, u'数据库插入数据失败！')
            mylogger.info(e)
        return result

    def opmany(self, stmt, data):
        '''
        :param stmt:
        :param data:
        :return:
        '''
        try:
            results = self.cur.executemany(stmt, data)
            self.conn.commit()
            result = {'code': '0000', 'message': u'执行批量查询操作成功', 'data': int(results)}
        except MySQLdb.Error, e:
            result = {'code': '9999', 'message': u'执行批量查询异常', 'data': []}
            print "Mysql Error %d:%s" % (e.args[0], e.args[1])
            mylogger.info(e, u'I/U/D数据库失败！')
            mylogger.info(e)
        return result

    def __del__(self):
        if self.cur != None:
            self.cur.close()
        if self.conn != None:
            self.conn.close()


if __name__ == '__main__':
    conn = OperationDbInterface(host='172.16.x.x', user='xxt', passwd='xxxx', db='xxxx')
    conn.opone(
        '''insert into config(id,key_config,value_config,description,status,create_time,update_time)  values(30, 7, 'we', 34, 0, now(), now())''')

    data = [(25, 7, 'we', 34, 0, now(), now()),
            (26, 8, 'we', 34, 0, now(), now()),
            (27, 9, 'we', 34, 0, now(), now()),
            (28, 11, 'we', 34, 0, now(), now()),
            (29, 11, 'we', 34, 0, now(), now())]
    update = [(5, 1)]
    stmt = '''insert into config(id,key_config,value_config,description,status,create_time,update_time) values(%s,%s,%s,%s,%s,%s,%s)'''
    updatestmt = ''' update config set status = %s where id =%s'''
    conn.opmany(stmt, data)
    conn.opmany(updatestmt, update)
    a = conn.selectone('select * from config where id = 7')
    b = conn.selectall('select * from config')
    print a
    print "======================="
    print b

问题一：为何要设置link_type?

• 通过link_type的值，来确定数据库查询结果的返回类型，0为字典类型，非0 为元组