Python Flask Web框架快速入门

2024-04-20 10:29:02

Flask 入门Demo

Flask 开发环境搭建，执行如下指令：

pip install flask

# 第一节: Flask 快速入门

from flask import Flask
app = Flask(__name__)


@app.route('/flask')
def hello_flask():
   return 'Hello Flask'


app.run()

核心代码剖析：

从 flask 包导入 Flask 类，通过实例化这个类，创建一个程序对象 app。

app = Flask(__name__)

注册一个处理函数，这个函数是处理某个请求的处理函数，Flask 官方把它叫做视图函数（view funciton）。使用 app.route() 装饰器来为这个函数绑定对应的 URL，当用户在浏览器访问这个 URL 的时候，就会触发这个函数，获取返回值，并把返回值显示到浏览器窗口：

@app.route('/flask')
def hello_flask():
   return 'Hello Flask'

最后，Flask类的run()方法在本地开发服务器上运行应用程序。

app.run(host, port, debug, options)

所有参数都是可选的

序号	参数与描述
1	host 要监听的主机名。默认为127.0.0.1（localhost）。设置为“0.0.0.0”以使服务器在外部可用
2	port 默认值为5000
3	debug 默认为false。如果设置为true，则提供调试信息
4	options 要转发到底层的Werkzeug服务器。

Flask 基础知识

Flask 入参类型

flask 支持入参数据类型，如下所示：

转换器	描述
int	整型
float	浮点型
path	接受用作目录分隔符的斜杠
string	默认，字符串

# 第二节: Flask 入参
from flask import Flask, redirect

app = Flask(__name__)


# 字符串入参
@app.route('/strs/<name>')
def strs(name):
    return "Hello %s" % name


# 浮点数入参
@app.route('/floats/<float:version>')
def floats(version):
    return version


# 整数入参
@app.route('/ints/<int:version>')
def ints(version):
    return '整数为 %d' % version


app.run(host='0.0.0.0', port=8888, debug=True)

Flask 重定向(redirect)

url_for()函数用于动态指定函数的URL地址。

# 第三节: Flask 重定向

from flask import Flask, redirect, url_for

app = Flask(__name__)


@app.route('/redicts')
def redicts():
    return redirect('https://www.baidu.com')


@app.route('/admin')
def get_admin():
    return 'Hello Admin'


@app.route('/guest/<guest>')
def get_guest(guest):
    return 'Hello %s as Guest' % guest


@app.route('/user/<name>')
def hello_user(name):
    if name == 'admin':
        return redirect(url_for('get_admin'))
    else:
        return redirect(url_for('get_guest', guest=name))


app.run()

Flask 支持到HTTP方法

默认情况下，Flask路由响应GET请求。但是，可以通过为route()装饰器提供方法参数来更改此首选项。

序号	方法与描述
1	GET 以未加密的形式将数据发送到服务器。最常见的方法。
2	HEAD 和GET方法相同，但没有响应体。
3	POST 用于将HTML表单数据发送到服务器。POST方法接收的数据不由服务器缓存。
4	PUT 用上传的内容替换目标资源的所有当前表示。
5	DELETE 删除由URL给出的目标资源的所有当前表示。

# 第四节: Flask 支持HTTP方法

from flask import Flask

app = Flask(__name__)


@app.route('/get_request', methods=['GET'])
def get_request():
    return 'GET请求'


@app.route('/post_request', methods=['POST'])
def post_request():
    return 'POST请求'


@app.route('/delete_request', methods=['DELETE'])
def delete_request():
    return 'DELETE请求'


@app.route('/put_request', methods=['PUT'])
def put_request():
    return 'PUT请求'


@app.route('/head_request', methods=['HEAD'])
def head_request():
    return 'HEAD请求'


app.run()

Flask 实战一：模拟用户登入

# Flask 模拟用户登入

from flask import Flask, request, jsonify

app = Flask(__name__)


@app.route('/')
def index():
    return '欢迎来到主页'


