快捷導(dǎo)航

Python SQLite3數(shù)據(jù)庫操作類分享

更新時間：2014年06月10日 11:55:16 作者：

這篇文章主要介紹了Python SQLite3數(shù)據(jù)庫操作類分享,需要的朋友可以參考下

接觸Python時間也不是很長的，最近有個項目需要分析數(shù)據(jù)，于是選用Python為編程語言，除了語言特性外主要還是看重Python對于SQLite3數(shù)據(jù)庫良好的支持能力了，因為需要靈活處理大量的中間數(shù)據(jù)。

剛開始一些模塊我還樂此不疲的寫SQL語句，后來漸漸厭倦了，回想到以前搗鼓C#的時候利用反射初步構(gòu)建了個SQL查詢構(gòu)造器，直到發(fā)現(xiàn)linq，于是放棄了這個計劃，當(dāng)然微軟后來又推出了Entity Framework，這些都是后話了，而且現(xiàn)在我對微軟的東西興趣不是很大的，好了，扯多了，下面繼續(xù)正文。

對了，再扯一句，優(yōu)秀的博客程序Drupal也使用了類似的查詢構(gòu)造器進行數(shù)據(jù)庫查詢，避免直接寫SQL語句，另外這樣做的一點點好處就是，可以一定程度的屏蔽平臺相關(guān)性，對于數(shù)據(jù)庫遷移還是有幫助的。

不過我今天介紹的數(shù)據(jù)庫輔助類查詢構(gòu)造器是個很簡單的東東，甚至只限于SQLite數(shù)據(jù)庫，如果有童鞋感興趣可以完善下，我目前只要操作SQLite順手就可以了，對于比較大的數(shù)據(jù)庫應(yīng)用就直接上ORM吧。

先看代碼：

復(fù)制代碼代碼如下:

import sqlite3

# ***************************************************
# *
# * Description: Python操作SQLite3數(shù)據(jù)庫輔助類(查詢構(gòu)造器)
# * Author: wangye
# *
# ***************************************************

def _wrap_value(value):
    return repr(value)

def _wrap_values(values):
    return list(map(_wrap_value, values))

def _wrap_fields(fields):
    for key,value in fields.items():
        fields[key] = _wrap_value(value)
    return fields

def _concat_keys(keys):
    return "[" + "],[".join(keys) + "]"

def _concat_values(values):
    return ",".join(values)

def _concat_fields(fields, operator = (None, ",")):
    if operator:
        unit_operator, group_operator = operator
    # fields = _wrap_fields(fields)
    compiled = []
    for key,value in fields.items():
        compiled.append("[" + key + "]")
        if unit_operator:
            compiled.append(unit_operator)
            compiled.append(value)
        compiled.append(group_operator)
    compiled.pop() # pop last group_operator
    return " ".join(compiled)

class DataCondition(object):
    """
        本類用于操作SQL構(gòu)造器輔助類的條件語句部分

        例如:
        DataCondition(("=", "AND"), id = 26)
        DataCondition(("=", "AND"), True, id = 26)
    """

    def __init__(self, operator = ("=", "AND"), ingroup = True, **kwargs):
        """
            構(gòu)造方法
            參數(shù):
                operator 操作符，分為(表達式操作符, 條件運算符)
                ingroup  是否分組，如果分組，將以括號包含
                kwargs   鍵值元組，包含數(shù)據(jù)庫表的列名以及值
                         注意這里的等于號不等于實際生成SQL語句符號
                         實際符號是由operator[0]控制的
            例如:
            DataCondition(("=", "AND"), id = 26)
            (id=26)
            DataCondition((">", "OR"), id = 26, age = 35)
            (id>26 OR age>35)
            DataCondition(("LIKE", "OR"), False, name = "John", company = "Google")
            name LIKE 'John' OR company LIKE "Google"
        """
        self.ingroup = ingroup
        self.fields = kwargs
        self.operator = operator

    def __unicode__(self):
        self.fields = _wrap_fields(self.fields)
        result = _concat_fields(self.fields, self.operator)
        if self.ingroup:
            return "(" + result + ")"
        return result

    def __str__(self):
        return self.__unicode__()

    def toString(self):
        return self.__unicode__()

