SQLAlchemy和Flask-SQLAlchemy

1、ORM 與 SQLAlchemy 簡介

ORM 全稱 Object Relational Mapping, 翻譯過來叫對象關係映射。簡單的說，ORM 將數據庫中的表與面嚮對象語言中的類創建了一種對應關係。這樣，咱們要操做數據庫，數據庫中的表或者表中的一條記錄就能夠直接經過操做類或者類實例來完成。html

SQLAlchemy 是Python 社區最知名的 ORM 工具之一，爲高效和高性能的數據庫訪問設計，實現了完整的企業級持久模型。mysql

2、SQLAlchemy的使用

一、建立單表結構：sql

from sqlalchemy.ext.declarative import declarative_base # 導入基類 from sqlalchemy import Column, Integer, String # 數據類型 # Base = ORM基類 - 要按照ORM的規則定義你的類
Base = declarative_base()

class Users(Base): __tablename__ = "user"
    # 建立ID數據字段 , 那麼ID是否是一個數據列呢? 也就是說建立ID字段 == 建立ID數據列
    # id = Column(數據類型,索引,主鍵,外鍵,等等)
    id = Column(Integer, primary_key=True, autoincrement=True,index=True) name = Column(String(32), nullable=False)  # nullable=False 不能爲空


# 打開數據庫的鏈接 -- 建立數據庫引擎
from sqlalchemy import create_engine # 建立數據庫引擎 # engine = create_engine("mysql://scott:tiger@hostname/dbname",encoding='latin1', echo=True)
engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8") Base.metadata.create_all(engine) # Base自動檢索全部繼承Base的ORM 對象 而且建立全部的數據表

2.單表的增刪改查：數據庫

from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from creatTable import Users # 建立引擎
engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8") Session = sessionmaker(engine) db_session = Session() # 1. 增長數據add(建立表結構的類名(字段名=添加的數據))
db_session.add(Users(name="ZWQ"))  # 至關於創建一條添加數據的sql語句
db_session.commit()  # 執行
db_session.close()   # 結束關閉

# 批量添加
db_session.add_all([Users(name="清風徐來"), Users(name="水波不興")]) db_session.commit() db_session.close() # 2.查詢 query(表結構的類名)
sqlres = db_session.query(Users) print(sqlres)  # 直接翻譯輸出對應的SQL查詢語句
 res = db_session.query(Users).all()  # 返回表中全部數據對象
print(res)# [<creatTable.Users object at 0x00000000038A1B00>,<creatTable.Users object at 0x00000000038A1B70>]

for u in res: print(u.id, u.name) res = db_session.query(Users).first()  # 取第一個，返回是對象
print(res.id, res.name) res = db_session.query(Users).filter(Users.id == 3).first()  # 返回符合條件查詢結果
print(res.name) res = db_session.query(Users).filter(Users.id <= 2, Users.name == "ZWQ").all() # filter中的條件能夠是模糊條件,多個條件
for u in res: print(u.id,u.name) # 3.更改數據 update({k:v})
res = db_session.query(Users).filter(Users.id == 1).update({"name":"DragonFire"}) print(res) db_session.commit() res = db_session.query(Users).update({"name":"ZWQ"})  # 所有修改，返回修改的數據個數
print(res) db_session.commit() # 4.刪除 delete()結合查詢條件刪除
res = db_session.query(Users).filter(Users.id == 1).delete()  # 刪除否合條件的數據，返回刪除數量
print(res) db_session.commit() res = db_session.query(Users).delete()  # 刪除表中全部數據，返回刪除數量
print(res) db_session.commit()

3.建立外鍵關聯的表結構：flask

from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import relationship class Student(Base): __tablename__ = "student" id = Column(Integer,primary_key=True) name = Column(String(32),nullable=False) sch_id = Column(Integer,ForeignKey("school.id"))  # 關聯的表的字段,表間的關係
 stu2sch = relationship("School",backref="sch2stu")  # 寫在哪邊那邊就是正向查詢，對象間的關係，backref(反向查詢)


class School(Base): __tablename__ = "school" id = Column(Integer,primary_key=True) name = Column(String(32),nullable=False) from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8") Base.metadata.create_all(engine)

四、外鍵關聯的表添加與查詢操做：session

