首頁 > 軟體

Flask-Sqlalchemy的基本使用詳解

2022-06-08 14:02:08

一： 基本使用：

1：環境的安裝：

pip install flask-sqlalchemy
pip install pymysql

2：元件初始化：

2.1： 基本的設定

1: 首先先安裝兩個依賴的包。
2：設定資料庫的連線：app.config[‘SQLALCHEMY_DATABASE_URI’] = “mysql://root:mysql@192.168.44.128:3306/test39”
3：關閉資料庫的跟蹤：app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False
4：開啟輸出sql語句：app.config[‘SQLALCHEMY_ECHO’] = True
5：兩種處理python2和python3的名字不一致問題。

from flask import Flask
from flask_restful import Api, Resource
from flask_sqlalchemy import SQLAlchemy

import pymysql
pymysql.install_as_MySQLdb()

"""
python2中資料庫使用者端： MySqldb
python3中資料庫使用者端：pymysql
解決方案一：讓python2和python3的包進行轉換。
import pymysql
pymysql.install_as_MySQLdb()

方案二：表示只使用python3的包，不使用python2的包
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+pymysql://root:mysql@192.168.44.128:3306/test39"
"""

app = Flask(__name__)
db = SQLAlchemy(app)

# app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://賬號:密碼@資料庫ip地址:埠號/資料庫名"
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
# app.config['SQLALCHEMY_BINDS'] = {}

# 關閉資料庫修改跟蹤操作[提高效能]：
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 開啟輸出底層執行的sql語句
app.config['SQLALCHEMY_ECHO'] = True

# 開啟資料庫的自動提交功能[一般不使用]
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True

@app.route('/')
def hello_word():
    return "hello, word"

if __name__ == '__main__':
    print(app.url_map)
    app.run(host='0.0.0.0', port= 8000, debug=True)

2.2：結合工廠方法進行設定：

1： 資料庫設定資訊存放在環境類中載入。
2：由於資料庫物件和app物件不一定誰先建立，所以可以先建立資料庫物件，等app物件建立之後再進行關聯。
3：進行關聯的函數是：資料庫物件呼叫自己的init_app()方法。需要傳入app物件。

settings中設定：

# 開發環境
class DevelopmentConfig(BaseConfig):
    """開發環境設定類"""
    DEBUG = True
    # SQL資料庫連線資訊
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:mysql@192.168.243.157:3306/test39"
    # 關閉資料庫修改跟蹤操作 【提高效能】
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # 開啟輸出底層執行sql語句
    SQLALCHEMY_ECHO = True

主模組：

from flask import Flask, make_response, Response, request, current_app
from settings import config_dict
from flask_sqlalchemy import SQLAlchemy

# 延後載入
# 建立了資料庫，此時資料庫物件還沒有跟app關聯
db = SQLAlchemy()

# 定義一個工廠方法：
def create_app(config_name):
    app = Flask(__name__)

    config_class = config_dict[config_name]
    app.config.from_object(config_class)

    app.config.from_envvar('CONFIG', silent=True)

    # 懶載入
    db.init_app(app)
    return app

app = create_app("dev")

@app.route('/login')
def login():
    return ""

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000, debug=True)

3：構建模型類：

  • 模型類必須繼承 db.Model, 其中 db 指對應的元件物件
  • 表名預設為類名小寫, 可以通過 __tablename__類屬性 進行修改
  • 類屬性對應欄位, 必須是通過 db.Column() 建立的物件
  • 可以通過 create_all() 和 drop_all()方法 來建立和刪除所有模型類對應的表
  • 注意點: 如果沒有給對應欄位的類屬性設定default引數, 且新增資料時也沒有給該欄位賦值, 則 sqlalchemy會給該欄位設定預設值 None

<模型類建立案例>

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 相關設定
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
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)  # 設定預設值約束 和 索引


if __name__ == '__main__':
    # 刪除所有繼承自db.Model的表
    db.drop_all()
    # 建立所有繼承自db.Model的表
    db.create_all()
    app.run(host="0.0.0.0", port=8000, debug=True)

二：資料操作：

1:增加資料：

1：給模型物件設定資料 可以通過 初始化引數 或者 賦值屬性 兩種方式
2：session.add(模型物件) 新增單條資料到對談中, session.add_all(列表) 新增多條資料到對談中
3：sqlalchemy 會 自動建立事務, 並將資料操作包含在事務中, 提交對談時就會提交事務，事務提交失敗會自動回滾。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 相關設定
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
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, index=True)


