Python-Flask web框架

一、环境部署

# mkvirtualenv flask_py2
# pip install flask==0.10.1
# pip freeze > requirements.txt
# pip install -r requirements.txt

二、HelloWorld

# coding:utf-8

from flask import Flask

# 创建flask的应用对象
app = Flask(__name__)  # __name__表示当前的模块名字

@app.route('/index')
def index():
    """定义视图函数"""
    return "Hello World"


if __name__ == '__main__':
    # 启动server
    app.run()

三、静态文件目录

# 创建flask的应用对象
app = Flask(__name__,  # __name__表示当前的模块名字
            static_url_path="/python",  # 访问静态资源的url前缀, http://127.0.0.1:5000/python/index.html
            static_folder="static",  # 静态文件目录，默认是static
            template_folder="templates",  # 模板文件目录，默认templates
            )

四、配置参数的设置

设置配置参数的3种方式

app.config.from_pyfile("config.cfg")  # 使用配置文件方式一

class Config(object):
    DEBUG = True
app.config.from_object(Config)  # 第二种配置方式

app.config["DEBUG"] = True  # 第三种配置方式

读取配置参数

print(app.config.get("DEBUG"))  # 读取配置参数方式一

from flask import current_app
print(current_app.config.get("DEBUG"))  # 读取配置参数方式二

五、run的使用

if __name__ == '__main__':
    # 启动server
    app.run(host="0.0.0.0", port=5000, debug=True)

六、路由

from flask import Flask
from flask import redirect
from flask import url_for


app = Flask(__name__)


@app.route("/")
def index():
    return "Hello, World!"


# 通过methods限制访问方式
@app.route("/post_only", methods=["POST"])
def post_only():
    return "post only page"


# 同一个视图函数2个URL
@app.route("/hi")
@app.route("/hi2")
def hi():
    return "hi page"


@app.route("/login")
def login():
    # url = "/"
    url = url_for("index")  # 获取index视图函数的url
    return redirect(url)  # 重定向


if __name__ == '__main__':
    print(app.url_map)  # 查看整个flask中的路由信息
    app.run(debug=True)

七、转换器

from flask import Flask
from werkzeug.routing import BaseConverter
from flask import redirect
from flask import url_for

app = Flask(__name__)


@app.route("/goods/<int:goods_id>")  # 转换器，接收参数；int如果不加则是匹配字符串
def goods_detail(goods_id):
    return "goods detail page, goods id: %d" % goods_id


# 自定义转换器
class MobileConverter(BaseConverter):
    """构建一个匹配手机号的普通转换器"""

    def __init__(self, url_map):
        super(MobileConverter, self).__init__(url_map)
        self.regex = r'1[34578]d{9}'


class RegexConverter(BaseConverter):
    """构建一个万能转换器"""

    def __init__(self, url_map, regex):
        # 调用父类的初始化方法
        super(RegexConverter, self).__init__(url_map)
        # 将正则表达式的参数保存到对象中，flask用这个属性实现路由的匹配
        self.regex = regex

    def to_python(self, value):
        print("to_python: ", value)
        return value

    def to_url(self, value):
        return value


# 将自定义转换器添加到flask的应用中
app.url_map.converters["mobile"] = MobileConverter
app.url_map.converters["re"] = RegexConverter


# 使用自定义转换器
@app.route("/send/<mobile:mobile>")
@app.route("/send2/<re(r'1[34578]d{9}'):mobile>")
def send_sms(mobile):
    return "send sms to %s" % mobile


@app.route("/index")
def index():
    url = url_for("send_sms", mobile="13312322156")
    return redirect(url)


if __name__ == '__main__':
    app.run()

八、接收request请求参数

request对象保存了一次http请求的所有信息

data：请求的数据，字符串
form：请求的表单数据，multiDict
args：请求的查询参数，multiDict
cookies：请求的cookie信息，Dict
headers：请求的报文头，EnvironHeaders
method：请求使用的HTTP方法，GET/POST等
url：请求的URL地址，string
files：请求上传的文件

from flask import Flask
from flask import request

app = Flask(__name__)


@app.route("/index", methods=["GET", "POST"])
def index():
    # request包含前端发送过来的所有请求信息
    # 提取表单格式的数据
    name = request.form.get("name", "no_name")  # no_name是默认值
    age = request.form.get("age", "0")

    # 取出多个重名的参数
    name_list = request.form.getlist("name")
    print(name_list)

    # 接收查询字符串参数
    city = request.args.get("city")
    print(city)

    # 获取请求体数据
    print(request.data)
    return "hello name=%s, age=%s" % (name, age)