class DataHelper(object):

    """
        SQLite3 數(shù)據(jù)查詢輔助類
    """

    def __init__(self, filename):
        """
            構(gòu)造方法
            參數(shù): filename 為SQLite3 數(shù)據(jù)庫文件名
        """
        self.file_name = filename

    def open(self):
        """
            打開數(shù)據(jù)庫并設(shè)置游標
        """
        self.connection = sqlite3.connect(self.file_name)
        self.cursor = self.connection.cursor()
        return self

    def close(self):
        """
            關(guān)閉數(shù)據(jù)庫，注意若不顯式調(diào)用此方法，
            在類被回收時也會嘗試調(diào)用
        """
        if hasattr(self, "connection") and self.connection:
            self.connection.close()

    def __del__(self):
        """
            析構(gòu)方法，做一些清理工作
        """
        self.close()

    def commit(self):
        """
            提交事務(wù)
            SELECT語句不需要此操作，默認的execute方法的
            commit_at_once設(shè)為True會隱式調(diào)用此方法，
            否則就需要顯示調(diào)用本方法。
        """
        self.connection.commit()

    def execute(self, sql = None, commit_at_once = True):
        """
            執(zhí)行SQL語句
            參數(shù):
                sql  要執(zhí)行的SQL語句，若為None，則調(diào)用構(gòu)造器生成的SQL語句。
                commit_at_once 是否立即提交事務(wù)，如果不立即提交，
                對于非查詢操作，則需要調(diào)用commit顯式提交。
        """
        if not sql:
            sql = self.sql
        self.cursor.execute(sql)
        if commit_at_once:
            self.commit()

    def fetchone(self, sql = None):
        """
            取一條記錄
        """
        self.execute(sql, False)
        return self.cursor.fetchone()

    def fetchall(self, sql = None):
        """
            取所有記錄
        """
        self.execute(sql, False)
        return self.cursor.fetchall()

    def __concat_keys(self, keys):
        return _concat_keys(keys)

    def __concat_values(self, values):
        return _concat_values(values)

    def table(self, *args):
        """
            設(shè)置查詢的表，多個表名用逗號分隔
        """
        self.tables = args
        self.tables_snippet = self.__concat_keys(self.tables)
        return self

    def __wrap_value(self, value):
        return _wrap_value(value)

    def __wrap_values(self, values):
        return _wrap_values(values)

    def __wrap_fields(self, fields):
        return _wrap_fields(fields)

    def __where(self):
        # self.condition_snippet
        if hasattr(self, "condition_snippet"):
            self.where_snippet = " WHERE " + self.condition_snippet

    def __select(self):
        template = "SELECT %(keys)s FROM %(tables)s"
        body_snippet_fields = {
            "tables" : self.tables_snippet,
            "keys" : self.__concat_keys(self.body_keys), 
        }
        self.sql = template % body_snippet_fields

    def __insert(self):
        template = "INSERT INTO %(tables)s (%(keys)s) VALUES (%(values)s)"
        body_snippet_fields = {
            "tables" : self.tables_snippet,
            "keys" : self.__concat_keys(list(self.body_fields.keys())),
            "values" : self.__concat_values(list(self.body_fields.values()))
        }
        self.sql = template % body_snippet_fields

    def __update(self):
        template = "UPDATE %(tables)s SET %(fields)s"
        body_snippet_fields = {
            "tables" : self.tables_snippet,
            "fields" : _concat_fields(self.body_fields, ("=",","))
        }
        self.sql = template % body_snippet_fields

    def __delete(self):
        template = "DELETE FROM %(tables)s"
        body_snippet_fields = {
            "tables" : self.tables_snippet
        }
        self.sql = template % body_snippet_fields

    def __build(self):
        {
            "SELECT": self.__select,
            "INSERT": self.__insert,
            "UPDATE": self.__update,
            "DELETE": self.__delete
        }[self.current_token]()

    def __unicode__(self):
        return self.sql

    def __str__(self):
        return self.__unicode__()

    def select(self, *args):
        self.current_token = "SELECT"
        self.body_keys = args
        self.__build()
        return self

    def insert(self, **kwargs):
        self.current_token = "INSERT"
        self.body_fields = self.__wrap_fields(kwargs)
        self.__build()
        return self

    def update(self, **kwargs):
        self.current_token = "UPDATE"
        self.body_fields = self.__wrap_fields(kwargs)
        self.__build()
        return self

    def delete(self, *conditions):
        self.current_token = "DELETE"
        self.__build()
        #if *conditions:
        self.where(*conditions)
        return self

    def where(self, *conditions):
        conditions = list(map(str, conditions))
        self.condition_snippet = " AND ".join(conditions)
        self.__where()
        if hasattr(self, "where_snippet"):
            self.sql += self.where_snippet
        return self

下面舉幾個例子供大家參考吧：

復(fù)制代碼代碼如下:

db = DataHelper("/home/wangye/sample.db3")
db.open() # 打開數(shù)據(jù)庫
db.execute("""
    CREATE TABLE [staffs] (
      [staff_id] INTEGER PRIMARY KEY AUTOINCREMENT,
      [staff_name] TEXT NOT NULL,
      [staff_cardnum] TEXT NOT NULL,
      [staff_reserved] INTEGER NOT NULL
)
""") # 直接執(zhí)行SQL語句，注意這里commit_at_once默認為True

db.table("staffs").insert(staff_name="John", staff_cardnum="1001", staff_reserved=0)
# 插入一條記錄

rs = db.table("staffs").select("staff_id", "staff_name").fetchall()
# 直接取出所有staff_id和staff_name

rs = db.table("staffs").select("staff_name").where(DataCondition(("=", "AND"), id = 1)).fetchone()
# 取一條staff_id為1的staff_name

rs = db.table("staffs").select("staff_name").where(DataCondition(("<", "AND"), id = 100), DataCondition(("=", "AND"), staff_reserved = 1)).fetchone()
# 取一條id小于100并且staff_reserved為1的staff_name記錄

db.close() # 關(guān)閉數(shù)據(jù)庫