from sqlalchemy.orm import sessionmaker
 from creatTableFk import engine from creatTableFk import Student,School # 導入建立表結構的類 Session = sessionmaker(engine) db_session = Session() # 1.添加數據
db_session.add(School(name="NCU")) db_session.commit() # relationship 正向添加
stu = Student(name="清風徐來",stu2sch=School(name="NCU")) db_session.add(stu) db_session.commit() # relationship 反向添加
sch = School(name="NCU") sch.sch2stu = [Student(name="YWB"),Student(name="CT")] db_session.add(sch) db_session.commit() # 2.查詢
res = db_session.query(Student).all() for stu in res: print(stu.name,stu.stu2sch.name)  # 正向跨表
 res = db_session.query(School).all() for sch in res: for stu in sch.sch2stu: print(sch.name,stu.name)  # 反向跨表

5.多對多的表的建立app

from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, String, Integer, ForeignKey, create_engine from sqlalchemy.orm import relationship Base = declarative_base() # 多對多關聯經過第三張表關聯，sqlalchemy要本身建立第三張表
class Girl(Base): __tablename__ = "girls" id = Column(Integer, primary_key=True) name = Column(String(32), nullable=False) g2b = relationship("Boy", backref="b2g", secondary="hotels") class Boy(Base): __tablename__ = "boys" id = Column(Integer, primary_key=True) name = Column(String(32), nullable=False) class Hotel(Base): __tablename__ = "hotels" id = Column(Integer, primary_key=True) boy_id = Column(Integer, ForeignKey("boys.id"), nullable=False) girl_id = Column(Integer, ForeignKey("girls.id"), nullable=False) engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8") Base.metadata.create_all(engine)

6.多對多表數據的添加與查詢：ide

from sqlalchemy.orm import sessionmaker from CreateTableM2M import engine Session = sessionmaker(engine) db_session = Session() from CreateTableM2M import Girl,Boy # 1.增長數據 # relationship 正向添加
g = Girl(name="ZLY",g2b=[Boy(name="ZWQ"),Boy(name="FSF")]) db_session.add(g) db_session.commit() # relationship 反向添加
b = Boy(name="ZS") b.b2g = [Girl(name="羅玉鳳"),Girl(name="娟兒"),Girl(name="芙蓉姐姐")] db_session.add(b) db_session.commit() # 2.查詢 # relationship 正向
res = db_session.query(Girl).all() for g in res: for b in g.g2b: print(g.name,b.name) # relationship 反向
res = db_session.query(Boy).all() for b in res: for g in b.b2g: print(b.name,g.name)