@app.route("/register")
def register():
    # 获取method
    if request.method == "GET":
        pass
    elif request.method == "POST":
        pass


if __name__ == '__main__':
    app.run()

九、上传文件

from flask import Flask
from flask import request

app = Flask(__name__)


@app.route("/upload", methods=["POST"])
def upload():
    """接收前端上传的文件"""
    file_obj = request.files.get("pic")
    if file_obj is None:
        # 表示没有发送文件
        return "未上传文件"
    # 保存文件
    # with open("files/demo.jpg", "wb") as f:
    #     data = file_obj.read()
    #     f.write(data)
    file_obj.save("files/demo2.jpg")
    return "上传成功"


if __name__ == '__main__':
    app.run()

十、abort 提前终止视图函数

from flask import Flask
from flask import request
from flask import abort
from flask import Response

app = Flask(__name__)


@app.route("/login", methods=["POST"])
def login():
    name = request.form.get("name")
    pwd = request.form.get("pwd")
    if name != "dongfei" or pwd != "dpwd":
        # 使用abort函数立即终止视图函数执行，并且返回前端特定信息
        abort(403)  # 传递状态码信息

        # 返回信息
        # resp = Response("login failed")
        # abort(resp)

    return "login success"


@app.errorhandler(404)
def handle_404_error(err):
    """自定义处理错误方法"""
    # 这个返回值是前端用户看到的结果
    return "页面找不到了，错误：%s" % err


if __name__ == '__main__':
    app.run()

十一、response构造

from flask import Flask
from flask import make_response

app = Flask(__name__)


@app.route("/index")
def index():
    # 使用元组返回自定义的响应信息
    # 响应体, 状态码, 响应头
    return "index page", 200, [("frame", "flask"), ]


@app.route("/index2")
def index2():
    # 使用字典返回自定义的响应信息
    # 响应体, 自定义状态码, 响应头
    return "index page", "600 my status", {"frame": "flask"}


@app.route("/index3")
def index3():
    # 使用make_response构造响应信息
    resp = make_response("index3 page")
    resp.status = "666 my index3 status"
    resp.headers["frame"] = "flask"
    return resp


if __name__ == '__main__':
    app.run()

十二、返回json响应

from flask import Flask
from flask import jsonify

app = Flask(__name__)


@app.route("/index")
def index():
    # json是字符串
    data = {
        "name": "dongfei",
        "age": 18
    }
    # return json.dumps(data), 200, {"Content-Type": "application/json"}
    # jsonify帮助转换json后并且设置响应头
    return jsonify(data)


if __name__ == '__main__':
    app.run()

from flask import Flask
from flask import make_response
from flask import request


app = Flask(__name__)


@app.route("/set_cookie")
def set_cookie():
    resp = make_response("success")
    # 设置cookie，默认有效期是临时cookie，浏览器关闭则失效
    resp.set_cookie("key_cookie", "value_cookie")

    # 设置cookie有效期，单位s
    resp.set_cookie("key2_cookie", "value2_cookie", max_age=3600)
    return resp


@app.route("/get_cookie")
def get_cookie():
    # 获取cookie
    cookie = request.cookies.get("key_cookie")
    return cookie


@app.route("/del_cookie")
def del_cookie():
    resp = make_response("del success")
    # 删除cookie
    resp.delete_cookie("key_cookie")
    return resp


if __name__ == '__main__':
    app.run()

十四、session

from flask import Flask
from flask import session

app = Flask(__name__)

# flask的session需要的秘钥
app.config["SECRET_KEY"] = "ALKSJDFLWJERJLKJSLDJF21324"


# flask默认把session保存到cookie中


@app.route("/login")
def login():
    # 设置session
    session["name"] = "dongfei"
    session["mobile"] = 111111
    return "login success"


@app.route("/index")
def index():
    # 获取session
    name = session.get("name")
    return "hello %s" % name


if __name__ == '__main__':
    app.run()

十五、请求钩子 hook

befor_first_request:  在处理第一个请求前运行
@app.befor_first_request

before_request:  在每次请求前运行

after_request(response):  如果没有未处理的异常抛出，在每次请求后运行

teardown_request(response):  在每次请求后运行，即使有未处理的异常抛出

from flask import Flask

app = Flask(__name__)


@app.route("/index")
def index():
    print("in the index")
    a = 1/0
    return "in the index"


@app.before_first_request
def handle_before_first_request():
    """在第一次请求处理之前先被执行"""
    print("in the handle_before_first_request")


