認識Flask-SQLAlchemy

1. Flask-SQLAlchemy 是一個為 Flask 應用增加 SQLAlchemy 支持的擴展。它致力于簡化在 Flask 中 SQLAlchemy 的使用。
2. SQLAlchemy 是目前python中最強大的 ORM框架, 功能全面, 使用簡單。

ORM優缺點
優點

1. 有語法提示, 省去自己拼寫SQL，保證SQL語法的正確性
2. orm提供方言功能(dialect, 可以轉換為多種數據庫的語法), 減少學習成本
3. 防止sql注入攻擊
4. 搭配數據遷移, 更新數據庫方便
5. 面向對象, 可讀性強, 開發效率高

缺點

1. 需要語法轉換, 效率比原生sql低
2. 復雜的查詢往往語法比較復雜 (可以使用原生sql替換)

環境安裝

pip install flask-sqlalchemy

flask-sqlalchemy 在安裝/使用過程中, 如果出現 ModuleNotFoundError: No module named 'MySQLdb’錯誤, 則表示缺少mysql依賴包, 可依次嘗試下列兩個方案后重試:
方案1: 安裝 mysqlclient依賴包 (如果失敗再嘗試方案2)

pip install mysqlclient

方案2: 安裝pymysql依賴包

pip install pymysql

mysqlclient 和 pymysql 都是用于mysql訪問的依賴包, 前者由C語言實現的, 而后者由python實現, 前者的執行效率比后者更高, 但前者在windows系統中兼容性較差, 工作中建議優先前者。

組件初始化
基本配置

flask-sqlalchemy 的相關配置也封裝到了 flask 的配置項中, 可以通過app.config屬性 或 配置加載方案 (如config.from_object) 進行設置

數據庫URI(連接地址)格式: 協議名://用戶名:密碼@數據庫IP:端口號/數據庫名, 如:
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test31'

注意點如果數據庫驅動使用的是 pymysql, 則協議名需要修改為
mysql+pymysql://xxxxxxx
?

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)# 設置數據庫連接地址
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test31'
# 是否追蹤數據庫修改(開啟后會觸發一些鉤子函數) ?一般不開啟, 會影響性能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 是否顯示底層執行的SQL語句
app.config['SQLALCHEMY_ECHO'] = True

兩種初始化方式
.方式1
flask-sqlalchemy 支持兩種組件初始化方式:

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)# 應用配置
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test31'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True# 方式1: 初始化組件對象, 直接關聯Flask應用
db = SQLAlchemy(app)

方式2: 先創建組件, 延后關聯Flass應用

from flask import Flask
from flask_sqlalchemy import SQLAlchemy# 方式2: 初始化組件對象, 延后關聯Flask應用
db = SQLAlchemy()def create_app(config_type):"""工廠函數"""# 創建應用flask_app = Flask(__name__)# 加載配置config_class = config_dict[config_type]flask_app.config.from_object(config_class)# 關聯flask應用db.init_app(app)return flask_app

構建模型類

flask-sqlalchemy 的關系映射和 Django-orm 類似

類 對應 表
類屬性 對應 字段
實例對象 對應 記錄

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os
import pymysql as MySQLdbbasedir = os.path.abspath(os.path.dirname(__file__))
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:yu201541010@127.0.0.1:3306/pythontest'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True# 創建組件對象
db = SQLAlchemy(app)# 構建模型類  類->表  類屬性->字段  實例對象->記錄
class User(db.Model):__tablename__ = 't_user'  # 設置表名, 表名默認為類名小寫id = db.Column(db.Integer, primary_key=True)  # 設置主鍵, 默認自增name = db.Column('username', db.String(20), unique=True)  # 設置字段名 和 唯一約束age = db.Column(db.Integer, default=10, index=True)  # 設置默認值約束 和 索引with app.app_context():db.create_all()if __name__ == '__main__':app.run(debug=True)

注意點

1. 模型類必須繼承 db.Model, 其中 db 指對應的組件對象
2. 表名默認為類名小寫, 可以通過 __tablename__類屬性 進行修改
3. 類屬性對應字段, 必須是通過 db.Column() 創建的對象
4. 可以通過 create_all() 和 drop_all()方法 來創建和刪除所有模型類對應的表常用的字段類型

常用的字段選項

注意點: 如果沒有給對應字段的類屬性設置default參數, 且添加數據時也沒有給該字段賦值, 則sqlalchemy會給該字段設置默認值 None

?

數據操作
增加數據

@app.route('/')
def index():# 增加數據user1 = User(name = 'zs', age = 20)#將模型對象添加到會話中db.session.add(user1)db.session.commit()return "index"