7. 高級ORM操做工具

 1 from CreateTable import Users,engine  2 from sqlalchemy.orm import sessionmaker  3 from CreateTableForeignKey import Student  4 
 5 Session = sessionmaker(engine)  6 db_session = Session()  7 
 8 # 查詢數據表操做
 9 # and or
 10 from sqlalchemy.sql import and_ , or_,desc  11 ret = db_session.query(Users).filter(and_(Users.id > 3, Users.name == 'DragonFire')).all()  12 ret = db_session.query(Users).filter(or_(Users.id < 2, Users.name == 'DragonFire')).all()  13 
 14 ret = db_session.query(Users).filter(  15  or_(  16         Users.id < 2,  17  and_(  18             Users.name == 'eric',  19             Users.id > 3
 20  ),  21         Users.name != ""
 22  )  23 )  24 print(ret)  25 select * from User where id<2 or (name="eric" and id>3) or extra != ""
 26 
 27 
 28 # 查詢全部數據
 29 r1 = db_session.query(User).all()  30 
 31 查詢數據 指定查詢數據列 加入別名  32 r2 = db_session.query(Student.name.label('username'), Student.id).first()  33 print(r2.id,r2.username) # 15 NBDragon
 34 
 35 # 表達式篩選條件
 36 r3 = db_session.query(User).filter(User.name == "DragonFire").all()  37 
 38 # 原生SQL篩選條件
 39 r4 = db_session.query(User).filter_by(name='DragonFire').all()  40 r5 = db_session.query(Users).filter_by(name='DragonFire').first()  41 
 42 字符串匹配方式篩選條件 並使用 order_by進行排序  43 r6 = db_session.query(Student).order_by(Student.name.desc()).all()  44 for i in r6:  45     print(i.id,i.name)  46 
 47 原生SQL查詢  48 from sqlalchemy.sql import text  49 r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='DragonFire').all()  50 
 51 # 篩選查詢列
 52 # query的時候咱們不在使用User ORM對象,而是使用User.name來對內容進行選取
 53 user_list = db_session.query(User.name).all()  54 print(user_list)  55 for row in user_list:  56     print(row.name)  57 
 58 # 別名映射 name as nick
 59 user_list = db_session.query(User.name.label("nick")).all()  60 print(user_list)  61 for row in user_list:  62     print(row.nick) # 這裏要寫別名了
 63 
 64 # 篩選條件格式
 65 user_list = db_session.query(User).filter(User.name == "DragonFire").all()  66 user_list = db_session.query(User).filter(User.name == "DragonFire").first()  67 user_list = db_session.query(User).filter_by(name="DragonFire").first()  68 for row in user_list:  69     print(row.nick)  70 
 71 # 複雜查詢
 72 from sqlalchemy.sql import text  73 user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="DragonFire")  74 
 75 # 查詢語句
 76 from sqlalchemy.sql import text  77 user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="DragonFire")  78 
 79 # 排序 :
 80 user_list = db_session.query(User).order_by(User.id).all()  81 user_list = db_session.query(User).order_by(User.id.desc()).all()  82 for row in user_list:  83     print(row.name,row.id)  84 
 85 #其餘查詢條件
 86 """
 87 ret = session.query(User).filter_by(name='DragonFire').all()  88 ret = session.query(User).filter(User.id > 1, User.name == 'DragonFire').all()  89 ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='DragonFire'))).all() 子查詢  90 from sqlalchemy import and_, or_  91 ret = session.query(User).filter(and_(User.id > 3, User.name == 'DragonFire')).all()  92 ret = session.query(User).filter(or_(User.id < 2, User.name == 'DragonFire')).all()  93 
 94 
 95 
 96 
 97 # 限制  98 ret = db_session.query(User)[1:2]  99 
100 # 排序 101 ret = db_session.query(User).order_by(User.name.desc()).all() 102 ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all() 103 
104 # 分組 105 from sqlalchemy.sql import func 106 
107 ret = db_session.query(User).group_by(User.extra).all() 108 ret = db_session.query( 109  func.max(User.id), 110  func.sum(User.id), 111  func.min(User.id)).group_by(User.name).all() 112 
113 ret = db_session.query( 114  func.max(User.id), 115  func.sum(User.id), 116  func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all() 117 """
118 
119 # 關閉鏈接
120 db_session.close() 121 
122 
123 ret = db_session.query(Student).filter(Student.id.between(1, 3)).all() # between 大於1小於3的
124 for i in ret: 125     print(i.id,i.name) 126 ret = db_session.query(Student).filter(~Student.id.in_([1,4])).all() # in_([1,3,4]) 只查詢id等於1,3,4的
127 for i in ret: 128     print(i.id,i.name) 129 ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查詢不等於1,3,4的
130 # 通配符
131 ret = db_session.query(Student).filter(Student.name.like('%e%')).all() 132 
133 
134 ret = db_session.query(Student).filter(~Student.name.like('Z%')).all() 135 for i in ret: 136     print(i.id,i.name) 137 
138 
139 
140 高級版更新操做 141 from my_create_table import User,engine 142 from sqlalchemy.orm import sessionmaker 143 
144 Session = sessionmaker(engine) 145 db_session = Session() 146 
147 #直接修改
148 db_session.query(Student).filter(Student.id > 3).update({Student.name: Student.name + "099"}, synchronize_session=False) 149 db_session.commit() 150 
151 db_session.query(Student).filter(Student.id > 3).update({"name": Student.name + "123"}, synchronize_session=False) 152 db_session.commit() 153 
154 
155 #在原有值基礎上添加 - 1
156 db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False) 157 
158 #在原有值基礎上添加 - 2
159 db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate") 160 db_session.commit()

View Code

3、Flask_SQLAlchemy

1. 安裝 Flask-SQLAlchemy模塊，pip安裝性能

　　pip install Flask-SQLAlchemy

2.先建立一個Flask項目

3.基於這個Flask項目,加入Flask-SQLAlchemy讓項目變得生動起來

 1 from flask import Flask  2 
 3 # 導入Flask-SQLAlchemy中的SQLAlchemy
 4 from flask_sqlalchemy import SQLAlchemy  5 
 6 # 實例化SQLAlchemy
 7 db = SQLAlchemy()  8 # PS : 實例化SQLAlchemy的代碼必需要在引入藍圖以前
 9 