@app.before_request
def handle_before_request():
    """在每次请求前运行"""
    print("in the handle_before_request")


@app.after_request
def handle_after_request(response):
    """视图函数没有出现异常的情况下，在每次请求之后被执行"""
    print("in the handle_after_request")
    return response


@app.teardown_request
def handle_teardown_request(response):
    """无论视图函数有没有出现异常的情况下，在每次请求之后被执行"""
    print("in the handle_teardown_request")
    return response


if __name__ == '__main__':
    app.run()

十六、Flask-Script

>pip install Flask-Script

from flask import Flask
# 启动命令的管理类
from flask_script import Manager

app = Flask(__name__)

# 让manager管理app
manager = Manager(app)


@app.route("/index")
def index():
    return "index page"


if __name__ == '__main__':
    # 通过管理对象启动flask
    manager.run()

>python 12_flask_script.py runserver -h 127.0.0.1 -p 5000

十七、模板template

1、模板变量

from flask import Flask
from flask import render_template

app = Flask(__name__)


@app.route("/index")
def index():
    data = {
        "name": "dongfei",
        "age": 20,
        "my_dict": {"city": "biejing"},
        "my_list": [1, 2, 3]
    }
    return render_template("index.html", **data)


if __name__ == '__main__':
    app.run()

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <p>name = {{ name }}</p>
    <p>age = {{ age }}</p>
    <p>my_dict: city = {{ my_dict.city }}</p>
    <p>my_list: {{ my_list }}</p>
</body>
</html>

2、过滤器

字符串过滤器

safe：禁用转义；
  <p>{{ '<em>hello</em>' | safe }}</p>

capitalize：把变量值的首字母转成大写，其余字母转小写；
  <p>{{ 'hello' | capitalize }}</p>

lower：把值转成小写；
  <p>{{ 'HELLO' | lower }}</p>

upper：把值转成大写；
  <p>{{ 'hello' | upper }}</p>

title：把值中的每个单词的首字母都转成大写；
  <p>{{ 'hello' | title }}</p>

trim：把值的首尾空格去掉；
  <p>{{ ' hello world ' | trim }}</p>

reverse:字符串反转；
  <p>{{ 'olleh' | reverse }}</p>

format:格式化输出；
  <p>{{ '%s is %d' | format('name',17) }}</p>

striptags：渲染之前把值中所有的HTML标签都删掉；
  <p>{{ '<em>hello</em>' | striptags }}</p>

列表过滤器

first：取第一个元素
  <p>{{ [1,2,3,4,5,6] | first }}</p>

last：取最后一个元素
  <p>{{ [1,2,3,4,5,6] | last }}</p>

length：获取列表长度
  <p>{{ [1,2,3,4,5,6] | length }}</p>

sum：列表求和
  <p>{{ [1,2,3,4,5,6] | sum }}</p>

sort：列表排序
  <p>{{ [6,2,3,1,5,4] | sort }}</p>

自定义过滤器

from flask import Flask
from flask import render_template

app = Flask(__name__)


@app.route("/index")
def index():
    data = {
        "name": "dongfei",
        "age": 20,
        "my_dict": {"city": "biejing"},
        "my_list": [1, 2, 3, 4, 5, 6]
    }
    return render_template("index2.html", **data)


# 自定义过滤器
def list_step_2(li):
    """处理列表，步长为2的取值"""
    return li[::2]


# 注册过滤器
app.add_template_filter(list_step_2, "li2")  # li2是过滤器名字


# 自定义过滤器方法二
@app.template_filter("li3")
def list_step_3(li):
    """处理列表，步长为2的取值"""
    return li[::3]