注意點:

這里的 會話 并不是 狀態保持機制中的 session，而是 sqlalchemy 的會話。它被設計為 數據操作的執行者, 從SQL角度則可以理解為是一個 加強版的數據庫事務
sqlalchemy 會 自動創建事務, 并將數據操作包含在事務中, 提交會話時就會提交事務
事務提交失敗會自動回滾
查詢數據

class Users(db.Model):__tablename__ = 'users'id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(64))email = db.Column(db.String(64))age = db.Column(db.Integer)def __repr__(self):return "(%s, %s, %s, %s)" % (self.id, self.name, self.email, self.age)
@app.route('/createusers')
def createusers():user1 = Users(name='wang', email='wang@163.com', age=20)user2 = Users(name='zhang', email='zhang@189.com', age=33)user3 = Users(name='chen', email='chen@126.com', age=23)user4 = Users(name='zhou', email='zhou@163.com', age=29)user5 = Users(name='tang', email='tang@itheima.com', age=25)user6 = Users(name='wu', email='wu@gmail.com', age=25)user7 = Users(name='qian', email='qian@gmail.com', age=23)user8 = Users(name='liu', email='liu@itheima.com', age=30)user9 = Users(name='li', email='li@163.com', age=28)user10 = Users(name='sun', email='sun@163.com', age=26)db.session.add_all([user1, user2, user3, user5, user4, user6, user7, user8,user9,user10])db.session.commit()return "success"
@app.route('/query')
def query():user = Users.query.first()return user.name+" "+user.email

# 查詢所有用戶數據
User.query.all() 返回列表, 元素為模型對象

# 查詢有多少個用戶
User.query.count()

# 查詢第1個用戶
User.query.first() ?返回模型對象/None

# 查詢id為4的用戶[3種方式]
# 方式1: 根據id查詢 ?返回模型對象/None
User.query.get(4) ?

# 方式2: 等值過濾器 關鍵字實參設置字段值 ?返回BaseQuery對象
# BaseQuery對象可以續接其他過濾器/執行器 ?如 all/count/first等
User.query.filter_by(id=4).all() ?

# 方式3: 復雜過濾器 ?參數為比較運算/函數引用等 ?返回BaseQuery對象
User.query.filter(User.id == 4).first() ?

# 查詢名字結尾字符為g的所有用戶[開始 / 包含]
User.query.filter(User.name.endswith("g")).all()
User.query.filter(User.name.startswith("w")).all()
User.query.filter(User.name.contains("n")).all()
User.query.filter(User.name.like("w%n%g")).all() ?# 模糊查詢

# 查詢名字和郵箱都以li開頭的所有用戶[2種方式]
User.query.filter(User.name.startswith('li'), User.email.startswith('li')).all()
from sqlalchemy import and_
User.query.filter(and_(User.name.startswith('li'), User.email.startswith('li'))).all()

# 查詢age是25 或者 `email`以`itheima.com`結尾的所有用戶
from sqlalchemy import or_
User.query.filter(or_(User.age==25, User.email.endswith("itheima.com"))).all()

# 查詢名字不等于wang的所有用戶[2種方式]
from sqlalchemy import not_
User.query.filter(not_(User.name == 'wang')).all()
User.query.filter(User.name != 'wang').all()

# 查詢id為[1, 3, 5, 7, 9]的用戶
User.query.filter(User.id.in_([1, 3, 5, 7, 9])).all()

# 所有用戶先按年齡從小到大, 再按id從大到小排序, 取前5個
User.query.order_by(User.age, User.id.desc()).limit(5).all()

# 查詢年齡從小到大第2-5位的數據 ? 2 3 4 5
User.query.order_by(User.age).offset(1).limit(4).all()

# 分頁查詢, 每頁3個, 查詢第2頁的數據 ?paginate(頁碼, 每頁條數)
pn = User.query.paginate(2, 3)
pn.pages 總頁數 ?pn.page 當前頁碼 pn.items 當前頁的數據 ?pn.total 總條數

# 查詢每個年齡的人數 ? ?select age, count(name) from t_user group by age ?分組聚合
from sqlalchemy import func
data = db.session.query(User.age, func.count(User.id).label("count")).group_by(User.age).all()
for item in data:
? ? # print(item[0], item[1])
? ? print(item.age, item.count) ?# 建議通過label()方法給字段起別名, 以屬性方式獲取數據

# 只查詢所有人的姓名和郵箱 ?優化查詢 ? User.query.all() ?# 相當于select *
from sqlalchemy.orm import load_only
data = User.query.options(load_only(User.name, User.email)).all() ?# flask-sqlalchem的語法
for item in data:
? ? print(item.name, item.email)

