最全總結 | 聊聊 Python 數據處理全家桶（Sqlite篇）

1. 前言

上篇文章 聊到 Python 處理 Mysql 數據庫最多見的兩種方式，本篇文章繼續說另一種比較經常使用的數據庫：Sqlite

Sqlite 是一種 嵌入式數據庫，數據庫就是一個文件，體積很小，底層由 C 語言編寫，常常被集成到移動應用程序中

事實上，Python 內置了 sqlite3 模塊，不須要安裝任何依賴，就能夠直接操做 Sqlite 數據庫

2. 準備

和 Python 操做 Mysql 相似，操做 Sqlite 主要包含下面 2 種方式：

• sqlite3 + 原生 SQL

• SQLAlchemy + ORM

3. sqlite3 + 原生 SQL

因爲 Python 內置了 sqlite3 模塊，這裏直接導入就可使用了

# 導入內置模塊sqlite3
import sqlite3

首先，咱們使用 sqlite3 的 connnect() 方法建立一個數據庫鏈接對象，若是數據庫不存在，就自動在對應目錄下新建一個數據庫文件

# 建立數據庫鏈接對象，若是數據庫不存在，就自動新建一個數據庫文件
# 還能夠指定其餘參數，包含：超時時間
 self.conn = sqlite3.connect(self.path_db)

而後，經過數據庫鏈接對象獲取一個操做數據庫的 遊標實例

# 獲取操做數據庫的遊標對象
self.cursor = self.conn.cursor()

接着，使用數據庫鏈接對象執行建立表的 SQL 語句，在數據庫內新建一張表

# 建立表
SQL_CREATE_TABLE = '''CREATE TABLE IF NOT EXISTS PEOPLE
       (ID INT PRIMARY KEY     NOT NULL,
       NAME           TEXT    NOT NULL,
       AGE            INT     NOT NULL);'''

def create_db_table(self):
    """
    初始化表
    :return:
    """
    self.conn.execute(SQL_CREATE_TABLE)

接下來，咱們經過增刪改查來操做數據表

一、新增

一樣以新增單條數據和多條數據爲例

對於單條數據的插入，只須要編寫一條插入的 SQL 語句，而後做爲參數執行上面數據庫鏈接對象的 execute(sql) 方法，最後使用數據庫鏈接對象的 commit() 方法將數據提交到數據庫中

# 插入一條數據
SQL_INSERT_ONE_DATA = "INSERT INTO PEOPLE(id,name,age) VALUES(3,'xag',23);"

def insert_one(self):
    """新增一條數據"""
    try:
        self.conn.execute(SQL_INSERT_ONE_DATA)
        # 必需要提交，才能正確執行
        self.conn.commit()
    except Exception as e:
        self.conn.rollback()
        print('插入一條記錄失敗，回滾~')

須要注意的是，插入操做常常會由於主鍵緣由致使新增異常，因此須要捕獲異常，執行回滾操做

使用數據庫鏈接對象的 executemany() 方法，傳入插入的 SQL 語句及 位置變量列表，能夠實現一次插入多條數據

# 插入多條數據（3個變量，包含：id、name、value）
SQL_INSERT_MANY_DATA = 'INSERT INTO PEOPLE (id,name,age) VALUES(?,?,?);'

# 待插入的數據
self.data = [(4, '張三', 11), (5, '李四', 12), (6, '王五', 13)]

def insert_many(self, data):
    """新增多條數據"""
    try:
        self.conn.executemany(SQL_INSERT_MANY_DATA, data)
        self.conn.commit()
    except Exception as e:
        self.conn.rollback()
        print('插入多條記錄失敗，回滾~')

二、查詢

查詢分爲 2 步，分別是：

• 經過遊標對象執行查詢的 SQL 語句

• 調用遊標對象的方法獲取查詢結果

好比：

要獲取全部數據，可使用遊標對象的 fetchall() 方法

要獲取第一條知足條件的數據，可使用 fetchone() 方法

另外，fetchmany(num) 能夠查詢固定數量的數據

# 查詢的SQL語句
SQL_QUERY_ONE_DATA = "SELECT * FROM PEOPLE WHERE id={}"

def query_one(self, id):
    """
    查詢一條數據
    :param id:
    :return:
    """
    self.cursor.execute(SQL_QUERY_ONE_DATA.format(id))

    # fetchone():查詢第一條數據
    # fetchall()：查詢全部數據
    # fetchmany(1):查詢固定的數量的數據
    result = self.cursor.fetchall()
    print(type(result))
    print(result)

三、更新

和 新增操做 相似，更新操做也是經過數據庫鏈接對象去執行更新的 SQL 語句，最後執行提交操做，將數據真實更新到數據表中

以更新某一條記錄爲例

# 更新數據
SQL_UPDATE_ONE_DATA = "UPDATE PEOPLE SET NAME = '{}',AGE={} where id = {}"

def update_one(self, id, name, age):
    """
    修改一條記錄
    :param id:
    :param name:
    :param age:
    :return:
    """
    sql_update = SQL_UPDATE_ONE_DATA.format(name, age, id)
    print(sql_update)
    self.conn.execute(sql_update)
    self.conn.commit()

四、刪除

刪除操做同查詢、新增操做相似，只須要執行刪除的 SQL 語句便可

以刪除某一條記錄爲例

# 刪除數據
SQL_DEL_ONE_DATA = "DELETE FROM PEOPLE where id ={}"

def del_one(self, id):
    """經過id去刪除一條數據"""
    sql_del = SQL_DEL_ONE_DATA.format(id)
    self.conn.execute(sql_del)
    self.conn.commit()

最後，咱們一樣須要將遊標對象和數據庫鏈接對象，資源釋放