if __name__ == '__main__':
    app.run()

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    {#去首尾空白字符#}
    <p>{{ "  flask world  " | trim}}</p>

    <p>{{ my_list | li2}}</p>
    <p>{{ my_list | li3}}</p>
</body>
</html>

3、表单Flask-WTF扩展

>pip install Flask-WTF

WTForms支持的HTML标准字段

WTForms常用验证函数

from flask import Flask
from flask import render_template
from flask import redirect
from flask import url_for
from flask import session
from flask_wtf import FlaskForm
from wtforms import StringField  # WTForms支持的HTML标准字段
from wtforms import PasswordField
from wtforms import SubmitField
from wtforms.validators import DataRequired  # WTForms验证函数
from wtforms.validators import EqualTo

app = Flask(__name__)

app.config["SECRET_KEY"] = "lkajsdklrjsdfwer"


# 定义表单的模型类
class RegisterForm(FlaskForm):
    """自定义的注册表单模型类"""
    username = StringField(label="用户名", validators=[DataRequired("用户名不能为空"), ], )  # DataRequired不能为空
    password = PasswordField(label="密码", validators=[DataRequired("密码不能为空")])
    password2 = PasswordField(label="确认密码",
                              validators=[DataRequired("确认密码不能为空"),
                                          EqualTo("password", "两次输入的密码不一致")])  # EqualTo和password字段比较是否一致
    submit = SubmitField(label="提交")


@app.route("/register", methods=["GET", "POST"])
def register():
    # 创建表单对象，如果是post请求，前端发送数据，flask会把数据构造form对象的时候存放在对象中
    form = RegisterForm()

    # 判断form中的数据是否合理，满足所有返回True
    if form.validate_on_submit():
        # 表示验证合格，提取数据
        username = form.username.data
        password = form.password.data
        password2 = form.password2.data
        print(username, password, password2)
        session["username"] = username
        return redirect(url_for("index"))

    return render_template("register.html", form=form)


@app.route("/index")
def index():
    username = session.get("username")
    return "hello %s" % username


if __name__ == '__main__':
    app.run()

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

<form action="" method="post" novalidate>
    {{ form.csrf_token }}

    {{ form.username.label }}
    <p>{{ form.username }}</p>
    {% for msg in form.username.errors %}
        <p>{{ msg }}</p>
    {% endfor %}

    {{ form.password.label }}
    <p>{{ form.password }}</p>
    {% for msg in form.password.errors %}
        <p>{{ msg }}</p>
    {% endfor %}

    {{ form.password2.label }}
    <p>{{ form.password2 }}</p>
    {% for msg in form.password2.errors %}
        <p>{{ msg }}</p>
    {% endfor %}

    {{ form.submit }}
</form>

</body>
</html>

4、模板宏

macro.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

{#定义没有参数的宏#}
{% macro input() %}
    <input type="text" value="" size="30">
{% endmacro %}
{#调用宏#}
<h1>{{ input() }}</h1>

{#定义有参数的宏#}
{% macro input2(type, value, size) %}
    <input type="{{ type }}" value="{{ value }}" size="{{ size }}">
{% endmacro %}
<h1>{{ input2("password", "", 50) }}</h1>

{#定义有参数的宏，有默认值#}
{% macro input3(type="text", value="default", size=70) %}
    <input type="{{ type }}" value="{{ value }}" size="{{ size }}">
{% endmacro %}
<h1>{{ input3() }}</h1>

{#导入宏#}
{% import "macro_input.html" as m_input %}
{{ m_input.input4() }}

</body>
</html>

macro_input.html

{#在外部定义宏#}
{% macro input4(type="text", value="default", size=70) %}
    <input type="{{ type }}" value="{{ value }}" size="{{ size }}">
{% endmacro %}
<h1>{{ input4() }}</h1>

5、模板闪现信息

from flask import Flask
from flask import flash
from flask import render_template

app = Flask(__name__)
app.config["SECRET_KEY"] = "sdfasdw"


@app.route("/index")
def index():
    # 添加闪现信息
    flash("message01")
    flash("message02")
    flash("message03")
    return render_template("flash.html")


if __name__ == '__main__':
    app.run()

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

<h1>闪现信息</h1>
{% for msg in get_flashed_messages() %}
    <p>{{ msg }}</p>
{% endfor %}

</body>
</html>

十八、数据库扩展sqlalchemy

SQLAlchemy是一个关系型数据库框架，它提供了高层的ORM和底层的原生数据库的操作
flask-sqlalchemy是一个简化了SQLAlchemy操作的flask扩展

>pip install flask-sqlalchemy
>pip install flask-mysqldb

sqlalchemy配置

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)


# 连接数据的URI
# app.config["SQLALCHEMY_DATABASE_URI"] = "mysql://flask:flask@192.168.100.1:3306/flask"

# 每次请求接收后自动提交数据到数据库，不建议打开
# app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True

# 跟踪数据库的修改而修改模型类对象，通常打开
# app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True

# 显示原生SQL语句
# app.config["SQLALCHEMY_ECHO"] = True

# 使用类定义配置
class Config(object):
    """配置参数"""
    SQLALCHEMY_DATABASE_URI = "mysql://flask:flask@192.168.100.1:3306/flask"
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    SQLALCHEMY_ECHO = True


# 注册配置
app.config.from_object(Config)

# 创建数据库sqlalchemy工具对象
db = SQLAlchemy(app)

1、模型类、创建数据表、插入数据

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)


# 使用类定义配置
class Config(object):
    """配置参数"""
    SQLALCHEMY_DATABASE_URI = "mysql://flask:flask@192.168.100.1:3306/flask"
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    # SQLALCHEMY_ECHO = True


# 注册配置
app.config.from_object(Config)

# 创建数据库sqlalchemy工具对象
db = SQLAlchemy(app)


# 创建模型类
class Role(db.Model):
    """角色表"""
    __tablename__ = "tbl_roles"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), unique=True)
    users = db.relationship("User", backref="role")  # User: 对应模型类关系, backref="role": 可以通过User类反推

    def __repr__(self):
        """可以让显示对象的时候更直观"""
        return "Role object: name=%s" % self.name


class User(db.Model):
    """用户表"""
    __tablename__ = "tbl_users"  # 指定数据库的表名

    id = db.Column(db.Integer, primary_key=True)  # 整型主键会默认自增
    name = db.Column(db.String(64), unique=True)
    email = db.Column(db.String(128), unique=True)
    password = db.Column(db.String(128), nullable=False)
    role_id = db.Column(db.Integer, db.ForeignKey("tbl_roles.id"))  # 定义外键, tbl_roles: 表名

    def __repr__(self):
        return "User object: name=%s" % self.name


if __name__ == '__main__':
    # 通过db方式创建数据库表
    # 清楚数据库里的所有表
    db.drop_all()

    # 创建所有的表
    db.create_all()

2、插入数据

    # 插入数据
    role1 = Role(name="admin")
    db.session.add(role1)
    db.session.commit()

    role2 = Role(name="stuff")
    db.session.add(role2)
    db.session.commit()

    us1 = User(name="dongfei", email="dongfei@126.com", password="dongfei", role_id=role1.id)
    us2 = User(name="dongfei2", email="dongfei2@126.com", password="dongfei2", role_id=role2.id)
    db.session.add_all([us1, us2])  # 一次保存多条数据
    db.session.commit()

3、数据库查询

    r_list = Role.query.all()  # 查询所有数据
    print(r_list[0].name)

    r_first = Role.query.first()  # 查询第一条数据
    print(r_first.name)

    r_get = Role.query.get(2)  # 根据主键id查询
    print(r_get.name)

    # 过滤
    user = User.query.filter_by(name="dongfei").first()
    print(user.email)

    user = User.query.filter_by(name="dongfei", role_id=2).first()  # 并且关系
    print(type(user))

    user = User.query.filter(User.name == "dongfei").first()
    print(user.name)

    from sqlalchemy import or_

    user_list = User.query.filter(or_(User.name == "dongfei", User.email.endswith("126.com"))).all()  # 或者关系
    print(user_list[1].name)

    # User.query.filter().offset().limit().order_by().all()

    # 跳过1条，查询1条
    user = User.query.offset(1).limit(1).first()
    print(user.name)

    # 倒叙排序
    user = User.query.order_by(User.id.desc()).all()
    print(user)

    # 分组
    from sqlalchemy import func

    # SELECT role_id, COUNT(role_id) from User group by role_id;
    group = db.session.query(User.role_id, func.count(User.role_id)).group_by(User.role_id).all()
    print(group[0])

    # 关联查询
    # 从Role往User查询
    ro = Role.query.get(1)
    print(ro.users[0].name)

    # 从User往Role查询
    user = User.query.get(1)
    print(user.role.name)

4、更新与删除

    # 修改方式一
    user = User.query.get(1)
    user.name = "jack"
    db.session.add(user)
    db.session.commit()

    # 修改方式二
    User.query.filter_by(name="dongfei2").update({"name": "maria", "password": "123456"})
    db.session.commit()

    # 删除
    user = User.query.get(1)
    db.session.delete(user)
    db.session.commit()

十九、管理作者和图书的小案例

后端

from flask import Flask
from flask import render_template
from flask_sqlalchemy import SQLAlchemy
from flask_wtf import FlaskForm
from wtforms import StringField
from wtforms.validators import DataRequired
from wtforms import SubmitField
from flask import request
from flask import redirect
from flask import url_for
from flask import jsonify

app = Flask(__name__)


class Config(object):
    SQLALCHEMY_DATABASE_URI = "mysql://flask:flask@192.168.100.1:3306/author_book"
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    SECRET_KEY = "LKSJLKFJSDIUETR"


app.config.from_object(Config)
db = SQLAlchemy(app)


# 创建数据表模型
class Author(db.Model):
    """作者表"""
    __tablename__ = "tbl_authors"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), unique=True)
    books = db.relationship("Book", backref="author")


class Book(db.Model):
    """书籍表"""
    __tablename__ = "tbl_books"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    author_id = db.Column(db.Integer, db.ForeignKey("tbl_authors.id"))


# 创建表单模型类
class AuthorBookForm(FlaskForm):
    """作者数据表单模型类"""
    author_name = StringField(label="作者", validators=[DataRequired("必填")])
    book_name = StringField(label="书籍", validators=[DataRequired("必填")])
    submit = SubmitField(label="保存")


@app.route("/", methods=["GET", "POST"])
def index():
    # 创建表单对象
    form = AuthorBookForm()

    if form.validate_on_submit():
        """表单验证成功"""
        # 提取数据
        author_name = form.author_name.data
        book_name = form.book_name.data

        # 保存数据
        author = Author(name=author_name)
        db.session.add(author)
        db.session.commit()

        book = Book(name=book_name, author_id=author.id)
        # book = Book(name=book_name, author=author)
        db.session.add(book)
        db.session.commit()

    # 查询数据
    authors_list = Author.query.all()
    return render_template("index.html", authors_list=authors_list, form=form)


@app.route("/delete_book", methods=["POST"])
def delete_book():
    """删除数据"""
    # 提取json参数，如果前端发送请求体格式json，get_json()自动解析json成字典
    req_dict = request.get_json()
    book_id = req_dict.get("book_id")

    # 删除数据
    book = Book.query.get(book_id)
    db.session.delete(book)
    db.session.commit()

    return jsonify(code=0, message="OK")


if __name__ == '__main__':
    app.run()
    # db.drop_all()
    # db.create_all()
    #
    # au_jin = Author(name="金庸")
    # au_tian = Author(name="天蚕土豆")
    # au_luo = Author(name="罗贯中")
    # db.session.add_all([au_jin, au_tian, au_luo])
    # db.session.commit()
    #
    # bk_tian = Book(name="天龙八部", author_id=au_jin.id)
    # bk_she = Book(name="射雕三部曲", author_id=au_jin.id)
    # bk_dou = Book(name="斗破苍穹", author_id=au_tian.id)
    # bk_san = Book(name="三国演义", author_id=au_luo.id)
    # db.session.add_all([bk_tian, bk_she, bk_dou, bk_san])
    # db.session.commit()

前端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<form action="" method="post" novalidate>
    {{ form.csrf_token }}

    {{ form.author_name.label }}
    <p>{{ form.author_name }}</p>
    {% for msg in form.author_name.errors %}
        <p>{{ msg }}</p>
    {% endfor %}

    {{ form.book_name.label }}
    <p>{{ form.book_name }}</p>
    {% for msg in form.book_name.errors %}
        <p>{{ msg }}</p>
    {% endfor %}

    {{ form.submit }}
</form>

<br/>
<ul>
    {% for author in authors_list %}
        <li>作者：{{ author.name }}</li>
        <ul>
            {% for book in author.books %}
                <li>书籍：{{ book.name }}</li>
                <a href="javascript:;" book-id={{ book.id }}>删除</a>
            {% endfor %}
        </ul>
    {% endfor %}
</ul>


<script src="/static/js/jquery-3.5.1.js" type="text/javascript"></script>

<script>
    $("a").click(
        function () {
            var data = {
                book_id: $(this).attr("book-id")
            };
            var req_json = JSON.stringify(data);
            $.ajax({
                url: "/delete_book",
                type: "post",
                data: req_json,
                contentType: "application/json",
                dataType: "json",
                success: function (response) {
                    if (response.code == 0) {
                        alert("删除完成");
                        location.href = "/";
                    }
                }
            })
        }
    )
</script>

</body>
</html>

二十、数据库maigrate扩展

数据库模块类修改后可以通过maigrate同步到数据库

依赖flask-script
>pip install flask-migrate

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand

app = Flask(__name__)


# 使用类定义配置
class Config(object):
    """配置参数"""
    SQLALCHEMY_DATABASE_URI = "mysql://flask:flask@192.168.100.1:3306/flask"
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    # SQLALCHEMY_ECHO = True


# 注册配置
app.config.from_object(Config)

# 创建数据库sqlalchemy工具对象
db = SQLAlchemy(app)

# 创建flask脚本管理工具对象
manager = Manager(app)

# 创建数据库迁移工具对象
Migrate(app, db)

# 向manager对象中添加数据库的操作命令
manager.add_command("db", MigrateCommand)


# 创建模型类
class Role(db.Model):
    """角色表"""
    __tablename__ = "tbl_roles"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), unique=True)
    users = db.relationship("User", backref="role")  # User: 对应模型类关系, backref="role": 可以通过User类反推

    def __repr__(self):
        """可以让显示对象的时候更直观"""
        return "Role object: name=%s" % self.name