10 from .views.users import user 11 
12 
13 def create_app(): 14     app = Flask(__name__) 15 
16     # 初始化App配置 這個app配置就厲害了,專門針對 SQLAlchemy 進行配置
17     # SQLALCHEMY_DATABASE_URI 配置 SQLAlchemy 的連接字符串兒
18     app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8"
19     # SQLALCHEMY_POOL_SIZE 配置 SQLAlchemy 的鏈接池大小
20     app.config["SQLALCHEMY_POOL_SIZE"] = 5
21     # SQLALCHEMY_POOL_TIMEOUT 配置 SQLAlchemy 的鏈接超時時間
22     app.config["SQLALCHEMY_POOL_TIMEOUT"] = 15
23     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False 24 
25     # 初始化SQLAlchemy , 本質就是將以上的配置讀取出來
26  db.init_app(app) 27 
28  app.register_blueprint(user) 29 
30     return app 31 
32 MyApp/__init__.py

4.創建models.py ORM模型文件

 1 from MyApp import db  2 
 3 Base = db.Model # 這句話你是否還記的?
 4 # from sqlalchemy.ext.declarative import declarative_base
 5 # Base = declarative_base()
 6 # 每一次咱們在建立數據表的時候都要作這樣一件事
 7 # 然而Flask-SQLAlchemy已經爲咱們把 Base 封裝好了
 8 
 9 # 創建User數據表
10 class Users(Base): # Base實際上就是 db.Model
11     __tablename__ = "users"
12     __table_args__ = {"useexisting": True} 13     # 在SQLAlchemy 中咱們是導入了Column和數據類型 Integer 在這裏
14     # 就和db.Model同樣,已經封裝好了
15     id = db.Column(db.Integer,primary_key=True) 16     username = db.Column(db.String(32)) 17     password = db.Column(db.String(32)) 18 
19 
20 if __name__ == '__main__': 21     from MyApp import create_app 22     app = create_app() 23     # 這裏你要回顧一下Flask應用上下文管理了
24     # 離線腳本:
25  with app.app_context(): 26  db.drop_all() 27  db.create_all() 28 
29 MyApp/models.py

5.實現一個簡單的用戶管理，登陸視圖

 1 from flask import Blueprint, request, render_template, redirect  2 from MyApp.models import User,db  3 
 4 users = Blueprint("user",__name__)  5 
 6 @users.route("/add_user",methods=["POST","GET"])  7 def add_user():  8     if request.method == "GET":  9         return render_template("add_user.html") 10     else: 11         username = request.form.get("username") 12         db.session.add(User(name=username)) 13  db.session.commit() 14         return redirect("/user_list") 15 
16 @users.route("/user_list",methods=["POST","GET"]) 17 def user_list(): 18     res = User.query.all() 19     print(res) 20     return render_template("user_list.html",user_list=res) 21 
22 
23 @users.route("/del/<uid>",methods=["POST","GET"]) 24 def delete_user(uid): 25     res = User.query.filter(User.id == int(uid)).delete() 26  db.session.commit() 27     print(res) 28     return redirect("/user_list") 29 
30 
31 
32 @user.route("/login",methods=["POST","GET"]) 33 def user_login(): 34     if request.method == "POST": 35         username = request.form.get("username") 36         password = request.form.get("password") 37 
38         # SQlalchemy須要這樣作
39         # from sqlalchemy.orm import sessionmaker
40         # Session = sessionmaker(engine)
41         # db_sesson = Session()
42         # 如今不用了,由於 Flask-SQLAlchemy 也已經爲咱們作好會話打開的工做
43         # 咱們在這裏作個弊:
44         db.session.add(Users(username=username,password=password)) 45  db.session.commit() 46 
47         # 而後再查詢
48         user_info = Users.query.filter(Users.username == username and User.password == password).first() 49         print(user_info.username) 50         if user_info: 51             return f"登陸成功{user_info.username}"
52 
53     return render_template("login.html")

其實Flask-SQLAlchemy比起SQLAlchemy更加的簡單自如,用法幾乎如出一轍,就是在配置和啓動上須要注意與Flask的配合就好啦