@app.route('/login', methods=['POST'])
def login():
    my_json = request.get_json()
    user = my_json.get('user')
    password = my_json.get('password')
    if user == 'admin' and password == '123456':
        # 前端和后端统一请求和返回的数据格式,返回json格式需要导入jsonify这个包
        # return jsonify({
        #     "token": "abcd123456",
        #     "birthday": "2024-04-18"
        # })
        return jsonify(token="abcd123456", birthdat="2024-04-18")


app.run()

Flask 模板

在项目下创建 templates 文件夹，用于存放所有模板文件，并在目录下创建一个模板文件: login.html 文件

<html>
   <body>
      <form action = "http://localhost:5000/login" method = "post">
         <p>用户名:</p>
         <p><input type = "text" name = "user" /></p>
         <p>密码:</p>
         <p><input type = "password" name = "password" /></p>
         <p><input type = "submit" value = "submit" /></p>
      </form>
   </body>
</html>

# Flask 模拟用户登入 : 基于Template

from flask import Flask, request, jsonify, render_template

app = Flask(__name__)


@app.route('/')
def index():
    return render_template("login.html")


@app.route('/login', methods=['POST'])
def login():
    # 由json 获取修改为表单获取
    user = request.form['user']
    password = request.form['password']
    if user == 'admin' and password == '123456':
        # 前端和后端统一请求和返回的数据格式,返回json格式需要导入jsonify这个包
        # return jsonify({
        #     "token": "abcd123456",
        #     "birthday": "2024-04-18"
        # })
        return jsonify(token="abcd123456", birthdat="2024-04-18")


app.run()

Flask 模板文件传参

在Python代码中传入字符串，列表，字典到模板中。

from flask import Flask, render_template

app = Flask(__name__)


@app.route('/')
def index():
    # 字符串
    my_str = 'Hello Word'
    # int 类型
    my_int = 10
    # 数组类型
    my_array = [3, 4, 2, 1, 7, 9]
    # 字典类型
    my_dict = {
        'name': 'zhouzhiwengang',
        'age': 31
    }
    return render_template('variable.html',
                           my_str=my_str,
                           my_int=my_int,
                           my_array=my_array,
                           my_dict=my_dict
                           )


app.run()

Flask 模板文件之静态文件

# 第六节: Flask 模板文件之静态文件
from flask import Flask, render_template

app = Flask(__name__)


@app.route("/")
def index():
    return render_template("static.html")


app.run(debug=True)

在项目下创建 templates 文件夹，用于存放所有模板文件，并在目录下创建一个模板文件: static.html 文件

<html>

   <head>
      <script type = "text/javascript"
         src = "{{ url_for('static', filename = 'static.js') }}" ></script>
   </head>

   <body>
      <input type = "button" onclick = "sayHello()" value = "Say Hello" />
   </body>

</html>

在项目下创建 static文件夹，用于存放javascript文件或支持网页显示的CSS文件，并在目录下创建一个js文件: static.js文件

function sayHello() {
   alert("Python 模板文件之静态资源文件")
}

Flask Request对象

Request对象的重要属性如下所列：

Form - 它是一个字典对象，包含表单参数及其值的键和值对。
args - 解析查询字符串的内容，它是问号（？）之后的URL的一部分。
Cookies - 保存Cookie名称和值的字典对象。
files - 与上传文件有关的数据。
method - 当前请求方法。

# 第七节: Flask Request 对象
# Request对象的重要属性如下所列：
# Form - 它是一个字典对象，包含表单参数及其值的键和值对。
# args - 解析查询字符串的内容，它是问号（？）之后的URL的一部分。
# Cookies  - 保存Cookie名称和值的字典对象。
# files - 与上传文件有关的数据。
# method - 当前请求方法。

from flask import Flask, render_template, request

app = Flask(__name__)


@app.route('/')
def student():
    return render_template('student.html')


@app.route('/result', methods=['POST', 'GET'])
def result():
    if request.method == 'POST':
        result = request.form
        return render_template("result.html", result=result)


app.run(debug=True)

在项目下创建 templates 文件夹，用于存放所有模板文件，并在目录下创建一个模板文件: student.html /result.html文件

student.html

