Python3操做MySQL基於PyMySQL封裝的類

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__()