作者：星安果

來(lái)源：AirPython

聊到 python中的Redis，本篇文章繼續(xù)說(shuō)另外一種比較常用的數(shù)據(jù)庫(kù)：Sqlite。

Sqlite 是一種 嵌入式數(shù)據(jù)庫(kù)，數(shù)據(jù)庫(kù)就是一個(gè)文件，體積很小，底層由 C 語(yǔ)言編寫(xiě)，經(jīng)常被集成到移動(dòng)應(yīng)用程序中事實(shí)上，python 內(nèi)置了 sqlite3 模塊，不需要安裝任何依賴，就可以直接操作 Sqlite 數(shù)據(jù)庫(kù)

——準(zhǔn)備

和 Python 操作 Mysql 類(lèi)似，操作 Sqlite 主要包含下面 2 種方式：

• sqlite3 + 原生 SQL
• SQLAlchemy + ORM

——sqlite3 + 原生 SQL

由于 Python 內(nèi)置了 sqlite3 模塊，這里直接導(dǎo)入就可以使用了

# 導(dǎo)入內(nèi)置模塊sqlite3
import sqlite3

首先，我們使用 sqlite3 的 connnect() 方法創(chuàng)建一個(gè)數(shù)據(jù)庫(kù)連接對(duì)象，如果數(shù)據(jù)庫(kù)不存在，就自動(dòng)在對(duì)應(yīng)目錄下新建一個(gè)數(shù)據(jù)庫(kù)文件

# 創(chuàng)建數(shù)據(jù)庫(kù)連接對(duì)象，如果數(shù)據(jù)庫(kù)不存在，就自動(dòng)新建一個(gè)數(shù)據(jù)庫(kù)文件
# 還可以指定其他參數(shù)，包含：超時(shí)時(shí)間
 self.conn = sqlite3.connect(self.path_db)

然后，通過(guò)數(shù)據(jù)庫(kù)連接對(duì)象獲取一個(gè)操作數(shù)據(jù)庫(kù)的 游標(biāo)實(shí)例

# 獲取操作數(shù)據(jù)庫(kù)的游標(biāo)對(duì)象
self.cursor = self.conn.cursor()

接著，使用數(shù)據(jù)庫(kù)連接對(duì)象執(zhí)行創(chuàng)建表的 SQL 語(yǔ)句，在數(shù)據(jù)庫(kù)內(nèi)新建一張表

# 創(chuàng)建表
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)

接下來(lái)，我們通過(guò)增刪改查來(lái)操作數(shù)據(jù)表

1、新增

同樣以新增單條數(shù)據(jù)和多條數(shù)據(jù)為例。對(duì)于單條數(shù)據(jù)的插入，只需要編寫(xiě)一條插入的 SQL 語(yǔ)句，然后作為參數(shù)執(zhí)行上面數(shù)據(jù)庫(kù)連接對(duì)象的 execute(sql) 方法，最后使用數(shù)據(jù)庫(kù)連接對(duì)象的 commit() 方法將數(shù)據(jù)提交到數(shù)據(jù)庫(kù)中

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

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

需要注意的是，插入操作經(jīng)常會(huì)因?yàn)橹麈I原因?qū)е滦略霎惓?，所以需要捕獲異常，執(zhí)行回滾操作。

使用數(shù)據(jù)庫(kù)連接對(duì)象的 executemany() 方法，傳入插入的 SQL 語(yǔ)句及 位置變量列表，可以實(shí)現(xiàn)一次插入多條數(shù)據(jù)。

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

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

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

2、查詢

查詢分為 2 步，分別是：

• 通過(guò)游標(biāo)對(duì)象執(zhí)行查詢的 SQL 語(yǔ)句
• 調(diào)用游標(biāo)對(duì)象的方法獲取查詢結(jié)果

比如：

要獲取所有數(shù)據(jù)，可以使用游標(biāo)對(duì)象的 fetchall() 方法

要獲取第一條滿足條件的數(shù)據(jù)，可以使用 fetchone() 方法

另外，fetchmany(num) 可以查詢固定數(shù)量的數(shù)據(jù)

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

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

    # fetchone():查詢第一條數(shù)據(jù)
    # fetchall()：查詢所有數(shù)據(jù)
    # fetchmany(1):查詢固定的數(shù)量的數(shù)據(jù)
    result = self.cursor.fetchall()
    print(type(result))
    print(result)

3、更新

和 新增操作 類(lèi)似，更新操作也是通過(guò)數(shù)據(jù)庫(kù)連接對(duì)象去執(zhí)行更新的 SQL 語(yǔ)句，最后執(zhí)行提交操作，將數(shù)據(jù)真實(shí)更新到數(shù)據(jù)表中

以更新某一條記錄為例

# 更新數(shù)據(jù)
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()

4、刪除

刪除操作同查詢、新增操作類(lèi)似，只需要執(zhí)行刪除的 SQL 語(yǔ)句即可

以刪除某一條記錄為例

# 刪除數(shù)據(jù)
SQL_DEL_ONE_DATA = "DELETE FROM PEOPLE where id ={}"

def del_one(self, id):
    """通過(guò)id去刪除一條數(shù)據(jù)"""
    sql_del = SQL_DEL_ONE_DATA.format(id)
    self.conn.execute(sql_del)
    self.conn.commit()

最后，我們同樣需要將游標(biāo)對(duì)象和數(shù)據(jù)庫(kù)連接對(duì)象，資源釋放

