python sqlite3操作類擴展，包含數據庫分頁

 一、原因

最近在使用python3和sqlite3編輯一些小程序，由於要使用數據庫，就離不開增、刪、改、查，sqlite3的操作同java里的jdbc很像，於是就想找現成的操作類，找來找去，發現一個相對來說簡單的封裝，代碼如下：

版權聲明：本文為博主原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處鏈接和本聲明。
本文鏈接：https://blog.csdn.net/u013314786/article/details/78226902
————————————————
import sqlite3
 
 
class EasySqlite:
    """
    sqlite數據庫操作工具類
    database: 數據庫文件地址，例如：db/mydb.db
    """
    _connection = None
 
    def __init__(self, database):
        # 連接數據庫
        self._connection = sqlite3.connect(database)
 
    def _dict_factory(self, cursor, row):
        d = {}
        for idx, col in enumerate(cursor.description):
            d[col[0]] = row[idx]
        return d
 
    def execute(self, sql, args=[], result_dict=True, commit=True)->list:
        """
        執行數據庫操作的通用方法
        Args:
        sql: sql語句
        args: sql參數
        result_dict: 操作結果是否用dict格式返回
        commit: 是否提交事務
        Returns:
        list 列表，例如：
        [{'id': 1, 'name': '張三'}, {'id': 2, 'name': '李四'}]
        """
        if result_dict:
            self._connection.row_factory = self._dict_factory
        else:
            self._connection.row_factory = None
        # 獲取游標
        _cursor = self._connection.cursor()
        # 執行SQL獲取結果
        _cursor.execute(sql, args)
        if commit:
            self._connection.commit()
        data = _cursor.fetchall()
        _cursor.close()
        return data
 
 
if __name__ == '__main__':
    db = EasySqlite('browser.db')
    # print(db.execute("select name from sqlite_master where type=?", ['table']))
    # print(db.execute("pragma table_info([user])"))
    # print(execute("insert into user(id, name, password) values (?, ?, ?)", [2, "李四", "123456"]))
    print(db.execute("select id, name userName, password pwd from user"))
    print(db.execute("select * from user", result_dict=False))
    print(db.execute("select * from user"))

    db = EasySqlite('browser.db')
    # print(db.execute("select name from sqlite_master where type=?", ['table']))
    # print(db.execute("pragma table_info([user])"))
    # print(execute("insert into user(id, name, password) values (?, ?, ?)", [2, "李四", "123456"]))
    #print(db.execute("select id, name userName, password pwd from user"))
    #print(db.execute("select * from user", result_dict=False))
    #print(db.execute("select * from user"))

執行的方式如上一段代碼，大體上是初始化時傳入sqlite3數據庫路徑，使用db.excecute方法來執行sql，返回的是Dict數組。

---

二、此工具類的擴展

但一個類寫相同的增、刪、改、查，感覺很費時間，於是想借鑒java的反射機制，嘗試使用python的反射來實現MVC中的module基類，得到以下代碼：