data = db.session.query(User.name, User.email).all() ?# sqlalchemy本體的語法
for item in data:
? ? print(item.name, item.email)

更新數據
flask-sqlalchemy 提供了兩種更新數據的方案?

先查詢, 再更新
對應SQL中的 先select, 再update
基于過濾條件的更新 (推薦方案)
對應SQL中的 update xx where xx = xx (也稱為 update子查詢 )
先查詢, 再更新
這種方式的缺點

查詢和更新分兩條語句, 效率低
如果并發更新, 可能出現更新丟失問題(Lost Update)

class Goods(db.Model):__tablename__ = 't_good'id = db.Column(db.Integer, primary_key = True)name = db.Column(db.String(20), unique=True)count = db.Column(db.Integer)
@app.route('/addgood')
def addgood():goods = Goods(name='方便面', count=10)db.session.add(goods)db.session.commit()return "success"
@app.route('/updategood')
def updategood():goods = Goods.query.filter(Goods.name=='方便面').first()goods.count = goods.count - 1db.session.commit()return "success"

基于過濾條件的更新
這種方式的優點:

一條語句, 被網絡IO影響程度低, 執行效率更高
查詢和更新在一條語句中完成, 單條SQL具有原子性, 不會出現更新丟失問題
會對滿足過濾條件的所有記錄進行更新, 可以實現批量更新處理
操作步驟如下:

配合 查詢過濾器filter() 和 更新執行器update() 進行數據更新
提交會話

@app.route('/updategood2')
def updategood2():Goods.query.filter(Goods.name=='方便面').update({'count':Goods.count-1})db.session.commit()return "success"

刪除數據
類似更新數據, 也存在兩種刪除數據的方案
先查詢, 再刪除

對應SQL中的 先select, 再delete
基于過濾條件的刪除 (推薦方案)

對應SQL中的 delete xx where xx = xx (也稱為 delete子查詢 )
這種方式的缺點:

查詢和刪除分兩條語句, 效率低

@app.route('/deletegood')
def deletegood():goods = Goods.query.filter(Goods.name=='方便面').first()db.session.delete(goods)db.session.commit()return "success"

基于過濾條件的刪除
這種方式的優點:

一條語句, 被網絡IO影響程度低, 執行效率更高
會對滿足過濾條件的所有記錄進行刪除, 可以實現批量刪除處理
操作步驟如下:

配合 查詢過濾器filter() 和 刪除執行器delete() 進行數據刪除
提交會話

@app.route('/deletegood2')
def deletegood2():Goods.query.filter(Goods.name=='方便面').delete()db.session.commit()return "success"

增刪改操作都需要提交會話, 對應事務中進行數據庫變化后提交事務
刷新數據
Session 被設計為數據操作的執行者, 會先將操作產生的數據保存到內存中
在執行 flush刷新操作 后, 數據操作才會同步到數據庫中
有兩種情況下會 隱式執行刷新操作
提交會話
執行查詢操作 (包括 update 和 delete 子查詢)
開發者也可以 手動執行刷新操作 session.flush()
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 相關配置
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test31'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True
db = SQLAlchemy(app)

# 構建模型類 ?
class Goods(db.Model):
? ? __tablename__ = 't_good' ?
? ? id = db.Column(db.Integer, primary_key=True)?
? ? name = db.Column(db.String(20), unique=True)?
? ? count = db.Column(db.Integer) ?

@app.route('/')
def purchase():

? ? goods = Goods(name='方便面', count=20)
? ? db.session.add(goods)
? ? # 主動執行flush操作, 立即執行SQL操作(數據庫同步)
? ? db.session.flush()

? ? # Goods.query.count() ?# 查詢操作會自動執行flush操作
? ? db.session.commit() ?# 提交會話會自動執行flush操作

? ? return "index"

if __name__ == '__main__':
? ? db.drop_all()
? ? db.create_all()

? ? app.run(debug=True)

多表查詢

案例中包含兩個模型類: User用戶模型 和 Address地址模型, 并且一個用戶可以有多個地址, 兩張表之間存在一對多關系

class Address(db.Model):__tablename__='t_adr'id = db.Column(db.Integer, primary_key=True)detail = db.Column(db.String(20))user_id = db.Column(db.Integer)
@app.route('/addadr')
def addadr():adr1 = Address(detail='中關村3號', user_id=1)adr2 = Address(detail='華強北5號', user_id=1)db.session.add_all([adr2, adr1])db.session.commit()return "success"

關聯查詢
關聯查詢步驟: (以主查從為例)

