本文的主要内容：flask视图&路由、虚拟环境安装、路由各种定义、状态保持、cookie、session、模板基本使用、过滤器&自定义过滤器、模板代码复用：宏、继承/包含、模板中特有变量和函数、Flask-WTF 表单、CSRF、数据库操作、ORM、Flask-SQLAlchemy、增删改查操作、案例、蓝图、单元测试
全套笔记和代码自取地址： 请移步这里
感兴趣的小伙伴可以自取哦，欢迎大家点赞转发~
共 4 章，42 子模块
数据库
- Flask-SQLAlchemy 安装及连接
- 使用数据库
- 综合案例演练
- 数据库迁移
学习目标
- 能够理解 ORM 工作原理以及其优缺点
- 能够写出在 Flask 中连接 MySQL 的配置项格式（IP，端口，数据库）
- 能够使用 SQLAlchemy 定义出关系为一对多模型类
- 能够使用 SQLAlchemy 的相关函数创建表及删除表
- 能够写出的指定模型数据对数据库的增删改代码
- 能够写出模型数据按照条件查询的功能逻辑
- 能够写出模型数据按照指定数量分页的功能逻辑
- 能够写出模型数据按照指定条件排序的功能逻辑
- 能够按照课件步骤实现综合图书管理的相关案例
- 能够使用 Flask-Migrate 扩展对数据库进行迁移
数据库基本操作
-
在Flask-SQLAlchemy中，插入、修改、删除操作，均由数据库会话管理。
- 会话用 db.session 表示。在准备把数据写入数据库前，要先将数据添加到会话中然后调用 commit() 方法提交会话。
-
在 Flask-SQLAlchemy 中，查询操作是通过 query 对象操作数据。
- 最基本的查询是返回表中所有数据，可以通过过滤器进行更精确的数据库查询。
在视图函数中定义模型类
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
#设置连接数据库的URL
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:[email protected]:3306/test'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
#查询时会显示原始SQL语句
app.config['SQLALCHEMY_ECHO'] = True
db = SQLAlchemy(app)
class Role(db.Model):
# 定义表名
__tablename__ = 'roles'
# 定义列对象
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
us = db.relationship('User', backref='role')
#repr()方法显示一个可读字符串
def __repr__(self):
return 'Role:%s'% self.name
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True, index=True)
email = db.Column(db.String(64),unique=True)
password = db.Column(db.String(64))
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
def __repr__(self):
return 'User:%s'%self.name
if __name__ == '__main__':
app.run(debug=True)
模型之前的关联
一对多
class Role(db.Model):
...
#关键代码
us = db.relationship('User', backref='role', lazy='dynamic')
...
class User(db.Model):
...
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
-
其中realtionship描述了Role和User的关系。在此文中，第一个参数为对应参照的类"User"
-
第二个参数backref为类User申明新属性的方法
-
第三个参数lazy决定了什么时候SQLALchemy从数据库中加载数据
-
如果设置为子查询方式(subquery)，则会在加载完Role对象后，就立即加载与其关联的对象，这样会让总查询数量减少，但如果返回的条目数量很多，就会比较慢
- 设置为 subquery 的话，role.users 返回所有数据列表
-
另外,也可以设置为动态方式(dynamic)，这样关联对象会在被使用的时候再进行加载，并且在返回前进行过滤，如果返回的对象数很多，或者未来会变得很多，那最好采用这种方式
- 设置为 dynamic 的话，role.users 返回查询对象，并没有做真正的查询，可以利用查询对象做其他逻辑，比如：先排序再返回结果
-
多对多
registrations = db.Table('registrations',
db.Column('student_id', db.Integer, db.ForeignKey('students.id')),
db.Column('course_id', db.Integer, db.ForeignKey('courses.id'))
)
class Course(db.Model):
...
class Student(db.Model):
...
courses = db.relationship('Course',secondary=registrations,
backref='students',
lazy='dynamic')
常用的SQLAlchemy查询过滤器
| 过滤器 | 说明 |
|---|---|
| filter() | 把过滤器添加到原查询上，返回一个新查询 |
| filter_by() | 把等值过滤器添加到原查询上，返回一个新查询 |
| limit | 使用指定的值限定原查询返回的结果 |
| offset() | 偏移原查询返回的结果，返回一个新查询 |
| order_by() | 根据指定条件对原查询结果进行排序，返回一个新查询 |
| group_by() | 根据指定条件对原查询结果进行分组，返回一个新查询 |
常用的SQLAlchemy查询执行器
| 方法 | 说明 |
|---|---|
| all() | 以列表形式返回查询的所有结果 |
| first() | 返回查询的第一个结果，如果未查到，返回None |
| first_or_404() | 返回查询的第一个结果，如果未查到，返回404 |
| get() | 返回指定主键对应的行，如不存在，返回None |
| get_or_404() | 返回指定主键对应的行，如不存在，返回404 |
| count() | 返回查询结果的数量 |
| paginate() | 返回一个Paginate对象，它包含指定范围内的结果 |
创建表：
db.create_all()
删除表
db.drop_all()
插入一条数据
ro1 = Role(name='admin')
db.session.add(ro1)
db.session.commit()
#再次插入一条数据
ro2 = Role(name='user')
db.session.add(ro2)
db.session.commit()
一次插入多条数据
us1 = User(name='wang',email='[email protected]',password='123456',role_id=ro1.id)
us2 = User(name='zhang',email='[email protected]',password='201512',role_id=ro2.id)
us3 = User(name='chen',email='[email protected]',password='987654',role_id=ro2.id)
us4 = User(name='zhou',email='[email protected]',password='456789',role_id=ro1.id)
us5 = User(name='tang',email='[email protected]',password='158104',role_id=ro2.id)
us6 = User(name='wu',email='[email protected]',password='5623514',role_id=ro2.id)
us7 = User(name='qian',email='[email protected]',password='1543567',role_id=ro1.id)
us8 = User(name='liu',email='[email protected]',password='867322',role_id=ro1.id)
us9 = User(name='li',email='[email protected]',password='4526342',role_id=ro2.id)
us10 = User(name='sun',email='[email protected]',password='235523',role_id=ro2.id)
db.session.add_all([us1,us2,us3,us4,us5,us6,us7,us8,us9,us10])
db.session.commit()
"""
查询所有用户数据
查询有多少个用户
查询第1个用户
查询id为4的用户[3种方式]
查询名字结尾字符为g的所有数据[开始/包含]
查询名字不等于wang的所有数据[2种方式]
查询名字和邮箱都以 li 开头的所有数据[2种方式]
查询password是 `123456` 或者 `email` 以 `itheima.com` 结尾的所有数据
查询id为 [1, 3, 5, 7, 9] 的用户列表
查询name为liu的角色数据
查询所有用户数据，并以邮箱排序
每页3个，查询第2页的数据
"""
查询:filter_by精确查询
返回名字等于wang的所有人
User.query.filter_by(name='wang').all()
first()返回查询到的第一个对象
User.query.first()
all()返回查询到的所有对象
User.query.all()
filter模糊查询，返回名字结尾字符为g的所有数据。
User.query.filter(User.name.endswith('g')).all()
get():参数为主键，如果主键不存在没有返回内容
User.query.get()
逻辑非，返回名字不等于wang的所有数据
User.query.filter(User.name!='wang').all()
not_ 相当于取反
from sqlalchemy import not_
User.query.filter(not_(User.name=='chen')).all()
逻辑与，需要导入and，返回and()条件满足的所有数据
from sqlalchemy import and_
User.query.filter(and_(User.name!='wang',User.email.endswith('163.com'))).all()
逻辑或，需要导入or_
from sqlalchemy import or_
User.query.filter(or_(User.name!='wang',User.email.endswith('163.com'))).all()
查询数据后删除
user = User.query.first()
db.session.delete(user)
db.session.commit()
User.query.all()
更新数据
user = User.query.first()
user.name = 'dong'
db.session.commit()
User.query.first()
关联查询示例：
角色和用户的关系是一对多的关系，一个角色可以有多个用户，一个用户只能属于一个角色。
- 查询角色的所有用户
#查询roles表id为1的角色
ro1 = Role.query.get(1)
#查询该角色的所有用户
ro1.us.all()
- 查询用户所属角色
#查询users表id为3的用户
us1 = User.query.get(3)
#查询用户属于什么角色
us1.role
综合案例-图书管理
- WTF表单
- 数据库操作
- 一对多关系演练
定义模型
模型表示程序使用的数据实体，在Flask-SQLAlchemy中，模型一般是Python类，继承自db.Model，db是SQLAlchemy类的实例，代表程序使用的数据库。
类中的属性对应数据库表中的列。id为主键，是由Flask-SQLAlchemy管理。db.Column类构造函数的第一个参数是数据库列和模型属性类型。
注：如果没有在创建数据库的时候指定编码的话，向数据库中插入中文后，会报错，那么需要修改数据库的编码集:
alter database 数据库名 CHARACTER SET utf8
如下示例：定义了两个模型类，作者和书名。
#coding=utf-8
from flask import Flask,render_template,redirect,url_for
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
#设置连接数据
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:[email protected]:3306/test2'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
#实例化SQLAlchemy对象
db = SQLAlchemy(app)
#定义模型类-作者
class Author(db.Model):
__tablename__ = 'author'
id = db.Column(db.Integer,primary_key=True)
name = db.Column(db.String(32),unique=True
au_book = db.relationship('Book',backref='author')
def __repr__(self):
return 'Author:%s' %self.name
#定义模型类-书名
class Book(db.Model):
__tablename__ = 'books'
id = db.Column(db.Integer,primary_key=True)
name = db.Column(db.String(32))
au_book = db.Column(db.Integer,db.ForeignKey('author.id'))
def __str__(self):
return 'Book:%s,%s'%(self.info,self.lead)
创建表
if __name__ == '__main__':
db.drop_all()
db.create_all()
app.run(debug=True)
- 查看创建结果
show tables;
查看author表结构
desc author;
查看books表结构
desc books;
添加测试数据
#生成数据
au1 = Author(name='老王')
au2 = Author(name='老尹')
au3 = Author(name='老刘')
# 把数据提交给用户会话
db.session.add_all([au1, au2, au3])
# 提交会话
db.session.commit()
bk1 = Book(name='老王回忆录', author_id=au1.id)
bk2 = Book(name='我读书少，你别骗我', author_id=au1.id)
bk3 = Book(name='如何才能让自己更骚', author_id=au2.id)
bk4 = Book(name='怎样征服美丽少女', author_id=au3.id)
bk5 = Book(name='如何征服英俊少男', author_id=au3.id)
# 把数据提交给用户会话
db.session.add_all([bk1, bk2, bk3, bk4, bk5])
# 提交会话
db.session.commit()
生成数据后，查看数据：
数据显示&表单添加
数据显示
- 定义路由函数，并将 Author 和 Book 的所有结果传到模板
@app.route('/',methods=['GET','POST'])
def index():
author = Author.query.all()
book = Book.query.all()
return render_template('index.html',author=author,book=book)
- 模版关键代码
<ul>
{% for x in author %}
<li>{{ x }}</li>
{% endfor %}
</ul>
<hr>
<ul>
{% for x in book %}
<li>{{ x }}</li>
{% endfor %}
</ul>
- 效果图
表单添加
- 定义表单类
from flask_wtf import FlaskForm
from wtforms.validators import DataRequired
from wtforms import StringField,SubmitField
#创建表单类，用来添加信息
class Append(FlaskForm):
au_info = StringField(validators=[DataRequired()])
bk_info = StringField(validators=[DataRequired()])
submit = SubmitField(u'添加')
- 传入至模版中
#创建表单对象
@app.route('/',methods=['GET','POST'])
def index():
author = Author.query.all()
book = Book.query.all()
form = Append()
return render_template('index.html',author=author,book=book,form=form)
- 模板中代码
<form method="post">
{{ form.csrf_token }}
<p>作者：{{ form.au_info }}</p>
<p>书名：{{ form.bk_info }}</p>
<p>{{ form.submit }}</p>
</form>
- 效果图
表单验证
@app.route('/', methods=['get', 'post'])
def index():
append_form = Append()
if request.method == 'POST':
if append_form.validate_on_submit():
author_name = append_form.au_info.data
book_name = append_form.bk_info.data
# 判断数据是否存在
author = Author.query.filter_by(name=author_name).first()
if not author:
try:
# 先添加作者
author = Author(name=author_name)
db.session.add(author)
db.session.commit()
# 再添加书籍
book = Book(name=book_name, author_id=author.id)
db.session.add(book)
db.session.commit()
except Exception as e:
db.session.rollback()
print e
flash("数据添加错误")
else:
book_names = [book.name for book in author.books]
if book_name in book_names:
flash('该作者已存在相同的书名')
else:
try:
book = Book(name=book_name, author_id=author.id)
db.session.add(book)
db.session.commit()
except Exception as e:
db.session.rollback()
print e
flash('数据添加错误')
else:
flash('数据输入有问题')
authors = Author.query.all()
books = Book.query.all()
return render_template('test2.html', authors=authors, books=books, append_form=append_form)
- 重新编写html文件展示列表书籍
<h2>书籍列表</h2>
<ul>
{% for author in authors %}
<li>
{{ author.name }}
<ul>
{% for book in author.books %}
<li>{{ book.name }}
{% else %}
<li>无书籍</li>
{% endfor %}
</ul>
</li>
{% endfor %}
</ul>
- 在form标签下添加 flash 消息的显示
{% for message in get_flashed_messages() %}
{{ message }}
{% endfor %}
删除数据
- 定义删除author和book的路由
# 删除作者
@app.route('/delete_author/<int:author_id>')
def delete_author(author_id):
author = Author.query.get(author_id)
if not author:
flash('数据不存在')
else:
try:
Book.query.filter_by(author_id=author_id).delete()
db.session.delete(author)
db.session.commit()
except Exception as e:
db.session.rollback()
print e
flash('操作数据库失败')
return redirect(url_for('index'))
# 删除书籍
@app.route('/delete_book/<int:book_id>')
def delete_book(book_id):
book = Book.query.get(book_id)
if not book:
flash('数据不存在')
else:
try:
db.session.delete(book)
db.session.commit()
except Exception as e:
db.session.rollback()
print e
flash('操作数据库失败')
return redirect(url_for('index'))
- 在模版中添加删除的 a 标签链接
<h2>书籍列表</h2>
<ul>
{% for author in authors %}
<li>
{{ author.name }} <a rel="nofollow" href="/delete_author/{{ author.id }}">删除</a>
<ul>
{% for book in author.books %}
<li>{{ book.name }} <a rel="nofollow" href="/delete_book/{{ book.id }}">删除</a></li>
{% else %}
<li>无书籍</li>
{% endfor %}
</ul>
</li>
{% endfor %}
</ul>