class DbSuper(object):
    dbHelper=None   #類變量，共用一個EasySqlite工具類
    
    def __init__(self):
        """
            初始化數據庫
            
        """
        super().__init__()
        
    def setDb(self,  dburl): 
        """
            參數：
                dburl——數據庫文件位置,str類型
        """
        DbSuper.dbHelper = EasySqlite(dburl)   
        
    
    def add(self, obj):
        """
            將實例儲存到數據庫，數據庫中的表名應與類名一致，表中字段名與類定義的變量名一致 ，順序也得一致
            參數：
                obj——類實例
            返回值：無    
        """
        sql = 'insert into '+type(obj).__name__+' values('  #通過type(obj).__name__獲得表名
        paras = []  #sql語句的參數
        tag = True
        for attr in obj.__dict__.keys():   #獲取實例對象的屬性名obj.__dict__
            if tag:
                tag=False  #第一項是ID，自動生成，跳過
                continue
            sql += ',?'   #循環幾次，就加幾次? 生成   insert into xxxx values(,?,?,?,?)的sql語句
            para = getattr(obj, attr)    # 使用getattr函數，利用反射獲得類屬性實際的值
            
            if type(para)==str:    #對值進行判斷，如果非str類型，應做轉換，避免sql執行錯誤
                paras.append(para)
            else :
                paras.append(str(para))
        sql = sql.replace(',','null,', 1)    #將多余的 ， 處理一下
        sql += ')'
        #print(sql)
        #print(paras)
        DbSuper.dbHelper.execute(sql, paras)   #利用工具類執行SQL
        
    def findByProperty(self, objclass, propertyName,  propertyValueStr,strict = True,  orderby='id', pager = False, numPerPage=1, page = 1):
        """
            通過類的某一個屬性查找
            參數：
                objclass——class類型，類名
                propertyName——str類型，篩選依據的屬性名
                propertyValueStr——object類型，篩選依據的屬性名對應的值
                strict——bool類型，文本字段是否精確匹配，非文本字段請勿改變此值
                orderby——str類型，排序的依據，默認ID排序
                pager——bool類型，查詢的結果是否分頁
                numPerPage——int類型，如pager=True，則此參數起作用，每頁顯示數據量
                page——int類型，如pager=True，則此參數起作用，頁數
            返回值：objclass的list
        """
        sql = 'select * from %s where ' % objclass.__name__
        #對propertyValueStr進行判斷，非str型，進行轉換
        if type(propertyValueStr) != str:
            propertyValueStr = str(propertyValueStr)
            
        if strict:#默認嚴格匹配
            sql += '%s = ? order by %s '% (propertyName, orderby)
        else:
            sql += '%s like ? order by %s '% (propertyName, orderby)
            propertyValueStr = '%' + propertyValueStr + '%'
        if pager: #對pager進行判斷，默認不進行分頁處理
            sql += 'limit %d offset %d' % (numPerPage, numPerPage * (page - 1))
        retObjects = []
        
            #DbSuper.dbHelper.execute(sql, [propertyValueStr, ])執行SQL，結果返回為Dict數組
        print(sql)
        for ret in  DbSuper.dbHelper.execute(sql, [propertyValueStr, ]):
            #利用變量生成實例
            obj = objclass()
            #調用initByStr方法，將Dict解釋，並賦值給對應屬性，因不同類實現方式不同，故此方法由類聲明時自行完成，類似接口
            obj.initByStr(ret)
            retObjects.append(obj)
        return retObjects
        
    def findByPropertyFirst(self, objclass, propertyName,  propertyValueStr, strict=True):
        """
            類似於findByProperty，做了一定簡化，且只查詢一個結果
            返回值：成功返回對象實例，失敗返回空
        """
        sql = 'select * from %s where %s = ? limit 1' % (objclass.__name__, propertyName)
        if strict==False:
            propertyValueStr = '%' + propertyValueStr + '%'
        ret = DbSuper.dbHelper.execute(sql,  [propertyValueStr, ])
        if len(ret)>0:
            obj = objclass()
            obj.initByStr(ret[0])
            return obj 
        else:
            return None
            
    def modify(self, obj, propertyIndex='id'):
        """
            更新類，並存於數據庫
            參數：
                obj——類實例
                propertyIndex——篩選依據的字段，默認ID
            返回值： 無
        """
        sql = 'update %s set ' % type(obj).__name__     #利用反射，通過實例獲得類名，即表名
        params = []
        for attr in obj.__dict__.keys():        #遍歷每個屬性，生成update語句中的set xxx=?，注意要跳過篩選依據的屬性
            if attr == propertyIndex:
                continue
            else:
                sql += ', %s=?'  % attr 
                
                #對屬性值進行處理，如果不是str型，要轉換
                if type(getattr(obj, attr)) == str:
                    params.append(getattr(obj, attr))
                else:
                    params.append(str(getattr(obj, attr)))
        #篩選條件語句生成
        sql += ' where %s = ?' % propertyIndex
        #加入參數
        params.append(getattr(obj, propertyIndex))
        #對生成的sql語句處理，去掉多余的，   執行SQL語句
        DbSuper.dbHelper.execute(sql .replace(',', '', 1), params)
    
    def delete(self, obj, propertyIndex='id'):
        """
            刪除對象
            參數：
                obj——待刪除的對象
                propertyIndex——篩選依據
                
        """
        sql = 'delete from %s where %s=?' % (type(obj).__name__, propertyIndex)
        param = getattr(obj, propertyIndex)
        if type(param) != str:
            param = str(param)
        DbSuper.dbHelper.execute(sql , [param, ])

 

 三、使用前提條件