<html>
   <body>
      <form action="http://localhost:5000/result" method="POST">
        <p>姓名 <input type = "text" name = "Name" /></p>
        <p>物理 <input type = "text" name = "Physics" /></p>
        <p>化学 <input type = "text" name = "chemistry" /></p>
        <p>数学 <input type ="text" name = "Mathematics" /></p>
        <p><input type = "submit" value = "提交" /></p>
   </form>
   </body>
</html>

result.html

<html>
   <body>
     <table border = 1>
     {% for key, value in result.items() %}
       <tr>
          <th> {{ key }} </th>
          <td> {{ value }}</td>
       </tr>
      {% endfor %}
</table>
   </body>
</html>

Flask Cookie

Cookie以文本文件的形式存储在客户端的计算机上。其目的是记住和跟踪与客户使用相关的数据，以获得更好的访问者体验和网站统计信息。

Cookie 核心方法

设置cookie

默认有效期是临时cookie,浏览器关闭就失效,可以通过 max_age 设置有效期，单位是秒

 res = make_response('set success')
 res.set_cookie('username', 'zhouzhiwengang', max_age=3600)

获取cookie

通过request.cookies的方式，返回的是一个字典。

cookie = request.cookies.get('username')

删除cookie

 res = make_response('del success')
 res.delete_cookie('username')

# 第七节: Flask Cookie 对象
from flask import Flask, make_response, request  # 注意需导入 make_response

app = Flask(__name__)


@app.route('/set_cookie')
def set_cookie():
    res = make_response('set success')
    res.set_cookie('username', 'zhouzhiwengang', max_age=3600)
    return res


@app.route('/get_cookie')
def get_cookie():
    cookie = request.cookies.get('username')
    return cookie


@app.route('/del_cookie')
def del_cookie():
    res = make_response('del success')
    res.delete_cookie('username')
    return res


app.run()

Flask Session

与Cookie不同，Session（会话）数据存储在服务器上。会话是客户端登录到服务器并注销服务器的时间间隔。需要在该会话中保存的数据会存储在服务器上的临时目录中。

为每个客户端的会话分配会话ID。会话数据存储在cookie的顶部，服务器以加密方式对其进行签名。对于此加密，Flask应用程序需要一个定义的SECRET_KEY。

Session对象也是一个字典对象，包含会话变量和关联值的键值对。

Session核心方法

设置密钥

app.secret_key = 'abcd12345678'

设置Session会话变量

session['username'] = request.form['username']

删除Session 会话变量

 session.pop('username', None)

# 第八节: Flask Session 对象

from flask import Flask, session, redirect, url_for, escape, request

app = Flask(__name__)

app.secret_key = 'abcd12345678'


@app.route('/')
def index():
    if 'username' in session:
        # session 会话获取值
        username = session['username']
        return '登录用户名是:' + username + '<br>' + \
               "<b><a href = '/logout'>点击这里注销</a></b>"

    return "您暂未登录， <br><a href = '/login'></b>" + \
           "点击这里登录</b></a>"


@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        user = request.form['username']
        password = request.form['password']
        if user == 'admin' and password == '123456':
            # session 会话设值
            session['username'] = request.form['username']
            return redirect(url_for('index'))
    return '''
   <form action = "" method = "post">
         <p>用户名:</p>
         <p><input type = "text" name = "username" /></p>
         <p>密码:</p>
         <p><input type = "password" name = "password" /></p>
         <p><input type = "submit" value = "submit" /></p>
        <p><input type="submit" value ="登录"/></p>
   </form>
   '''


@app.route('/logout')
def logout():
    # session 会话移除值
    session.pop('username', None)
    return redirect(url_for('index'))


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

Flask 错误代码

Flask类具有带有错误代码的abort()函数。

Flask.abort(code)

Code 参数采用以下值之一：

400 - 用于错误请求
401 - 用于未身份验证的
403 - Forbidden
404 - 未找到
406 - 表示不接受
415 - 用于不支持的媒体类型
429 - 请求过多

# 第九节: Flask 错误

from flask import Flask, session, redirect, url_for, escape, request, abort

app = Flask(__name__)

app.secret_key = 'abcd12345678'