class User(db.Model):
    """用户表"""
    __tablename__ = "tbl_users"  # 指定数据库的表名

    id = db.Column(db.Integer, primary_key=True)  # 整型主键会默认自增
    name = db.Column(db.String(64), unique=True)
    age = db.Column(db.Integer, unique=True)
    email = db.Column(db.String(128), unique=True)
    password = db.Column(db.String(128), nullable=False)
    role_id = db.Column(db.Integer, db.ForeignKey("tbl_roles.id"))  # 定义外键, tbl_roles: 表名

    def __repr__(self):
        return "User object: name=%s" % self.name


if __name__ == '__main__':
    # 通过manager对象启动程序
    manager.run()

>python 18_数据库maigrate扩展.py db init  # 初始化
>python 18_数据库maigrate扩展.py db migrate -m 'add age column'  # 生成迁移文件
>python 18_数据库maigrate扩展.py db upgrade  # 同步数据库
>python 18_数据库maigrate扩展.py db history  # 查询历史版本
>python 18_数据库maigrate扩展.py db downgrade 3a5dde35dad1  # 回滚

二十一、蓝图

1、蓝图的基本使用

main.py

from flask import Flask
from orders import app_orders

app = Flask(__name__)

# 注册蓝图
# app.register_blueprint(app_orders)