• 類名要與數據庫中表名一致
• 類中屬性與數據庫中字段名一致
• 為解決查詢結果轉換成類的問題，類中要實現一個方法initByStr

四、使用舉例

1.數據庫中表創建示例，注意表名operators，此處模擬一用戶基本信息

CREATE TABLE [operators] (
[id] INTEGER  NOT NULL PRIMARY KEY AUTOINCREMENT,
[loginname] vaRCHAR(20)  UNIQUE NOT NULL,
[loginpass] vaRCHAR(100)  NOT NULL,
[showname] vaRCHAR(30)  NULL,
[level] vaRCHAR(100)  NULL
)

 

2.類operators聲明

# -*- coding=utf-8 -*-
from enum import Enum
import abc

class Levels(Enum):
    """
        枚舉類，標明權限類型
    """
    DATA_INPUTER='查詢數據,錄入數據,修改數據'
    USER_MANAGER='增加用戶,修改用戶基本信息'
    POWER_MANAGER='增加用戶,修改用戶基本信息,修改用戶權限'

class DbEntity(object):
    
    @abc.abstractmethod 
    def initByStr(self, attrDict):
        pass


class Operators(DbEntity):    
    """
        用戶類
    """
    def __init__(self):
        super().__init__()
        self.id=0
        self.loginName=''
        self.loginPass=''
        self.showName=''
        self.level=Levels.DATA_INPUTER

    def initByStr(self, attrDict):
        if len(attrDict)==5:
            self.id = int(attrDict['id'])
            self.loginName = attrDict['loginname']
            self.loginPass = attrDict['loginpass']
            self.showName = attrDict['showname']
            self.level = Levels(attrDict['level'])

 

DbEntity是基類，只聲明了一個接口initByStr，子類必須實現，原本我想在擴展類里實現這個方法，但也只能實現基本數據類型，一旦類里的屬性比較復雜也不好實現，所以還是由類中聲明每一個字符串如何轉化成類。
3.准備工作完成后，下面實現OperatorDao，代碼如下：

class OperatorDao(DbSuper):
    
    def __init__(self):
        super().__init__()
    
    def findById(self, id):
        """
            根據ID查找類
            返回類，如未找到返回空
        """
        return super().findByPropertyFirst(Operators, 'id', id)
            

    def findByLoginname(self, loginname):
        """
            根據登錄名查找類
            返回類，如未找到返回空
        """
        return super().findByPropertyFirst(Operators, 'loginName', loginname)
        #return super().findByProperty(Operators, 'loginName', loginname)
        #return super().findByProperty(Operators, 'loginName', loginname,strict=False)
        #return super().findByProperty(Operators, 'loginName', loginname, pager = True, numPerPage=5, page = 1)
    
    def addOper(self, oper):
        #可以對實例進一步處理，比如MD5加密 oper.loginPass = MDUtils.md5Text(oper.loginPass)
        return super().add(oper)
    
    def modiOper(self, oper):
        return super().modify(oper)
    
    def delOper(self, oper):
        return super().delete(oper)


    
if __name__ == '__main__':
    operatorDao = OperatorDao()
    operatorDao.setDb('xxxxxx.s3db')
    oper = operatorDao.findByLoginname('test')
    for op in oper:
        print(op)

 

 

只是簡單擴展，還可以加入配置文件，標出類屬性與數據庫字段關系，這樣就可以不用字段名與類屬性一致，但實現更復雜，目前先做到這個程度，有時間再進一步處理。