@app.route('/')
def index():
    if 'username' in session:
        # session 会话获取值
        username = session['username']
        return '登录用户名是:' + username + '<br>' + \
               "<b><a href = '/logout'>点击这里注销</a></b>"

    return "您暂未登录， <br><a href = '/login'></b>" + \
           "点击这里登录</b></a>"


@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        user = request.form['username']
        password = request.form['password']
        if user == 'admin' and password == '123456':
            # session 会话设值
            session['username'] = request.form['username']
            return redirect(url_for('index'))
        else:
            # 用户验证不通过
            abort(401)
    return '''
   <form action = "" method = "post">
         <p>用户名:</p>
         <p><input type = "text" name = "username" /></p>
         <p>密码:</p>
         <p><input type = "password" name = "password" /></p>
         <p><input type = "submit" value = "submit" /></p>
        <p><input type="submit" value ="登录"/></p>
   </form>
   '''


@app.route('/logout')
def logout():
    # session 会话移除值
    session.pop('username', None)
    return redirect(url_for('index'))


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

Flask 消息反馈

Flask 模块包含 flash() 方法。它将后端处理消息传递给前端。

# 第十节: Flask 消息反馈

from flask import Flask, redirect, url_for, request, render_template, flash

app = Flask(__name__)
# seesion 会话存储临时目录地址
app.secret_key = 'abcde'


@app.route('/')
def index():
    return render_template('response.html')


@app.route('/login', methods=['GET', 'POST'])
def login():
    error = None
    if request.method == 'POST':
        user = request.form['user']
        password = request.form['password']
        if user == 'admin' and password == '123456':
            flash('登入成功')
            return redirect(url_for('index'))
        else:
            # 用户验证不通过,反馈相关信息
            error = '非法用户名或密码,请重新登入'
    return render_template('login.html', error=error)


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

在项目下创建 templates 文件夹，用于存放所有模板文件，并在目录下创建一个模板文件: login.html /response.html文件

login.html

<html>
   <body>
      <form action = "http://localhost:5000/login" method = "post">
         <p>用户名:</p>
         <p><input type = "text" name = "user" /></p>
         <p>密码:</p>
         <p><input type = "password" name = "password" /></p>
         <p><input type = "submit" value = "submit" /></p>
      </form>
      {% if error %}
         <p><strong>错误信息</strong>: {{ error }}</p>
      {% endif %}
   </body>
</html>

response.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Flask 消息反馈</title>
</head>
<body>
    {% with messages = get_flashed_messages() %}
         {% if messages %}
               {% for message in messages %}
                    <p>{{ message }}</p>
               {% endfor %}
         {% endif %}
    {% endwith %}
<h3>Welcome!</h3>
<a href = "{{ url_for('login') }}">登入</a>
</body>
</html>

Flask 文件上传

在 Flask 中处理文件上传非常简单。它需要一个 HTML 表单，其 enctype 属性设置为“multipart/form-data”，将文件发布到 URL。

URL 处理程序从 request.files[] 对象中提取文件，并将其保存到所需的位置。

每个上传的文件首先会保存在服务器上的临时位置，然后将其实际保存到它的最终位置。

目标文件的名称可以是硬编码的，也可以从 request.files[file] 对象的 filename 属性中获取。但是，建议使用 secure_filename() 函数获取它的安全版本。

可以在 Flask 对象的配置设置中定义默认上传文件夹的路径和上传文件的最大大小。

app.config['UPLOAD_FOLDER'] 定义上传文件夹的路径 

app.config['MAX_CONTENT_LENGTH'] 指定要上传的文件的最大大小（以字节为单位）

# 第十一节: Flask 文件上传

from flask import Flask, render_template, request
from werkzeug.utils import secure_filename

import os

app = Flask(__name__)
UPLOAD_FOLDER = 'upload'
if not os.path.exists(UPLOAD_FOLDER):
    os.makedirs(UPLOAD_FOLDER)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER


@app.route('/upload')
def upload_file():
    return render_template('upload.html')


@app.route('/uploader', methods=['GET', 'POST'])
def uploader():
    if request.method == 'POST':
        f = request.files['file']
        print(request.files)
        f.save(os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename)))
        return '文件上传成功'
    else:
        return render_template('upload.html')


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