# 添加前缀 http://127.0.0.1:5000/orders/get_orders
app.register_blueprint(app_orders, url_prefix="/orders")

if __name__ == '__main__':
    print(app.url_map)
    app.run()

orders.py

from flask import Blueprint

# 创建一个蓝图的对象
app_orders = Blueprint("app_orders", __name__)


@app_orders.route("/get_orders")
def get_orders():
    return "orders page"

2、以目录形式定义蓝图

main.py

from flask import Flask
from orders import app_orders
from cart import app_cart

app = Flask(__name__)

# 注册蓝图
# app.register_blueprint(app_orders)

# 添加前缀 http://127.0.0.1:5000/orders/get_orders
app.register_blueprint(app_orders, url_prefix="/orders")
app.register_blueprint(app_cart, url_prefix="/cart")


if __name__ == '__main__':
    print(app.url_map)
    app.run()

cart/__init.py

from flask import Blueprint


app_cart = Blueprint("app_cart", __name__, template_folder="templates")

# 在__init__被执行的时候，把视图加载进来，让蓝图发现views存在
from .views import get_cart

cart/views.py

from . import app_cart
from flask import render_template

@app_cart.route("get_cart")
def get_cart():
    return render_template("cart.html")

二十二、单元测试

1、页面的单元测试

