網(wǎng)站首頁 編程語言 正文
一： 基本使用：
1：環(huán)境的安裝：
pip install flask-sqlalchemy
pip install pymysql
2：組件初始化：
2.1： 基本的配置
1: 首先先安裝兩個依賴的包。
2：配置數(shù)據(jù)庫的連接：app.config[‘SQLALCHEMY_DATABASE_URI’] = “mysql://root:mysql@192.168.44.128:3306/test39”
3：關(guān)閉數(shù)據(jù)庫的跟蹤：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中數(shù)據(jù)庫客戶端： MySqldb
python3中數(shù)據(jù)庫客戶端：pymysql
解決方案一：讓python2和python3的包進行轉(zhuǎn)換。
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://賬號:密碼@數(shù)據(jù)庫ip地址:端口號/數(shù)據(jù)庫名"
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
# app.config['SQLALCHEMY_BINDS'] = {}
# 關(guān)閉數(shù)據(jù)庫修改跟蹤操作[提高性能]：
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 開啟輸出底層執(zhí)行的sql語句
app.config['SQLALCHEMY_ECHO'] = True
# 開啟數(shù)據(jù)庫的自動提交功能[一般不使用]
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：結(jié)合工廠方法進行配置：
1： 數(shù)據(jù)庫配置信息存放在環(huán)境類中加載。
2：由于數(shù)據(jù)庫對象和app對象不一定誰先創(chuàng)建，所以可以先創(chuàng)建數(shù)據(jù)庫對象，等app對象創(chuàng)建之后再進行關(guān)聯(lián)。
3：進行關(guān)聯(lián)的函數(shù)是：數(shù)據(jù)庫對象調(diào)用自己的init_app()方法。需要傳入app對象。
settings中配置：
# 開發(fā)環(huán)境
class DevelopmentConfig(BaseConfig):
"""開發(fā)環(huán)境配置類"""
DEBUG = True
# SQL數(shù)據(jù)庫連接信息
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:mysql@192.168.243.157:3306/test39"
# 關(guān)閉數(shù)據(jù)庫修改跟蹤操作 【提高性能】
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 開啟輸出底層執(zhí)行sql語句
SQLALCHEMY_ECHO = True
主模塊：
from flask import Flask, make_response, Response, request, current_app
from settings import config_dict
from flask_sqlalchemy import SQLAlchemy
# 延后加載
# 創(chuàng)建了數(shù)據(jù)庫，此時數(shù)據(jù)庫對象還沒有跟app關(guān)聯(lián)
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：構(gòu)建模型類：
- 模型類必須繼承 db.Model, 其中 db 指對應(yīng)的組件對象
- 表名默認為類名小寫, 可以通過 __tablename__類屬性 進行修改
- 類屬性對應(yīng)字段, 必須是通過 db.Column() 創(chuàng)建的對象
- 可以通過 create_all() 和 drop_all()方法 來創(chuàng)建和刪除所有模型類對應(yīng)的表
- 注意點: 如果沒有給對應(yīng)字段的類屬性設(shè)置default參數(shù), 且添加數(shù)據(jù)時也沒有給該字段賦值, 則 sqlalchemy會給該字段設(shè)置默認值 None
<模型類創(chuàng)建案例>
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 相關(guān)配置
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
# 創(chuàng)建組件對象
db = SQLAlchemy(app)
# 構(gòu)建模型類 類->表 類屬性->字段 實例對象->記錄
class User(db.Model):
__tablename__ = 't_user' # 設(shè)置表名, 表名默認為類名小寫
id = db.Column(db.Integer, primary_key=True) # 設(shè)置主鍵, 默認自增
name = db.Column('username', db.String(20), unique=True) # 設(shè)置字段名 和 唯一約束
age = db.Column(db.Integer, default=10, index=True) # 設(shè)置默認值約束 和 索引
if __name__ == '__main__':
# 刪除所有繼承自db.Model的表
db.drop_all()
# 創(chuàng)建所有繼承自db.Model的表
db.create_all()
app.run(host="0.0.0.0", port=8000, debug=True)
二：數(shù)據(jù)操作：
1:增加數(shù)據(jù)：
1：給模型對象設(shè)置數(shù)據(jù) 可以通過 初始化參數(shù) 或者 賦值屬性 兩種方式
2：session.add(模型對象) 添加單條數(shù)據(jù)到會話中, session.add_all(列表) 添加多條數(shù)據(jù)到會話中
3：sqlalchemy 會 自動創(chuàng)建事務(wù), 并將數(shù)據(jù)操作包含在事務(wù)中, 提交會話時就會提交事務(wù)，事務(wù)提交失敗會自動回滾。
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 相關(guān)配置
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
# 創(chuàng)建組件對象
db = SQLAlchemy(app)
# 構(gòu)建模型類
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():
"""增加數(shù)據(jù)"""
# 1.創(chuàng)建模型對象
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.提交會話 (會提交事務(wù))
# sqlalchemy會自動創(chuàng)建隱式事務(wù)
# 事務(wù)失敗會自動回滾
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：查詢數(shù)據(jù)：
1:數(shù)據(jù)的準備工作：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 配置數(shù)據(jù)庫連接
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:mysql@192.168.44.128:3306/test39"
# 配置取消數(shù)據(jù)庫跟蹤
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 配置數(shù)據(jù)庫輸出SQL語句
app.config['SQLALCHEMY_ECHO'] = True
# 創(chuàng)建數(shù)據(jù)庫對象
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()
# # 創(chuàng)建所有表
# db.create_all()
# # 添加測試數(shù)據(jù)
# 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)
#
# # 一次添加多條數(shù)據(jù)
# 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:查詢所有的用戶數(shù)據(jù)：
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的三種方式：
#<方案一>：根據(jù)id查詢，返回模型類對象
user4 = User.query.get(4)
print("第四個用戶的信息是{}".format(user4))
# <方案二>：等值過濾器 關(guān)鍵字實參設(shè)置字段值 返回BaseQuery對象
user4 = User.query.filter_by(id=4).first()
print("第四個用戶的信息是{}".format(user4))
# <方案三>：使用復(fù)雜過濾器，返回BaseQuery對象
user4 = User.query.filter(User.id == 4).first()
print("第四個用戶的信息是{}".format(user4))
# 5:查詢用戶名字，開始，結(jié)尾，包含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結(jié)尾的用戶{}".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結(jié)尾的所有用戶
users = User.query.filter(or_(User.age==25, User.email.endswith('com'))).all()
print("age是25或者email以com結(jié)尾的所有用戶 : {}".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位的數(shù)據(jù)
users = User.query.order_by(User.age).offset(1).limit(4).all()
print("查詢年齡從小到大第2-5位的數(shù)據(jù): {}".format(users))
# 12: 分頁查詢, 每頁3個, 查詢第2頁的數(shù)據(jù) paginate(頁碼, 每頁條數(shù))
pn = User.query.paginate(2, 3)
print("總頁數(shù)是：", pn.pages)
print("當前頁：", pn.page)
print("當前頁的數(shù)據(jù)：", pn.items)
print("當前頁的總條數(shù)", pn.total)
# 13: 查詢每個年齡段的人數(shù)：(分組聚合)
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()方法給字段起別名, 以屬性方式獲取數(shù)據(jù)
# 14：只查詢所有人的姓名和郵箱,這種相當于全表查詢，效率非常低。
data = db.session.query(User.name, User.email).all()
for item in data:
print(item.name, item.email)
# 15：優(yōu)化查詢
data = User.query.options(load_only(User.name, User.email)).all()
for item in data:
print(item.name, item.email)
return "index"
3：修改數(shù)據(jù)：
1: 推薦采用方案二：
2: 一條語句, 被網(wǎng)絡(luò)IO影響程度低, 執(zhí)行效率更高
3：查詢和更新在一條語句中完成, 單條SQL具有原子性, 不會出現(xiàn)更新丟失問題
4：會對滿足過濾條件的所有記錄進行更新, 可以實現(xiàn)批量更新處理
方案一：先查詢再更新：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 相關(guān)配置
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
# 創(chuàng)建組件對象
db = SQLAlchemy(app)
# 構(gòu)建模型類 商品表
class Goods(db.Model):
__tablename__ = 't_good' # 設(shè)置表名
id = db.Column(db.Integer, primary_key=True) # 設(shè)置主鍵
name = db.Column(db.String(20), unique=True) # 商品名稱
count = db.Column(db.Integer) # 剩余數(shù)量
@app.route('/')
def purchase():
"""購買商品"""
# 更新方式1: 先查詢后更新
# 缺點: 并發(fā)情況下, 容易出現(xiàn)更新丟失問題 (Lost Update)
# 1.執(zhí)行查詢語句, 獲取目標模型對象
goods = Goods.query.filter(Goods.name == '方便面').first()
# 2.對模型對象的屬性進行賦值 (更新數(shù)據(jù))
goods.count = goods.count - 1
# 3.提交會話
db.session.commit()
return "index"
if __name__ == '__main__':
# 刪除所有繼承自db.Model的表
db.drop_all()
# 創(chuàng)建所有繼承自db.Model的表
db.create_all()
# 添加一條測試數(shù)據(jù)
goods = Goods(name='方便面', count=1)
db.session.add(goods)
db.session.commit()
app.run(host="0.0.0.0", port=8000, debug=True)
方案二：配合查詢過濾器filter() 和 更新執(zhí)行器update() 進行數(shù)據(jù)更新
Goods.query.filter(Goods.name == '方便面').update({'count': Goods.count - 1})
db.session.commit()
4：刪除數(shù)據(jù)：
方案一：
# 方式1: 先查后刪除
goods = Goods.query.filter(Goods.name == '方便面').first()
# 刪除數(shù)據(jù)
db.session.delete(goods)
# 提交會話 增刪改都要提交會話
db.session.commit()
方案二：
# 方式2: delete子查詢
Goods.query.filter(Goods.name == '方便面').delete()
# 提交會話
db.session.commit()
三：高級機制：
1：刷新數(shù)據(jù)：
1：Session 被設(shè)計為數(shù)據(jù)操作的執(zhí)行者, 會先將操作產(chǎn)生的數(shù)據(jù)保存到內(nèi)存中。
2: 在執(zhí)行 flush刷新操作 后, 數(shù)據(jù)操作才會同步到數(shù)據(jù)庫中。
3：隱式刷新操作：1：提交會話 2：查詢操作（包括更新和刪除中的子查詢）。
4：手動刷新：session.flush()
刷新機制的理解：
答：刷新機制就是通過事務(wù)，將SQl語句執(zhí)行一遍，然后將執(zhí)行結(jié)果存儲在變量中，但是數(shù)據(jù)庫做回滾操作。導(dǎo)致變量中有了新值，但是數(shù)據(jù)庫卻沒有改變。
goods = Goods(name='方便面', count=20)
db.session.add(goods)
# 主動執(zhí)行flush操作, 立即執(zhí)行SQL操作(數(shù)據(jù)庫同步)
print(goods.id) # 此時是None
db.session.flush()
print(goods.id) # 此時是1
# Goods.query.count() # 查詢操作會自動執(zhí)行flush操作
db.session.commit() # 提交會話會自動執(zhí)行flush操作
2：多表查詢：
2.1：外鍵關(guān)聯(lián)查詢：
生成主表對象后，必須刷新數(shù)據(jù)庫，否則后面無法使用主表對象的屬性。
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：添加關(guān)聯(lián)數(shù)據(jù)：
def index():
"""添加并關(guān)聯(lián)數(shù)據(jù)"""
user1 = User(name='張三')
db.session.add(user1)
db.session.flush() # 必須刷新，不然后面的user1.id是None
adr1 = Address(detail='中關(guān)村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：關(guān)聯(lián)查詢：
# 1.先根據(jù)姓名查找到主表主鍵
user1 = User.query.filter_by(name='張三').first()
# 2.再根據(jù)主鍵到從表查詢關(guān)聯(lián)地址
adrs = Address.query.filter_by(user_id=user1.id).all()
for adr in adrs:
print(adr.detail)
原文鏈接：https://blog.csdn.net/qq_41341757/article/details/109462158
相關(guān)推薦
- 2023-07-28 elementui 通過事件觸發(fā)動畫
- 2022-10-20 Android?Flutter實現(xiàn)自定義下拉刷新組件_Android
- 2022-07-28 C++超詳細講解逗號操作符_C 語言
- 2022-07-27 Python?pyecharts?Line折線圖的具體實現(xiàn)_python
- 2022-06-07 python中字符串String及其常見操作指南(方法、函數(shù))_python
- 2022-06-06 Docker鏡像構(gòu)建之docker?commit的使用_docker
- 2022-01-03 踩坑解決mongoose對已經(jīng)存在的集合查詢，查詢條件不起限制作用的問題
- 2023-01-15 Python?pytest.main()運行測試用例_python
- 最近更新
-
- window11 系統(tǒng)安裝 yarn
- 超詳細win安裝深度學習環(huán)境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎(chǔ)操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區(qū)別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權(quán)
- redisson分布式鎖中waittime的設(shè)
- maven:解決release錯誤：Artif
- restTemplate使用總結(jié)
- Spring Security之安全異常處理
- MybatisPlus優(yōu)雅實現(xiàn)加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務(wù)發(fā)現(xiàn)-Nac
- Spring Security之基于HttpR
- Redis 底層數(shù)據(jù)結(jié)構(gòu)-簡單動態(tài)字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應(yīng)用詳解
- 聊聊消息隊列，發(fā)送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速