Flask 拓展

Flask常用扩展包：

Flask-SQLalchemy：操作数据库；

Flask-script：插入脚本；

Flask-migrate：管理迁移数据库；

Flask-Session：Session存储方式指定；

Flask-WTF：表单；

Flask-Mail：邮件；

Flask-Bable：提供国际化和本地化支持，翻译；

Flask-Login：认证用户状态；

Flask-OpenID：认证；

Flask-RESTful：开发REST API的工具；

Flask-Bootstrap：集成前端Twitter Bootstrap框架；

Flask-Moment：本地化日期和时间；

Flask-Admin：简单而可扩展的管理接口的框架

Flask 拓展之Flask-SQLalchemy

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

安装flask-sqlalchemy扩展, 执行如下指令:

pip install -U Flask-SQLAlchemy

pip install flask-mysqldb

pip install pymysql

SQLAlchemy支持字段类型

类型名	python中类型	说明
Integer	int	普通整数，一般是32位
SmallInteger	int	取值范围小的整数，一般是16位
BigInteger	int或long	不限制精度的整数
Float	float	浮点数
Numeric	decimal.Decimal	普通整数，一般是32位
String	str	变长字符串
Text	str	变长字符串，对较长或不限长度的字符串做了优化
Unicode	unicode	变长Unicode字符串
UnicodeText	unicode	变长Unicode字符串，对较长或不限长度的字符串做了优化
Boolean	bool	布尔值
Date	datetime.date	时间
Time	datetime.datetime	日期和时间
LargeBinary	str	二进制文件

SQLAlchemy列选项

选项名	说明
primary_key	如果为True，代表表的主键
unique	如果为True，代表这列不允许出现重复的值
index	如果为True，为这列创建索引，提高查询效率
nullable	如果为True，允许有空值，如果为False，不允许有空值
default	为这列定义默认值

SQLAlchemy关系选项

选项名	说明
backref	在关系的另一模型中添加反向引用
primary join	明确指定两个模型之间使用的联结条件
uselist	如果为False，不使用列表，而使用标量值
order_by	指定关系中记录的排序方式
secondary	指定多对多中记录的排序方式
secondary join	在SQLAlchemy中无法自行决定时，指定多对多关系中的二级联结条件

Flask-SQLalchemy 实战之快速入门

# 第十一节: Flask 拓展之数据库(flask-sqlalchemy), 快速入门
# 导入Flask及相关扩展库
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

# 创建Flask应用实例
app = Flask(__name__)

# 配置MySQL数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@192.168.43.10/house'
db = SQLAlchemy(app)


# 定义ORM模型，表示数据库表
class Table(db.Model):
    __tablename__ = 'base_house'
    id = db.Column(db.String(255), primary_key=True)


# 路由函数，查询库下所有表名，并返回
@app.route('/')
def get_tables():
    tables = Table.query.all()
    house_list = []
    for user in tables:
        user_data = {
            'id': user.id
        }
        house_list.append(user_data)
    return {'users': house_list}


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

Flask-SQLalchemy 实战之分页查询和参数筛选

# 第十一节: Flask 拓展之数据库(flask-sqlalchemy), 分页 + 入参查询
# 导入Flask及相关扩展库
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

# 创建Flask应用实例
app = Flask(__name__)

# 配置MySQL数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@192.168.43.10/house'
db = SQLAlchemy(app)


# 定义ORM模型，表示数据库表
class Table(db.Model):
    __tablename__ = 'base_house'
    id = db.Column(db.String(255), primary_key=True)
    project_no = db.Column(db.String(255))
    project_name = db.Column(db.String(255))
    project_address = db.Column(db.String(255))


# 路由函数，查询库下所有表名，并返回
@app.route('/', methods=['POST'])
def get_tables():
    parame_json = request.get_json()
    page = parame_json.get('page')
    size = parame_json.get('size')
    name = parame_json.get('name')

    if name:
        houses = Table.query.filter_by(project_name=name).paginate(page=page, per_page=size, error_out=False)
    else:
        houses = Table.query.paginate(page=page, per_page=size, error_out=False)

    house_list = []
    for house in houses:
        house_data = {
            'id': house.id,
            'projectNo': house.project_no,
            'projectName': house.project_name,
            'projectAddress': house.project_address
        }
        house_list.append(house_data)
    return jsonify({
        'users': house_list,
        'total_pages': houses.pages,
        'current_page': houses.page
    })


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