from flask import Flask
from flask import request
from flask import jsonify

app = Flask(__name__)


@app.route("/login", methods=["POST"])
def login():
    # 接收参数
    username = request.form.get("username")
    password = request.form.get("password")

    # 参数判断
    if not all([username, password]):  # 列表中的变量都为真，all返回真
        resp = {
            "code": 1,
            "message": "invalid params"
        }
        return jsonify(resp)

    if username == "admin" and password == "admin":
        resp = {
            "code": 0,
            "message": "login success"
        }
        return jsonify(resp)
    else:
        resp = {
            "code": 2,
            "message": "wrong useername or password"
        }
        return jsonify(resp)


if __name__ == '__main__':
    app.run()

test.py

import unittest
from login import app
import json


class LoginTest(unittest.TestCase):
    """构造单元测试案例"""

    def setUp(self):
        """在进行测试之前先被执行"""
        # 设置flask工作在测试模式
        # app.config["TESTING"] = True
        app.testing = True

        # 创建web请求的客户端
        self.client = app.test_client()

    def test_empty_username_password(self):
        """测试用户名密码不完整情况"""
        # 测试用户名密码都不传
        # 利用client模拟发送web请求
        ret = self.client.post("/login", data={})

        # ret是视图返回的响应对象，ret.data是响应数据
        resp = ret.data
        resp = json.loads(resp)

        # 开始断言测试
        self.assertIn("code", resp)  # "code"是否在resp中
        self.assertEqual(resp["code"], 1)  # resp["code"]是否等于1

        # 测试只传用户名
        # 利用client模拟发送web请求
        ret = self.client.post("/login", data={"username": "admin"})

        # ret是视图返回的响应对象，ret.data是响应数据
        resp = ret.data
        resp = json.loads(resp)

        # 开始断言测试
        self.assertIn("code", resp)  # "code"是否在resp中
        self.assertEqual(resp["code"], 1)  # resp["code"]是否等于1

        # 测试只传密码
        # 利用client模拟发送web请求
        ret = self.client.post("/login", data={"password": "admin"})

        # ret是视图返回的响应对象，ret.data是响应数据
        resp = ret.data
        resp = json.loads(resp)

        # 开始断言测试
        self.assertIn("code", resp)  # "code"是否在resp中
        self.assertEqual(resp["code"], 1)  # resp["code"]是否等于1

    def test_wrong_username_password(self):
        """测试用户名或密码错误"""
        # 密码错误的情况
        ret = self.client.post("/login", data={"username": "admin", "password": "123456"})
        resp = ret.data
        resp = json.loads(resp)
        print(resp["code"])

        # 断言
        self.assertIn("code", resp)
        self.assertEqual(resp["code"], 2)