def teardown(self):
    # 關(guān)閉游標(biāo)和數(shù)據(jù)庫(kù)連接，避免資源浪費(fèi)
    self.cursor.close()
    self.conn.close()

——SQLAlchemy + ORM

使用 SQLAlchemy 操作 sqlite 數(shù)據(jù)庫(kù)同樣先需要安裝依賴庫(kù)

# 安裝依賴包
pip3 install sqlalchemy

通過(guò)內(nèi)置方法 declarative_base() 創(chuàng)建一個(gè)基礎(chǔ)類(lèi) Base。然后，自定義一個(gè) Base 類(lèi)的子類(lèi)，內(nèi)部通過(guò)定義靜態(tài)變量指定表名、表的字段。

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

# 基礎(chǔ)類(lèi)
Base = declarative_base()


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

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

    def __repr__(self):
        """
        便于打印結(jié)果
        :return:
        """
        return "

	接著，通過(guò) SQLAlchemy 的 create_engine(sqlite數(shù)據(jù)庫(kù)路徑) 方法中創(chuàng)建數(shù)據(jù)庫(kù)連接對(duì)象，格式為：sqlite:///數(shù)據(jù)庫(kù)相對(duì)路徑


# 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('sqlite:///./xh.db', echo=True)

	最后，通過(guò)數(shù)據(jù)庫(kù)引擎在數(shù)據(jù)庫(kù)中創(chuàng)建表結(jié)構(gòu)，并實(shí)例化一個(gè) 數(shù)據(jù)庫(kù)會(huì)話對(duì)象


	PS：數(shù)據(jù)庫(kù)會(huì)話對(duì)象內(nèi)置的方法非常方便我們進(jìn)行增刪改查操作

# 創(chuàng)建表結(jié)構(gòu)
# checkfirst：判斷表是否存在，如果存在，就不重復(fù)創(chuàng)建
Base.metadata.create_all(engine, checkfirst=True)

# 實(shí)例化會(huì)話
self.session = sessionmaker(bind=engine)()

	這樣所有的準(zhǔn)備工作已經(jīng)完成，接下來(lái)可以進(jìn)行增刪改查操作了。


	1、新增


	新增操作同樣以新增一條和多條記錄為例，它們分別對(duì)應(yīng)會(huì)話對(duì)象的 add()、add_all() 方法。


	對(duì)于一條記錄的新增操作，只需要實(shí)例化一個(gè) People 對(duì)象，執(zhí)行上面的會(huì)話對(duì)象的 add(instance)和 commit() 兩個(gè)方法，即可以將數(shù)據(jù)插入到數(shù)據(jù)表中。

def add_one_data(self):
    """新增一條數(shù)據(jù)"""
    # 創(chuàng)建一個(gè)表的實(shí)例對(duì)象
    people = People(name='xag1', age=24)
    self.session.add(people)

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

	如果需要一次插入多條數(shù)據(jù)，只需要調(diào)用 add_all(列表數(shù)據(jù)) 即可。

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


	2、查詢


	查詢數(shù)據(jù)表的操作對(duì)應(yīng)會(huì)話對(duì)象的 query() 方法


	同時(shí)，還可以結(jié)合 all()、first()、filter_by(限制條件) 級(jí)聯(lián)方法限制要查詢的數(shù)據(jù)


	以查詢所有記錄和根據(jù) id 查詢一條記錄為例

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

def query_all(self):
    """
    查詢所有數(shù)據(jù)
    :return:
    """
    peoples = self.session.query(People).all()
    print(peoples)
    print(type(peoples))

	3、更新


	更新操作一般做法是：


	

		query 查詢出待更新的對(duì)象
	
	

		直接更新對(duì)象中的數(shù)據(jù)
	
	

		使用會(huì)話對(duì)象提交修改，完成更新操作
	

def update1(self, id, name, age):
    """
    更新記錄
    :param id:
    :param name:
    :param age:
    :return:
    """
    # 更新步驟：先查詢、修改數(shù)據(jù)、然后確認(rèn)修改
    people_temp = self.session.query(People).filter_by(id=id).first()

    # 修改數(shù)據(jù)
    people_temp.name = name
    people_temp.age = age

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

	需要指出的是，這里也可以使用內(nèi)置方法 update() ，對(duì)上面的更新操作進(jìn)行簡(jiǎn)寫(xiě)

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

	4、刪除


	和更新操作一樣，刪除操作也有兩種實(shí)現(xiàn)方式


	第一種方式的思路是，先查詢，后刪除，最后提交會(huì)話完成刪除操作


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

def del_one_data1(self, id):
    """
    刪除一條數(shù)據(jù)方法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('此條記錄不存在，刪除失??！')

	需要注意的是，查詢的結(jié)果必須判斷是否為空，否則直接執(zhí)行刪除操作，可以會(huì)拋出異常。另外一種方式是，直接使用級(jí)聯(lián)函數(shù)將上面的刪除操作進(jìn)行簡(jiǎn)寫(xiě)。

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

	同樣，這種刪除操作需要捕獲異常，避免查詢的結(jié)果為空的情況。最后，完成所有操作之后，我們需要結(jié)束會(huì)話、銷(xiāo)毀數(shù)據(jù)庫(kù)引擎

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

    # 銷(xiāo)毀引擎
    self.engine.dispose()

	本篇文章通過(guò)一張表的增刪改查，詳細(xì)講解了 python 操作 sqlite 的兩種使用方式。