涉及base_house 表DDL：

CREATE TABLE `base_house` (
  `id` varchar(64) NOT NULL,
  `project_no` varchar(128) DEFAULT NULL,
  `project_name` varchar(256) DEFAULT NULL,
  `project_address` varchar(256) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

Flask 拓展之Flask-Login

扩展 Flask-Login 提供了实现用户认证需要的各类功能函数，我们将使用它来实现程序的用户认证，首先来安装它：

pip install flask-login

app.py：初始化 Flask-Login

from flask_login import LoginManager

login_manager = LoginManager(app)  # 实例化扩展类

@login_manager.user_loader
def load_user(user_id):  # 创建用户加载回调函数，接受用户 ID 作为参数
    user = User.query.get(int(user_id))  # 用 ID 作为 User 模型的主键查询对应的用户
    return user  # 返回用户对象

Table模型类继承 Flask-Login 提供的 UserMixin 类：

# 定义ORM模型，表示数据库表
class Table(UserMixin, db.Model):
    __tablename__ = 't_admin'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(30))
    userpwd = db.Column(db.String(100))

# 第十二节: Flask 拓展之数据库(flask-sqlalchemy)/认证框架(flask-login), 使用用户登入并方法鉴权
from flask import Flask, request, jsonify, redirect, url_for, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user


# 创建Flask应用实例
app = Flask(__name__)
# seesion 会话存储临时目录地址
app.secret_key = 'abcde1234'

# 配置MySQL数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@192.168.43.10/house'
db = SQLAlchemy(app)

# 初始化登录管理器
login_manager = LoginManager()
login_manager.init_app(app)


# 定义ORM模型，表示数据库表
class Table(UserMixin, db.Model):
    __tablename__ = 't_admin'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(30))
    userpwd = db.Column(db.String(100))


# 加载用户的回调函数
@login_manager.user_loader
def load_user(user_id):
    return Table.query.get(int(user_id))


# 定义登录路由
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        user = Table.query.filter_by(username=username).first()
        if user and user.userpwd == password:
            login_user(user)
            return redirect(url_for('dashboard'))
    return render_template('auth_login.html')


# 定义需要鉴权的页面
@app.route('/dashboard')
@login_required
def dashboard():
    return render_template('dashboard.html')


# 定义登出路由
@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('login'))


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

Flask 拓展之Flask-JWT-Extended

JWT简介

具体原理请参考：JSON Web Token 入门教程

JWT结构

JWT由三部分组成：

头部（Header）：通常包含令牌的类型（JWT）和使用的加密算法。
载荷（Payload）：包含有关用户或其他数据的信息。例如，用户ID、角色或其他自定义数据。
签名（Signature）：由头部、载荷和密钥组合而成的签名，用于验证令牌的完整性和来源可信度。

JWT生成和校验

用户登录时，服务器使用密钥签署JWT，并将其返回给客户端。
客户端在以后的请求中发送JWT作为身份验证令牌。
服务器验证JWT的签名以确保其完整性，然后使用载荷中的信息进行用户身份验证和授权。

Flask-JWT-Extended

Flask-JWT-Extended是一个Python库，用于在 Flask 应用程序中添加JSON Web令牌（JWT）支持。它是一个插件，可以通过安装它来扩展Flask应用程序的功能。

官网地址：Flask-JWT_Extended 官网

Flask-JWT-Extended 安装

pip install Flask-JWT-Extended

实战：Flask-SQLalchemy + MySQL 8 + Flask-JWT-Extended 实现前后端分离用户认证和鉴权。