@app.route('/')
def index():
    """增加資料"""

    # 1.建立模型物件
    user1 = User(name='zs', age=20)
    # user1.name = 'zs'
    # user1.age = 20

    # 2.將模型物件新增到對談中
    db.session.add(user1)
    # 新增多條記錄
    # db.session.add_all([user1, user2, user3])

    # 3.提交對談 (會提交事務)
    # sqlalchemy會自動建立隱式事務
    # 事務失敗會自動回滾
    db.session.commit()

    return "index"


if __name__ == '__main__':
    db.drop_all()
    db.create_all()
    app.run(host="0.0.0.0", port=8000, debug=True)

2：查詢資料：

1:資料的準備工作：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 設定資料庫連線
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
# 設定取消資料庫跟蹤
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 設定資料庫輸出SQL語句
app.config['SQLALCHEMY_ECHO'] = True
# 建立資料庫物件
db = SQLAlchemy(app)

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


if __name__ == '__main__':
    # 刪除所有表
    # db.drop_all()
    # # 建立所有表
    # db.create_all()
    # # 新增測試資料
    # user1 = User(name='wang', email='wang@163.com', age=20)
    # user2 = User(name='zhang', email='zhang@189.com', age=33)
    # user3 = User(name='chen', email='chen@126.com', age=23)
    # user4 = User(name='zhou', email='zhou@163.com', age=29)
    # user5 = User(name='tang', email='tang@itheima.com', age=25)
    # user6 = User(name='wu', email='wu@gmail.com', age=25)
    # user7 = User(name='qian', email='qian@gmail.com', age=23)
    # user8 = User(name='liu', email='liu@itheima.com', age=30)
    # user9 = User(name='li', email='li@163.com', age=28)
    # user10 = User(name='sun', email='sun@163.com', age=26)
    #
    # # 一次新增多條資料
    # db.session.add_all([user1, user2, user3, user4, user5, user6, user7, user8, user9, user10])
    # db.session.commit()
    app.run(host="0.0.0.0",port=8000, debug=True)

2：進行查詢操作：

	# 1:查詢所有的使用者資料：
    users = User.query.all()
    print(type(users))
    print(users)

    # 2:查詢一共有多少個使用者：
    count = User.query.count()
    print("一共有{}個人".format(count))

    # 3:查詢第一個使用者資訊：
    user1 = User.query.first()
    print("第一個使用者的資訊是：{}".format(user1))

    # 4:查詢id為4的三種方式：
    #<方案一>：根據id查詢，返回模型類物件
    user4 = User.query.get(4)
    print("第四個使用者的資訊是{}".format(user4))

    # <方案二>：等值過濾器 關鍵字實參設定欄位值  返回BaseQuery物件
    user4 = User.query.filter_by(id=4).first()
    print("第四個使用者的資訊是{}".format(user4))

    # <方案三>：使用複雜過濾器，返回BaseQuery物件
    user4 = User.query.filter(User.id == 4).first()
    print("第四個使用者的資訊是{}".format(user4))

    # 5:查詢使用者名稱字，開始，結尾，包含n的使用者
    user = User.query.filter(User.name.startswith('n')).all()
    print("名字以n開頭的使用者{}".format(user))
    user = User.query.filter(User.name.endswith("n")).all()
    print("名字以n結尾的使用者{}".format(user))
    user = User.query.filter(User.name.contains("n")).all()
    print("名字中包含n的使用者：{}".format(user))

    # 6:查詢名字和郵箱都以li開頭的所有使用者[2種方式]
    users = User.query.filter(User.name.startswith('li'), User.email.startswith('li')).all()
    print("查詢名字和郵箱都以li開頭的所有使用者：{}".format(users))

    users = User.query.filter(and_(User.name.startswith('li'), User.email.startswith('li'))).all()
    print("查詢名字和郵箱都以li開頭的所有使用者：{}".format(users))

    # 7：查詢age是25或者email以com結尾的所有使用者
    users = User.query.filter(or_(User.age==25, User.email.endswith('com'))).all()
    print("age是25或者email以com結尾的所有使用者 : {}".format(users))

    # 8: 查詢名字不等於wang的所有使用者
    users = User.query.filter(User.name != "wang").all()
    print("名字不等於wang的所有使用者： {}".format(users))
    users= User.query.filter(not_(User.name=="wang")).all()
    print("名字不等於wang的所有使用者： {}".format(users))

    # 9: 查詢id是[1, 3, 5, 7, 9]的使用者
    users = User.query.filter(User.id.in_([1, 3, 5, 7, 9])).all()
    print("id是[1, 3, 5, 7, 9]的使用者: {}".format(users))

    # 10：所有使用者先按年齡從小到大, 再按id從大到小排序, 取前5個
    users = User.query.order_by(User.age, User.id.desc()).limit(5).all()
    print("所有使用者先按年齡從小到大, 再按id從大到小排序, 取前5個: {}".format(users))

