Python3操作MySQL基於PyMySQL封裝的類

  在未使用操作數據庫的框架開發項目的時候，我們需要自己處理數據庫連接問題，今天在做一個Python的演示項目，寫一個操作MySQL數據庫的類，基於PyMySQL庫在Python3上實現。在寫業務邏輯代碼的時候，可以方便很多，時間關系，沒有寫太完善，只寫了常用的操作。

  直接上代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

'Python連接到 MySQL 數據庫及相關操作(基於Python3)'

import pymysql.cursors


class Database:
    """ Python連接到 MySQL 數據庫及相關操作 """
    """
    conf: 類參數，數據庫的連接參數配置字典，含host、port、user、pw、db、charset(可選,默認utf8)
    connected: 屬性，True數據庫連接成功，False連接失敗

    insert(self, table, val_obj): 方法，插入數據到數據表
        table: 數據表名稱
        val_obj: 待插入數據的字段名和值的鍵值對字典
        返回: 成功則返回新插入數據的主鍵ID，失敗返回False

    update(self, table, val_obj, range_str): 方法，更新數據表中的數據
        table: 數據表名稱
        val_obj: 待更新數據的字段名和值的鍵值對字典
        range_str: 更新范圍的條件語句字符串
        返回: 成功返回更新的行數，失敗返回False

    delete(self, table, range_str): 方法，在數據表中刪除數據
        table: 數據表名稱
        range_str: 刪除范圍的條件語句字符串
        返回: 成功返回刪除的行數，失敗返回False

    select_one(self, table, factor_str, field='*'): 方法，查詢表中符合條件唯一的一條數據
        table: 數據表名稱
        factor_str: 查詢唯一條件語句字符串
        field: 查詢結果返回哪些字段，多個用逗號分隔，可選參數，默認返回所有字段
        返回: 成功返回一條數據的字段名與值的一維字典，失敗返回False

    select_more(self, table, range_str, field='*'): 方法，查詢表中符合條件的所有數據
        table: 數據表名稱
        range_str: 查詢條件語句字符串
        field: 查詢結果返回哪些字段，多個用逗號分隔，可選參數，默認返回所有字段
        返回: 成功返回多條數據的字段名與值的二維字典，失敗返回False

    count(self, table, range_str='1'): 方法，統計數據表中符合條件的總函數
        table: 數據表名稱
        range_str: 查詢條件語句字符串，可選參數，默認表中所有行數
        返回: 成功返回符合條件的行數，失敗返回False

    sum(self, table, field, range_str='1'): 方法，對數據表中某數值類型字段求和
        table: 數據表名稱
        field: 需要求和的字段，可以是多個字段的計算公式
        range_str: 需要求和的條件語句字符串，可選參數，默認表中所有行
        返回: 成功返回求和結果，失敗返回False

    close(self): 方法，關閉數據庫連接，對象銷毀時也會自動關閉，所以多數時候不用特意調用
    """
    connected = False
    __conn = None

    # 構造函數，初始化時直接連接數據庫
    def __init__(self, conf):
        if type(conf) is not dict:
            print('錯誤: 參數不是字典類型！')
        else:
            for key in ['host', 'port', 'user', 'pw', 'db']:
                if key not in conf.keys():
                    print('錯誤: 參數字典缺少 %s' % key)
            if 'charset' not in conf.keys():
                conf['charset'] = 'utf8'
        try:
            self.__conn = pymysql.connect(
                host=conf['host'],
                port=conf['port'],
                user=conf['user'],
                passwd=conf['pw'],
                db=conf['db'],
                charset=conf['charset'],
                cursorclass=pymysql.cursors.DictCursor)
            self.connected = True
        except pymysql.Error as e:
            print('數據庫連接失敗:', end='')

    # 插入數據到數據表
    def insert(self, table, val_obj):
        sql_top = 'INSERT INTO ' + table + ' ('
        sql_tail = ') VALUES ('
        try:
            for key, val in val_obj.items():
                sql_top += key + ','
                sql_tail += val + ','
            sql = sql_top[:-1] + sql_tail[:-1] + ')'
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return self.__conn.insert_id()
        except pymysql.Error as e:
            self.__conn.rollback()
            return False

    # 更新數據到數據表
    def update(self, table, val_obj, range_str):
        sql = 'UPDATE ' + table + ' SET '
        try:
            for key, val in val_obj.items():
                sql += key + '=' + val + ','
            sql = sql[:-1] + ' WHERE ' + range_str
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return cursor.rowcount
        except pymysql.Error as e:
            self.__conn.rollback()
            return False

    # 刪除數據在數據表中
    def delete(self, table, range_str):
        sql = 'DELETE FROM ' + table + ' WHERE ' + range_str
        try:
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return cursor.rowcount
        except pymysql.Error as e:
            self.__conn.rollback()
            return False

    # 查詢唯一數據在數據表中
    def select_one(self, table, factor_str, field='*'):
        sql = 'SELECT ' + field + ' FROM ' + table + ' WHERE ' + factor_str
        try:
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return cursor.fetchall()[0]
        except pymysql.Error as e:
            return False

    # 查詢多條數據在數據表中
    def select_more(self, table, range_str, field='*'):
        sql = 'SELECT ' + field + ' FROM ' + table + ' WHERE ' + range_str
        try:
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return cursor.fetchall()
        except pymysql.Error as e:
            return False

    # 統計某表某條件下的總行數
    def count(self, table, range_str='1'):
        sql = 'SELECT count(*)res FROM ' + table + ' WHERE ' + range_str
        try:
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return cursor.fetchall()[0]['res']
        except pymysql.Error as e:
            return False

    # 統計某字段（或字段計算公式）的合計值
    def sum(self, table, field, range_str='1'):
        sql = 'SELECT SUM(' + field + ') AS res FROM ' + table + ' WHERE ' + range_str
        try:
            with self.__conn.cursor() as cursor:
                cursor.execute(sql)
            self.__conn.commit()
            return cursor.fetchall()[0]['res']
        except pymysql.Error as e:
            return False

    # 銷毀對象時關閉數據庫連接
    def __del__(self):
        try:
            self.__conn.close()
        except pymysql.Error as e:
            pass

    # 關閉數據庫連接
    def close(self):
        self.__del__()