# _*_ coding : UTF-8_*_
# 开发者 ： zhuozhiwengang
# 开发时间 : 2024/4/19 9:26
# 文件名称 : 19
# 开发工具 : PyCharm
# 第十二节: Flask 拓展之数据库(flask-sqlalchemy)/认证框架(flask-jwt-extend), 实现用户前后端分离认证和鉴权
from flask import Flask, request, jsonify, redirect, url_for, render_template, abort
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import (JWTManager, jwt_required, create_access_token, get_jwt_identity)


# 创建Flask应用实例
app = Flask(__name__)
# 用于签名JWT的密钥
app.config['JWT_SECRET_KEY'] = 'abc123'

# 配置MySQL数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@192.168.43.10/house'
db = SQLAlchemy(app)

# 初始化JWT扩展
jwt = JWTManager(app)


# 定义ORM模型，表示数据库表
class Table(db.Model):
    __tablename__ = 't_admin'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(30))
    userpwd = db.Column(db.String(100))


# 定义登录路由
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        user = Table.query.filter_by(username=username).first()
        if user and user.userpwd == password:
            access_token = create_access_token(identity=username)
            return jsonify(access_token=access_token)
        else:
            # 用户名或密码错误
            abort(401)


# 定义需要鉴权的页面
@app.route('/dashboard')
@jwt_required()  # 这个装饰器要求请求必须携带有效的JWT令牌
def dashboard():
    # 使用get_jwt_identity访问当前用户的身份
    current_user = get_jwt_identity()
    return jsonify(logged_in_as=current_user)


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

第一种情况：输入错误用户名或密码，提示401错误代码

控制台输出信息：

WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit
127.0.0.1 - - [19/Apr/2024 09:36:45] "POST /login HTTP/1.1" 401 -

第二种情况：输入正确用户名和密码，获取凭证Token

第三种情况：拼接头信息，访问受保护资源

Flask-JWT-Extended 核心代码讲解

初始化

from flask import Flask
from flask_jwt_extended import (JWTManager, jwt_required, create_access_token, get_jwt_identity)

app = Flask(__name__)
# 用于签名JWT的密钥
app.config['JWT_SECRET_KEY'] = 'your-secret-key' 

# 初始化JWT扩展
jwt = JWTManager(app)

生成和校验

定义了 /login 路由，用于用户登录并获取JWT令牌。在这个路由中，首先从请求中获取用户名和密码（这里是 “zzg” 和 “123456”）。如果匹配成功，就使用 create_access_token 函数生成JWT令牌，并返回给客户端。
定义了 /protected 路由，它是受保护的路由，只有在请求中包含有效的JWT令牌时才能访问。这是通过 @jwt_required() 装饰器实现的。
- 如果请求中没有有效的JWT令牌，访问该路由会返回未授权的响应。
- 如果令牌有效，路由会使用 get_jwt_identity() 函数获取JWT中的身份信息（在示例中为用户名）然后返回一个JSON响应，显示已登录的用户

Flask-JWT-Extended 优化：设置Token有效期、刷新Token

设置Token有效期

设置JWT的过期时间是一种重要的安全措施，可以帮助确保令牌不会无限期有效，提高了应用程序的安全性。

方法一：

使用 app.config['JWT_ACCESS_TOKEN_EXPIRES'] 来设置JWT的访问token默认过期时间为1小时。

# 设置ACCESS_TOKEN的默认过期时间为1小时
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)

方法二：

当使用create_access_token函数创建JWT令牌时，也可以通过传递expires_delta参数来覆盖默认的过期时间，例如：

这将覆盖默认的过期时间，使得令牌在30分钟后过期。

from datetime import timedelta
# 设置ACCESS_TOKEN的默认过期时间为30分钟
access_token = create_access_token(identity=username, expires_delta=timedelta(minutes=30))

刷新Token

认证Token与刷新Token差异

	访问token（Access Token）	刷新token（Refresh Token）
用途	用于访问受保护的资源	用于获取新的访问token
生命周期	默认为15分钟	默认为30天
显式指定生命周期	JWT_ACCESS_TOKEN_EXPIRES	JWT_REFRESH_TOKEN_EXPIRES
储存方式	在请求的头信息（Header）中的 “Authorization” 字段中	一般存储在服务器端的数据库

每个用户生成的刷新token和访问token是一一对应的，