# 11:查詢年齡從小到大第2-5位的資料
    users = User.query.order_by(User.age).offset(1).limit(4).all()
    print("查詢年齡從小到大第2-5位的資料: {}".format(users))

    # 12: 分頁查詢, 每頁3個, 查詢第2頁的資料  paginate(頁碼, 每頁條數)
    pn = User.query.paginate(2, 3)
    print("總頁數是：", pn.pages)
    print("當前頁：", pn.page)
    print("當前頁的資料：", pn.items)
    print("當前頁的總條數", pn.total)

    # 13: 查詢每個年齡段的人數：(分組聚合)
    data = db.session.query(User.age, func.count(User.id)).group_by(User.age).all()
    for item in data:
        print(item[0], item[1])
    # 注意可以給列起別名，但是windows下會報錯，linux下不會報錯。
    # 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()方法給欄位起別名, 以屬性方式獲取資料

    # 14：只查詢所有人的姓名和郵箱,這種相當於全表查詢，效率非常低。
    data = db.session.query(User.name, User.email).all()
    for item in data:
        print(item.name, item.email)

    # 15：優化查詢
    data = User.query.options(load_only(User.name, User.email)).all()
    for item in data:
        print(item.name, item.email)
    return "index"

3：修改資料：

1: 推薦採用方案二：
2: 一條語句, 被網路IO影響程度低, 執行效率更高
3：查詢和更新在一條語句中完成, 單條SQL具有原子性, 不會出現更新丟失問題
4：會對滿足過濾條件的所有記錄進行更新, 可以實現批次更新處理

方案一：先查詢再更新：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 相關設定
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
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():
    """購買商品"""

    # 更新方式1: 先查詢後更新
    # 缺點: 並行情況下, 容易出現更新丟失問題 (Lost Update)
    # 1.執行查詢語句, 獲取目標模型物件
    goods = Goods.query.filter(Goods.name == '方便麵').first()
    # 2.對模型物件的屬性進行賦值 (更新資料)
    goods.count = goods.count - 1
    # 3.提交對談
    db.session.commit()

    return "index"


if __name__ == '__main__':
    # 刪除所有繼承自db.Model的表
    db.drop_all()
    # 建立所有繼承自db.Model的表
    db.create_all()

    # 新增一條測試資料
    goods = Goods(name='方便麵', count=1)
    db.session.add(goods)
    db.session.commit()
    app.run(host="0.0.0.0", port=8000, debug=True)

方案二：配合查詢過濾器filter() 和 更新執行器update() 進行資料更新

  Goods.query.filter(Goods.name == '方便麵').update({'count': Goods.count - 1})
    db.session.commit()

4：刪除資料：

方案一：

 # 方式1: 先查後刪除
    goods = Goods.query.filter(Goods.name == '方便麵').first()
    # 刪除資料
    db.session.delete(goods)
    # 提交對談 增刪改都要提交對談
    db.session.commit()

方案二：

# 方式2: delete子查詢
    Goods.query.filter(Goods.name == '方便麵').delete()
    # 提交對談
    db.session.commit()

三：高階機制：

1：重新整理資料：

1：Session 被設計為資料操作的執行者, 會先將操作產生的資料儲存到記憶體中。
2: 在執行 flush重新整理操作 後, 資料操作才會同步到資料庫中。
3：隱式重新整理操作：1：提交對談 2：查詢操作（包括更新和刪除中的子查詢）。
4：手動重新整理：session.flush()

重新整理機制的理解：

答：重新整理機制就是通過事務，將SQl語句執行一遍，然後將執行結果儲存在變數中，但是資料庫做回滾操作。導致變數中有了新值，但是資料庫卻沒有改變。

 	goods = Goods(name='方便麵', count=20)
    db.session.add(goods)
    # 主動執行flush操作, 立即執行SQL操作(資料庫同步)
    print(goods.id)  # 此時是None
    db.session.flush()
	print(goods.id) # 此時是1
    # Goods.query.count()  # 查詢操作會自動執行flush操作
    db.session.commit()  # 提交對談會自動執行flush操作

2：多表查詢：

