python操作MySQL數據庫報錯問題解決

編寫好Python操作數據庫的腳本后，運行報錯如下：

報錯1：“AttributeError: 'NoneType' object has no attribute 'encoding'”

解決辦法：設置charset時要用utf8，不能用utf-8

    if link_type == 0:
                # 創建數據，返回字典
                self.conn = pymysql.connect(host=host_db, user=user_db, password=password_db, db=name_db, port=port_db,
                                            charset='utf8', cursorclass=pymysql.cursors.DictCursor)
            else:
                # 創建數據庫，返回元祖
                self.conn = pymysql.connect(host=host_db, user=user_db, password=password_db, db=name_db, port=port_db,
                                            charset='utf8')

報錯2：“AttributeError: 'OperationDbInterface' object has no attribute 'cur'”

錯誤原因：編寫代碼時，換行時沒有對齊，導致調用cur時報錯

貼上Python操作mysql數據庫的源碼：

# coding:utf8

'''
定義對mysql數據庫操作的封裝
1、包括基本的單條語句操作，如刪除、修改、插入
2、獨立地查詢單條、多條數據
3、獨立地添加多條數據
'''

import logging, os, pymysql
from public import config


class OperationDbInterface(object):
    # 定義初始化數據庫連接
    def __init__(self, host_db='127.0.0.1', user_db='root', password_db='123456',
                 name_db='interface_test', port_db=3306, link_type=0):
        '''
        :param host_db: 數據庫服務主機IP
        :param user_db: 數據庫連接用戶名
        :param password_db: 數據庫密碼
        :param name_db: 數據庫名稱
        :param port_db: 數據庫端口號，整型數據
        :param link_type: 連接類型，用於設置輸出數據是元祖還是字典，默認是字典，link_type=0
        :return:游標
        '''
        try:
            if link_type == 0:
                # 創建數據，返回字典
                self.conn = pymysql.connect(host=host_db, user=user_db, password=password_db, db=name_db, port=port_db,
                                            charset='utf8', cursorclass=pymysql.cursors.DictCursor)
            else:
                # 創建數據庫，返回元祖
                self.conn = pymysql.connect(host=host_db, user=user_db, password=password_db, db=name_db, port=port_db,
                                            charset='utf8')
            self.cur = self.conn.cursor()
            print("輸出：%s" % self.cur)
        except pymysql.Error as e:
            print("創建數據庫連接失敗|Mysql Error %d: %s" % (e.args[0], e.args[1]))
            logging.basicConfig(filename=config.src_path + '/log/syserror.log', level=logging.DEBUG,
                                format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
            logger = logging.getLogger(__name__)
            logger.exception(e)

    # 定義單條數據操作，包含刪除和更新操作
    def op_sql(self, condition):
        '''
        :param condition: sql語句，該通用方法可用來替代updateone,deleteone
        :return: 字典形式
        '''

        try:
            self.cur.execute(condition)  # 執行sql語句
            self.conn.commit()  # 提交游標數據
            result = {'code': '0000', 'message': '執行通用操作成功', 'data': []}
        except pymysql.Error as e:
            self.conn.rollback()  # 執行回滾操作
            result = {'code': '9999', 'message': '執行通用操作異常', 'data': []}
            print("數據庫錯誤|op_sql %d:%s" % (e.args[0], e.args[1]))
            logging.basicConfig(filename=config.src_path + '/log/syserror.log', level=logging.DEBUG,
                                format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
            logger = logging.getLogger(__name__)
            logger.exception(e)
        return result

    # 查詢表中單條數據
    def select_one(self, condition):
        '''
        :param condition: 查詢單條sql語句
        :return: 字典形式的單條查詢語句
        '''

        try:
            rows_affect = self.cur.execute(condition)
            if rows_affect > 0:
                results = self.cur.fetchone()  # 獲取一條數據
                result = {'code': '0000', 'message': '執行單條查詢操作成功', 'data': results}
            else:
                result = {'code': '0000', 'message': '執行單條查詢操作成功', 'data': []}
        except pymysql.Error as e:
            self.conn.rollback()  # 執行回滾操作
            result = {'code': '9999', 'message': '執行單條查詢異常', 'data': []}
            print("數據庫錯誤|select_one %d: %s" % (e.args[0], e.args[1]))
            logging.basicConfig(filename=config.src_path + '/log/syserror.log', level=logging.DEBUG,
                                format='%(asctime)s %(filename)s [line:%(lineno)d] %(levelname)s %(message)s')
            logger = logging.getLogger(__name__)
            logger.exception(e)
        return result

    # 查詢表中多條數據
    def select_all(self, condition):
        '''
        :param condition: sql語句
        :return: 字典形式的批量查詢結果
        '''

        try:
            rows_affect = self.cur.execute(condition)
            if rows_affect > 0:
                # 將鼠標光標放回到初始位置
                self.cur.scroll(0, mode='absolute')
                # 返回游標中所有結果
                results = self.cur.fetchall()
                result = {'code': '0000', 'message': '執行批量查詢操作成功', 'data': results}
            else:
                result = {'code': '0000', 'message': '執行批量查詢操作成功', 'data': []}
        except pymysql.Error as e:
            # 執行回滾操作
            self.conn.rollback()
            result = {'code': '9999', 'message': '執行批量查詢異常', 'data': []}
            logging.basicConfig(filename=config.src_path + '/log/syserror.log', level=logging.DEBUG,
                                format='%(asctime)s %(filename)s [line:%(lineno)d] %(levelname)s %(message)s')
            logger = logging.getLogger(__name__)
            logger.exception(e)
        return result

    # 定義表中插入數據操作的方法
    def insert_data(self, condition, params):
        '''
        :param condition: 插入語句
        :param params: 插入數據，列表形式[(,,,),(,,,)]
        :return: 字典形式批量插入數據結果
        '''
        try:
            results = self.cur.executemany(condition, params)  # 返回插入數據的條數
            self.conn.commit()
            result = {'code': '0000', 'message': '執行批量插入數據成功', 'data': results}
        except pymysql.Error as e:
            self.conn.rollback()  # 執行回滾操作
            result = {'code': '9999', 'message': '執行批量插入數據異常', 'data': []}
            print("數據庫錯誤|insert_data %d: %s" % (e.args[0], e.args[1]))
            logging.basicConfig(filename=config.src_path + '/log/syserror.log', level=logging.DEBUG,
                                format='%(asctime)s %(filename)s [line:%(lineno)d] %(levelname)s %(massage)s')
            logger = logging.getLogger(__name__)
            logger.exception(e)
        return result

    # 關閉數據庫
    def __del__(self):
        if self.cur is not None:
            self.cur.close()  # 關閉游標
        if self.conn is not None:
            self.conn.close()  # 釋放數據庫資源


if __name__ == "__main__":
    test = OperationDbInterface()  # 實例化類
    result_select_all = test.select_all("select * from config_test")  # 查詢所有數據
    result_select_one = test.select_one("select * from config_test where id=1")  # 查詢單條數據
    result_oP_sql = test.op_sql("UPDATE config_test SET value_config='修改后的valuetest' WHERE id=1")  # 通用操作，修改數據庫
    result__insert_sql = test.insert_data(
        "insert into config_test (key_config, value_config, description, status) values (%s, %s, %s, %s)",
        [('mytest1', 'mytestvalue1', '我插入的測試數據1', 1), ('mytest2', 'mytestvalue2', '我插入的測試數據2', 0)])
    print(result_select_all['data'], result_select_all['message'])
    print(result_select_one['data'], result_select_one['message'])
    print(result_oP_sql['data'], result_oP_sql['message'])
    print(result__insert_sql['data'], result__insert_sql['message'])