当用户登录成功后，服务器会为该用户生成一对刷新token和访问token，并将它们关联到用户的身份（通常是用户的用户名或ID）。这样，每个用户都有自己唯一的刷新token和访问token。

刷新token用于获取新的访问token，以延长用户的会话时间。只有拥有有效的刷新token的用户才能获取新的访问token，而访问token则用于实际访问受保护的资源。

实战：Flask-SQLalchemy + MySQL 8 + Flask-JWT-Extended 实现前后端分离用户认证和鉴权。添加刷新Token和使用刷新Token鉴权。

# 第十二节: Flask 拓展之数据库(flask-sqlalchemy)/认证框架(flask-jwt-extend), 实现用户前后端分离认证和鉴权
from flask import Flask, request, jsonify, redirect, url_for, render_template, abort
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import (JWTManager, jwt_required, create_access_token, get_jwt_identity, create_refresh_token)


# 创建Flask应用实例
app = Flask(__name__)
# 用于签名JWT的密钥
app.config['JWT_SECRET_KEY'] = 'abc123'

# 配置MySQL数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@192.168.43.10/house'
db = SQLAlchemy(app)

# 初始化JWT扩展
jwt = JWTManager(app)


# 定义ORM模型，表示数据库表
class Table(db.Model):
    __tablename__ = 't_admin'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(30))
    userpwd = db.Column(db.String(100))


# 定义登录路由
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        user = Table.query.filter_by(username=username).first()
        if user and user.userpwd == password:
            access_token = create_access_token(identity=username)
            refresh_token = create_refresh_token(identity=username)
            return jsonify(access_token=access_token, refresh_token=refresh_token)
        else:
            # 用户名或密码错误
            abort(401)


# 使用刷新token获取新的访问token
@app.route("/refresh", methods=["POST"])
@jwt_required(refresh=True)  # 使用刷新token进行验证
def refresh():
    current_user = get_jwt_identity()
    access_token = create_access_token(identity=current_user)
    return jsonify(access_token=access_token)


# 定义需要鉴权的页面
@app.route('/dashboard')
@jwt_required()  # 这个装饰器要求请求必须携带有效的JWT令牌
def dashboard():
    # 使用get_jwt_identity访问当前用户的身份
    current_user = get_jwt_identity()
    return jsonify(logged_in_as=current_user)


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

Flask 拓展之flask_restful

flask_restful安装

pip install flask_restful

实战：Flask-SQLalchemy + MySQL 8 +Flask_Restful 实现Restful 接口

from flask import Flask, request, jsonify
from flask_restful import Api, Resource
from flask_sqlalchemy import SQLAlchemy

# 创建Flask应用实例
app = Flask(__name__)

# 配置MySQL数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@192.168.43.10/house'
db = SQLAlchemy(app)
api = Api(app)


# 定义ORM模型，表示数据库表
class Table(db.Model):
    __tablename__ = 'base_house'
    id = db.Column(db.String(255), primary_key=True)
    project_no = db.Column(db.String(255))
    project_name = db.Column(db.String(255))
    project_address = db.Column(db.String(255))


class UserAPI(Resource):
    def get(self, user_id):
        user = Table.query.get(user_id)
        if user:
            return {'id': user.id, 'projectNo': user.project_no, 'projectName': user.project_name, 'projectAddress': user.project_address}
        else:
            return {'message': 'House not found'}, 404

    def post(self):
        data = request.get_json()
        new_user = Table(project_no=data['projectNo'], project_name=data['projectName'], project_address=data['projectAddress'], id=data['id'])
        db.session.add(new_user)
        db.session.commit()
        return {'message': '创建成功'}, 201

    def delete(self, user_id):
        user = Table.query.get(user_id)
        if user:
            db.session.delete(user)
            db.session.commit()
            return {'message': '删除成功'}
        else:
            return {'message': '记录未找到'}, 404


api.add_resource(UserAPI, '/user', '/user/<user_id>')

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

Flask 高级

待补充Flask高级内容主要涉及：Python 多线程、Python连接Redis、Python连接MongoDB、Python 连接Elasticsearch、Python MinoIO 文件服务器。

码农公寓