2.1：外來鍵關聯查詢：

生成主表物件後，必須重新整理資料庫，否則後面無法使用主表物件的屬性。

1：主從表的定義：

# 使用者表  一   一個使用者可以有多個地址
class User(db.Model):
    __tablename__ = 't_user'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(20))


# 地址表   多
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)  # 定義外來鍵

2：新增關聯資料：

def index():
    """新增並關聯資料"""

    user1 = User(name='張三')
    db.session.add(user1)
    db.session.flush()  # 必須重新整理，不然後面的user1.id是None
    adr1 = Address(detail='中關村3號', user_id=user1.id)
    adr2 = Address(detail='華強北5號', user_id=user1.id)
    db.session.add_all([adr1, adr2])
    db.session.commit()
    return "index"

3：關聯查詢：

# 1.先根據姓名查詢到主表主鍵
    user1 = User.query.filter_by(name='張三').first()

    # 2.再根據主鍵到從表查詢關聯地址
    adrs = Address.query.filter_by(user_id=user1.id).all()
    for adr in adrs:
        print(adr.detail)

到此這篇關於Flask-Sqlalchemy的基本使用詳解的文章就介紹到這了,更多相關Flask-Sqlalchemy 使用內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！


相關文章

  • 德漢姆中學就讀條件 德漢姆中學就讀條件

    <em>Mac</em>Book项目 2009年学校开始实施<em>Mac</em>Book项目,所有师生配备一本<em>Mac</em>Book,并同步更新了校园无线网络。学校每周进行电脑技术更新,每月发送技术支持资料,极大改变了教学及学习方式。因此2011

    2021-06-01 09:32:01

  • 黑科技的輕簡出行，告別「電量焦慮症」：Anker氮化鎵超能充系列 黑科技的輕簡出行，告別「電量焦慮症」：Anker氮化鎵超能充系列

    综合看Anker超能充系列的性价比很高，并且与不仅和iPhone12/苹果<em>Mac</em>Book很配，而且适合多设备充电需求的日常使用或差旅场景，不管是安卓还是Switch同样也能用得上它，希望这次分享能给准备购入充电器的小伙伴们有所

    2021-06-01 09:31:42

  • 吳亦凡廠牌首秀，L4WUDU居然忘詞了，Rapper和明星比還是有差距！ 吳亦凡廠牌首秀，L4WUDU居然忘詞了，Rapper和明星比還是有差距！

    除了L4WUDU与吴亦凡已经多次共事，成为了明面上的厂牌成员，吴亦凡还曾带领20XXCLUB全队参加2020年的一场音乐节，这也是20XXCLUB首次全员合照，王嗣尧Turbo、陈彦希Regi、<em>Mac</em> Ova Seas、林渝植等人全部出场。然而让

    2021-06-01 09:31:34

  • IPFS、Chia、Bzz和ICP挖礦該怎麼選？哪一個更有優勢？ IPFS、Chia、Bzz和ICP挖礦該怎麼選？哪一個更有優勢？

    目前应用IPFS的机构：1 谷歌<em>浏览器</em>支持IPFS分布式协议 2 万维网 (历史档案博物馆)数据库 3 火狐<em>浏览器</em>支持 IPFS分布式协议 4 EOS 等数字货币数据存储 5 美国国会图书馆，历史资料永久保存在 IPFS 6 加

    2021-06-01 09:31:24

  • 有哪些事是買了雪佛蘭後才知道的？美事偷著樂，開拓者車主隨聊 有哪些事是買了雪佛蘭後才知道的？美事偷著樂，開拓者車主隨聊

    开拓者的车机是兼容苹果和<em>安卓</em>，虽然我不怎么用，但确实兼顾了我家人的很多需求：副驾的门板还配有解锁开关，有的时候老婆开车，下车的时候偶尔会忘记解锁，我在副驾驶可以自己开门：第二排设计很好，不仅配置了一个很大的

    2021-06-01 09:30:48

  • iPhone12在618最新定價，跌價幅度超過1400元，還等iPhone13嗎？ iPhone12在618最新定價，跌價幅度超過1400元，還等iPhone13嗎？

    不仅是<em>安卓</em>手机，苹果手机的降价力度也是前所未有了，iPhone12也“跳水价”了，发布价是6799元，如今已经跌至5308元，降价幅度超过1400元，最新定价确认了。iPhone12是苹果首款5G手机，同时也是全球首款5nm芯片的智能机，它

    2021-06-01 09:30:45

IT145.com E-mail:sddin#qq.com