先查詢主表數據
再通過外鍵字段查詢 關聯的從表數據

@app.route('/queryadr')
def queryadr():user1 = User.query.filter_by(name='zs').first()adrs = Address.query.filter_by(user_id=user1.id).all()for adr in adrs:print(adr.detail)return "success"

連接查詢

開發中有 聯表查詢需求 時, 一般會使用 join連接查詢
sqlalchemy 也提供了對應的查詢語法
db.session.query(主表模型字段1, 主表模型字段2, 從表模型字段1, xx.. ).join(從表模型類, 主表模型類.主鍵 == 從表模型類.外鍵)
1
join語句 屬于查詢過濾器, 返回值也是 BaseQuery 類型對象

@app.route('/queryadr2')
def queryadr2():data = db.session.query(User.id, Address.detail).join(Address, User.id==Address.user_id).filter(User.name=='zs').all()for item in data:print(item.detail, item.id)return "success"

關聯查詢的性能優化

通過前邊的學習, 可以發現 無論使用 外鍵 還是 關系屬性 查詢關聯數據, 都需要查詢兩次, 一次查詢用戶數據, 一次查詢地址數據
兩次查詢就需要發送兩次請求給數據庫服務器, 如果數據庫和web應用不在一臺服務器中, 則 網絡IO會對查詢效率產生一定影響
可以考慮使用 連接查詢 join 使用一條語句就完成關聯數據的查詢
# 使用join語句優化關聯查詢
adrs = Address.query.join(User, Address.user_id == User.id).filter(User.name == '張三').all() ?# 列表中包含地址模型對象

Session機制
生命周期
flask-sqlalchemy 對于 sqlalchemy本體 的 Session 進行了一定的封裝:
Session的生命周期和請求相近?

請求中的首次數據操作會創建Session
整個請求過程中使用的Session為同一個, 并且線程隔離
請求結束時會自動銷毀Session(釋放內存)

Session和事務

Session中可以包含多個事務, 提交事務失敗后, 會自動執行SQL的回滾操作
同一個請求中, 想要在前一個事務失敗的情況下創建新的事務, 必須先手動回滾事務 Session.rollback
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 相關配置
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/toutiao'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True
db = SQLAlchemy(app)

# 構建模型類?
class User(db.Model):
? ? __tablename__ = 't_user' ?
? ? id = db.Column(db.Integer, primary_key=True) ?
? ? name = db.Column('username', db.String(20), unique=True)?
? ? age = db.Column(db.Integer, default=0, index=True) ?

@app.route('/')
def index():

? ? """事務1"""
? ? try:
? ? ? ? user1 = User(name='zs', age=20)
? ? ? ? db.session.add(user1)
? ? ? ? db.session.commit()
? ? except BaseException:
? ? ? ? # 手動回滾 ? 同一個session中, 前一個事務如果失敗, 必須手動回滾, 否則無法創建新的事務
? ? ? ? db.session.rollback()

? ? """事務2"""
? ? user1 = User(name='lisi', age=30)
? ? db.session.add(user1)
? ? db.session.commit()

? ? return "index"

if __name__ == '__main__':
? ? """為了進行測試, 首次運行 建表并添加一條測試數據后, 注釋下方代碼, 并重新運行測試"""

? ? # 重置所有繼承自db.Model的表
? ? # db.drop_all()
? ? # db.create_all()

? ? # 添加一條測試數據
? ? # user1 = User(name='zs', age=20)
? ? # db.session.add(user1)
? ? # db.session.commit()

? ? app.run(debug=True)

數據遷移

flask-migrate組件 為flask-sqlalchemy提供了數據遷移功能, 以便進行數據庫升級, 如增加字段、修改字段類型等
安裝組件 pip install flask-migrate

# hm_數據遷移.py?

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrateapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:yu201541010@127.0.0.1:3306/pythontest'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Falsedb = SQLAlchemy(app)
#遷移組件初始化
Migrate(app, db)
class User(db.Model):__tablename__ ='t_user'id = db.Column(db.Integer, primary_key=True)name = db.Column('username', db.String(20), unique=True)@app.route('/')
def index():return "index"if __name__ =='__main__':app.run()

執行遷移命令

終端進入當前文件目錄下，注意export命令后面等號不能有空格

1. export FLASK_APP=hm_數據遷移.py ?# 設置環境變量指定啟動文件
2. flask db init ?# 生成遷移文件夾 ?只執行一次
3. flask db migrate ?# ?成遷移版本, 保存到遷移文件夾中
4. flask db upgrade ?# 執行遷移

執行遷移命令前需要先設置環境變量指定啟動文件