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()
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的配合就好啦