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