if __name__ == '__main__':
    unittest.main()  # 执行所有测试案例

2、数据库测试

db_demo.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)


# 使用类定义配置
class Config(object):
    """配置参数"""
    SQLALCHEMY_DATABASE_URI = "mysql://flask:flask@192.168.100.1:3306/flask"
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    # SQLALCHEMY_ECHO = True


# 注册配置
app.config.from_object(Config)

# 创建数据库sqlalchemy工具对象
db = SQLAlchemy(app)


# 创建模型类
class Role(db.Model):
    """角色表"""
    __tablename__ = "tbl_roles"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), unique=True)
    users = db.relationship("User", backref="role")  # User: 对应模型类关系, backref="role": 可以通过User类反推

    def __repr__(self):
        """可以让显示对象的时候更直观"""
        return "Role object: name=%s" % self.name


class User(db.Model):
    """用户表"""
    __tablename__ = "tbl_users"  # 指定数据库的表名

    id = db.Column(db.Integer, primary_key=True)  # 整型主键会默认自增
    name = db.Column(db.String(64), unique=True)
    email = db.Column(db.String(128), unique=True)
    password = db.Column(db.String(128), nullable=False)
    role_id = db.Column(db.Integer, db.ForeignKey("tbl_roles.id"))  # 定义外键, tbl_roles: 表名

    def __repr__(self):
        return "User object: name=%s" % self.name

test_db_demo.py

import unittest
from db_demo import User, db, app


class DatabaseTest(unittest.TestCase):
    def setUp(self):
        app.testing = True
        # 使用测试的数据库测试，不影响生产的数据库
        app.config["SQLALCHEMY_DATABASE_URI"] = "mysql://flask:flask@192.168.100.1:3306/flask_test"
        db.create_all()

    def test_add_user(self):
        """添加用户的数据库测试"""
        # 添加数据
        user = User(name="root", email="root@localhost", password="123456")
        db.session.add(user)
        db.session.commit()

        import time
        time.sleep(20)

        # 查询数据是否存在
        result_user = User.query.filter_by(name="root").first()
        self.assertIsNotNone(result_user)

    def tearDown(self):
        """所有的测试执行后执行，通常用来执行清理操作"""
        db.session.remove()
        db.drop_all()


if __name__ == '__main__':
    unittest.main()

二十三、flask部署

>pip install gunicorn
>gunicorn -w 4 -b 127.0.0.1:5000 -D --access-logfile ./logs/access.log main:app

-w : 进程数
-b : 监听
-D : 后台运行
--access-logfile : 日志文件
main:app : 主文件:app实例

Python-Flask web框架

一、环境部署

二、HelloWorld

三、静态文件目录

四、配置参数的设置

五、run的使用

六、路由

七、转换器

八、接收request请求参数

九、上传文件

十、abort 提前终止视图函数

十一、response构造

十二、返回json响应

十三、cookie

十四、session

十五、请求钩子 hook

十六、Flask-Script

十七、模板template

1、模板变量

2、过滤器

3、表单Flask-WTF扩展

4、模板宏

5、模板闪现信息

十八、数据库扩展sqlalchemy

1、模型类、创建数据表、插入数据

2、插入数据

3、数据库查询

4、更新与删除

十九、管理作者和图书的小案例

二十、数据库maigrate扩展

二十一、蓝图

1、蓝图的基本使用

2、以目录形式定义蓝图

二十二、单元测试

1、页面的单元测试

2、数据库测试

二十三、flask部署