def teardown(self):
    # 關閉遊標和數據庫鏈接，避免資源浪費
    self.cursor.close()
    self.conn.close()

4. SQLAlchemy + ORM

使用 SQLAlchemy 操做 sqlite 數據庫一樣先須要安裝依賴庫

# 安裝依賴包
pip3 install sqlalchemy

經過內置方法 declarative_base() 建立一個基礎類 Base

而後，自定義一個 Base 類的子類，內部經過定義靜態變量指定表名、表的字段

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base

# 基礎類
Base = declarative_base()


# 自定義的表
class People(Base):
    # 表名
    __tablename__ = 'people'

    # 定義字段
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

    def __repr__(self):
        """
        便於打印結果
        :return:
        """
        return "<People(id:{},name:{},age:{})".format(self.id, self.name, self.age)

接着，經過 SQLAlchemy 的 create_engine(sqlite數據庫路徑) 方法中建立數據庫鏈接對象

格式爲：sqlite:///數據庫相對路徑

# 建立數據庫鏈接
engine = create_engine('sqlite:///./xh.db', echo=True)

最後，經過數據庫引擎在數據庫中建立表結構，並實例化一個 數據庫會話對象

PS：數據庫會話對象內置的方法很是方便咱們進行增刪改查操做

# 建立表結構
# checkfirst：判斷表是否存在，若是存在，就不重複建立
Base.metadata.create_all(engine, checkfirst=True)

# 實例化會話
self.session = sessionmaker(bind=engine)()

這樣全部的準備工做已經完成，接下來能夠進行增刪改查操做了

一、新增

新增操做一樣以新增一條和多條記錄爲例，它們分別對應會話對象的 add()、add_all() 方法

對於一條記錄的新增操做，只須要實例化一個 People 對象，執行上面的會話對象的 add(instance) 和 commit() 兩個方法，便可以將數據插入到數據表中

def add_one_data(self):
    """新增一條數據"""
    # 建立一個表的實例對象
    people = People(name='xag1', age=24)
    self.session.add(people)

    # 必須提交，才能更新到數據庫中
    self.session.commit()

若是須要一次插入多條數據，只須要調用 add_all(列表數據) 便可

def add_datas(self, data):
    """
    新增多條數據
    :return:
    """
    self.session.add_all(data)
    self.session.commit()

二、查詢

查詢數據表的操做對應會話對象的 query() 方法

同時，還能夠結合 all()、first()、filter_by(限制條件) 級聯方法限制要查詢的數據

以查詢全部記錄和根據 id 查詢一條記錄爲例

def query_one_data(self, id):
    """
    經過id去查詢一條數據
    :param id:
    :return:
    """
    # 經過id去查詢數據，取第一條
    people = self.session.query(People).filter_by(id=id).first()
    print(people)
    print(type(people))

def query_all(self):
    """
    查詢全部數據
    :return:
    """
    peoples = self.session.query(People).all()
    print(peoples)
    print(type(peoples))

三、更新

更新操做通常作法是：

• query 查詢出待更新的對象

• 直接更新對象中的數據

• 使用會話對象提交修改，完成更新操做

def update1(self, id, name, age):
    """
    更新記錄
    :param id:
    :param name:
    :param age:
    :return:
    """
    # 更新步驟：先查詢、修改數據、而後確認修改
    people_temp = self.session.query(People).filter_by(id=id).first()

    # 修改數據
    people_temp.name = name
    people_temp.age = age

    # 確認提交修改
    self.session.commit()

須要指出的是，這裏也可使用內置方法 update() ，對上面的更新操做進行簡寫

def update2(self, id, name, age):
    """
    更新記錄方法2
    :param id:
    :param name:
    :param age:
    :return:
    """
    self.session.query(People).filter_by(id=id).update({People.name: name, People.age: age})
    self.session.commit()

四、刪除

和更新操做同樣，刪除操做也有兩種實現方式

第一種方式的思路是，先查詢，後刪除，最後提交會話完成刪除操做

以按照 id 刪除某一條記錄爲例：

def del_one_data1(self, id):
    """
    刪除一條數據方法1
    :param id:
    :return:
    """
    people_temp = self.session.query(People).filter_by(id=id).first()

    # 判斷是否爲空
    if people_temp is not None:
        self.session.delete(people_temp)
        self.session.commit()
    else:
        print('此條記錄不存在，刪除失敗！')

須要注意的是，查詢的結果必須判斷是否爲空，不然直接執行刪除操做，能夠會拋出異常

另一種方式是，直接使用級聯函數將上面的刪除操做進行簡寫

def del_one_data2(self, id):
    """
    刪除一條數據方法2
    :param id:
    :return:
    """
    try:
        self.session.query(People).filter_by(id=id).first().delete()
    except Exception as e:
        print('刪除失敗')

一樣，這種刪除操做須要捕獲異常，避免查詢的結果爲空的狀況

最後，完成全部操做以後，咱們須要結束會話、銷燬數據庫引擎

def teardown(self):
    """
    釋放資源
    :return:
    """
    # 結束會話
    self.session.close()

    # 銷燬引擎
    self.engine.dispose()

5.最後

本篇文章經過一張表的增刪改查，詳細講解了 Python 操做 sqlite 的兩種使用方式

我已經將文中所有源碼上傳到後臺，關注公衆號「 AirPython 」後回覆「 dball 」便可得到所有源碼

若是你以爲文章還不錯，請你們 點贊、分享、留言下，由於這將是我持續輸出更多優質文章的最強動力！

推薦閱讀

聊聊 Python 數據處理全家桶（Mysql 篇）

Python 如何使用 HttpRunner 作接口自動化測試

Python 自動化，Helium 憑什麼取代 Selenium？