python轻量级框架-flask

简述

Flask 是 Python 生态圈中一个基于 Python 的Web 框架。其轻量、模块化和易于扩展的特点导致其被广泛使用，适合快速开发 Web 应用以及构建小型到中型项目。它提供了开发 Web 应用最基础的工具和组件。之所以称为微框架，是因为它与一些大型 Web 框架（如 Django）不同，并不捆绑数据库管理、表单验证等功能，而是保持了极简的核心，只包含了路由、模板引擎和WSGI服务器的基本功能，使开发者可以根据需求选择合适的扩展来构建应用。掌握 Flask 的基础之后可以深入学习如何实现更复杂的功能，例如用户认证、API 版本管理、性能优化，以及如何与前端框架（如 Vue.js 或 React）进行集成，以构建现代的全栈应用。Flask 的设计哲学是简单优先、灵活性强，让开发者对应用的构建过程有更多的控制。Flask 由 Armin Ronacher 开发，基于两个核心工具库：Werkzeug和Jinja2。

Werkzeug是一个 WSGI工具包，用于处理 HTTP 请求和响应。

Jinja2是一个灵活的模板引擎，用于生成动态 HTML 页面。

Flask 的架构是典型的 MVC（Model-View-Controller）风格中的一种实现。这种架构可以有效分离应用的业务逻辑、数据管理和视图展示，有助于保持代码的清晰性和模块化，核心功能主要包括：

基本要点

1. 路由系统

Flask 使用 @app.route() 装饰器定义 URL 路由，这使得定义一个请求的处理函数变得非常简洁。每个 URL 路径都对应一个视图函数，它负责处理来自客户端的请求并返回合适的响应。类似于之前提到过的fastapi：

from flask import Flask
app = Flask(__name__)

# 路由与视图函数一一对应，程序实例需要知道每个url请求所对应的运行代码是谁，所以程序中必须要创建一个url请求地址到python运行函数的一个映射
@app.route("/hello") # 将路径 /hello 映射到函数 hello_world，当用户访问这个路径时，浏览器会得到 Hello, World! 的响应，url映射的函数，要传参则在上述route（路由）中添加参数申明
def hello_world():
    return "Hello, World!"

# 注意：如果ip设置成0.0.0.0（不仅监听本地端口）且端口对外开发，那么在任意主机上都可以访问该地址，请确保你本地数据处于安全状态不受攻击
if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8080, debug=False)

访问上述路径是会打印出该函数结果：

2. 请求和响应处理

Flask 对 HTTP 请求和响应的处理非常灵活。通过 flask.request 对象，可以访问请求的所有细节，如查询参数、表单数据、上传的文件等。对于响应，开发者可以使用 flask.Response 类来自定义 HTTP 响应。

from flask import Flask, request

app = Flask(__name__)

#注意路由路径不要重名，映射的视图函数也不要重名,methods表示请求方式
@app.route("/greet", methods=["GET", "POST"])
def greet():
    if request.method == "POST":
        name = request.form.get("name")
        return f"Hello, {name}!"
    return """
        <form method="POST">
            Name: <input type="text" name="name">
            <input type="submit" value="Submit">
        </form>
    """

if __name__ == "__main__":
    app.run(debug=False)

访问上述路径时如下图：

3、路由

可以在路径内以/<参数名>的形式指定参数，默认接收到的参数类型是string

以下为flask框架自带的转换器，可以置于参数前将接收的参数转化为对应类型
        string 接受任何不包含斜杠的文本
        int 接受正整数
        float 接受正浮点数
        path 接受包含斜杠的文本

当然也可以自定义转换器（pip install werkzeug）：

from werkzeug.routing import BaseConverter #导入转换器的基类，用于继承方法
from flask import Flask

app = Flask(__name__)

# 自定义转换器类
class RegexConverter(BaseConverter):
    def __init__(self,url_map,regex):
        # 重写父类定义方法
        super(RegexConverter,self).__init__(url_map)
        self.regex = regex

    def to_python(self, value):
        # 重写父类方法，后续功能已经实现好了
        print('to_python方法被调用')
        return value

# 将自定义的转换器类添加到flask应用中
# 具体过程是添加到Flask类下url_map属性（一个Map类的实例）包含的转换器字典属性中
app.url_map.converters['re'] = RegexConverter
# 此处re后括号内的匹配语句，被自动传给我们定义的转换器中的regex属性
# value值会与该语句匹配，匹配成功则传达给url映射的视图函数
@app.route("/index/<re('1\d{10}'):value>")
def index(value):
    print(value)
    return "Hello World!"

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

4、endpoint

每个实例app中都存在一个url_map，这个url_map中包含了url到endpoint的映射；当request请求传来一个url的时候，会在url_map中先通过rule找到endpoint，然后再在view_functions中根据endpoint再找到对应的视图函数view_func

from flask import Flask

app = Flask(__name__)

# endpoint默认为视图函数的名称
@app.route('/branch')
def test():
    return 'check success!'
# 可以在路由中修改endpoint，相当于为视图函数起别名（当视图函数名称很长时适用）
@app.route('/hello',endpoint='hello_test')
def hello_world():
    return 'Hello World!'

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

可以通过view_functions查看到当前endpoint与视图函数的对应情况；
可以通过url_map查看当前url与endpoint的绑定情况；

endpoint默认为视图函数的名称，endpoint相当于给url起一个名字，view_functions内存储的就是url的名字到视图函数的映射，且endpoint在同一个蓝图下也不能重名，要注意即使修改endpoint为其他视图函数名，路由依然是绑定其正下方的第一个视图函数，说明endpoint作用于url，而不是作用于函数名。

5、redirect重定向

在flask 中，重定向是通过flask.redirect(location, code=302)这个函数来实现的，location表示需要重定向的url, 应该配合url_for函数来使用， code表示采用哪个重定向，默认是302，即临时性重定向, 可以修改为301来实现永性重定向；

from flask import Flask,jsonify

app = Flask(__name__)

# endpoint默认为视图函数的名称
#用jsonify库实现将json数据返回给前端
@app.route('/branch')
def test():
    data={'suatus':'check success!'}
    return jsonify(data)
# 可以在路由中修改endpoint，相当于为视图函数起别名（当视图函数名称很长时适用）
@app.route('/hello',endpoint='hello_test')
def hello_world():
    #doing something
    #...
    # redirect重定位（服务器向外部发起一个请求跳转）到一个url界面；
    # url_for给指定的函数构造 URL；
    # return redirect('/items') 不建议这样做,将界面限死了
    return redirect(url_for('items'))

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

6、abort函数

类似于python中的raise函数，可以在需要退出请求的地方抛出错误，并结束该请求；
可以使用errorhandler()装饰器来进行异常的捕获与自定义

from flask import Flask,jsonify

app = Flask(__name__)

# endpoint默认为视图函数的名称
#用jsonify库实现将json数据返回给前端
@app.route('/branch')
def test():
    data={'suatus':'check success!'}
    return jsonify(data)
# 可以在路由中修改endpoint，相当于为视图函数起别名（当视图函数名称很长时适用）

@app.route('/hello',endpoint='hello_test',methods=['GET','POST'])
def hello_world():
    
    if request.method == 'GET':
        #doing something
        pass
    else:
        # abort的用法类似于python中的raise，在网页中主动抛出错误
        abort(404)
        # redirect重定位（服务器向外部发起一个请求跳转）到一个url界面；
        # url_for给指定的函数构造 URL；
        # return redirect('/items') 不建议这样做,将界面限死了
        return redirect(url_for('items'))

# 自定义错误处理方法,将404这个error与Python函数绑定
# 当需要抛出404error时，将会访问下面的代码
@app.errorhandler(404)
def handle_404_error(err):
    # return "发生了错误，错误情况：%s"%err
    # 自定义一个界面
    return render_template('404.html')

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

自定义的404界面：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<!-- 注意图片文件需要放在一个静态文件夹static里 -->
<img src="../static/error404.jpg" alt="" width="1428px" height="57px">
</body>
</html>

二、Flask 的扩展与灵活性

虽然 Flask 是一个微框架，但它具有极强的灵活性，可以自由选择各种扩展来增强功能：

Flask-SQLAlchemy：用于与关系数据库交互，提供 ORM（对象关系映射）支持。是一个流行的 Flask 扩展，它为数据库操作提供了一种更简洁、更 Pythonic 的方式。

Flask-WTF：用于表单处理和验证，简化表单开发。

Flask-Login：用于用户认证与会话管理。

这些扩展可以无缝集成到 Flask 应用中，使开发者在不牺牲灵活性的同时实现复杂的功能。

1、Flask-SQLAlchemy与数据库sqllite交互

#app.py
from flask import Flask
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///127.0.0.1.db"  # 相对路径
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False  # 禁用模型修改跟踪，节省资源
db = SQLAlchemy(app)
migrate = Migrate(app, db) # 数据库表结构更新迁移

class bank_info(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    branch_id = db.Column(db.String(80), unique=True, nullable=False)
    application = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f"<bank_info branch_id={self.branch_id}, application={self.application}>"

@app.route("/add_branch_id")
def add_branch_id():
    # 添加一个新的分支机构
    branch = bank_info(branch_id="****", application="授信总额：****")
    
    # 检查是否已经存在该 branch_id 或 application
    existing_branch = bank_info.query.filter_by(branch_id=branch.branch_id).first()
    if existing_branch:
        return "Branch ID already exists!"
    
    db.session.add(branch)
    db.session.commit()
    return "Branch added!"

@app.route("/check_branch_id")
def get_branch():
    # 查询所有信息
    branches = bank_info.query.all()
    return "<br>".join([f"ID: {branch.id}, branch_id: {branch.branch_id}, application: {branch.application}" for branch in branches])

     
if __name__ == "__main__":
    with app.app_context():
        db.create_all()# 在应用上下文中创建数据库表
    app.run(debug=False)  # 启动

运行后输入路由结果为：

查询结果同理。

注意当表结构需要变更时，需要单独执行命令：

cd ./path/ # 项目目录下，即app.py所在目录
flask db init # 初始化迁移环境，创建一个 migrations/ 文件夹
flask db migrate -m "Add email column to bank_info model"  # -m 用于写一个描述迁移的消息
flask db upgrade # 执行迁移

# 如果删除了原来数据库文件需要重建
from your_app import db
db.create_all()

#直接查看表结构
sqlite3 your_database.db
.schema bank_info

同时需要更新脚本：

#app.py
from flask import Flask
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///127.0.0.1.db"  # 使用相对路径
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False  # 禁用模型修改跟踪，以节省资源
db = SQLAlchemy(app)
migrate = Migrate(app, db)#执行数据库迁移

class bank_info(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    branch_id = db.Column(db.String(80), unique=False, nullable=False)
    application = db.Column(db.String(120), unique=False, nullable=False)
    email = db.Column(db.String(120), nullable=True)  # 添加email字段
    def __repr__(self):
        return f"<bank_info branch_id={self.branch_id}, application={self.application}, email={self.email}>"

@app.route("/add_branch_id")
def add_branch_id():
    # 添加一个新的分支机构
    branch = bank_info(branch_id="**1", application="授信总额：*",email="***@163.com")
    
    # 检查是否已经存在该 branch_id 或 application
    existing_branch = bank_info.query.filter_by(branch_id=branch.branch_id).first()
    if existing_branch:
        return "Branch ID already exists!"
    
    db.session.add(branch)
    db.session.commit()  # 提交到数据库
    return "Branch added!"

@app.route("/check_branch_id")
def get_branch():
    # 查询所有信息
    branches = bank_info.query.all()
    return "<br>".join([f"ID: {branch.id}, branch_id: {branch.branch_id}, application: {branch.application},email:{branch.email}" for branch in branches])

     
if __name__ == "__main__":
#    with app.app_context():
#       db.create_all()# 在应用上下文中创建数据库表
    app.run(debug=False)  # 启动

注意，当字段设置unique为True时，每次插入的数据必须要求不一致，否则无法插入。在项目目录下会生成对应的文件：

2、Jinja2 模板引擎的使用

Jinja2 是 Flask 的默认模板引擎，允许开发者将动态数据嵌入到 HTML 中，生成富有交互性的页面。它支持变量、控制结构（如 for 循环和 if 判断）以及宏（类似于函数的代码片段，可以复用）:

#模板文件存在于 templates/hello.html
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Greeting</title>
</head>
<body>
    <h1>Hello, {
  
  { name }}!</h1>
</body>
</html>

render_template()：可以用于呈现一个开发人员编写的html文件模板
request.method用于获取url接收到的请求方式，以此返回不同的响应页面

在 Flask 应用中使用这个模板:

from flask import Flask, render_template

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

# 路由处理函数
@app.route("/greet/<name>",methods=['GET','POST'])#url映射的函数，要传参则在上述route（路由）中添加参数申明
def greet(name):
    if request.method == 'GET':
        # 想要html文件被该函数访问到，首先要创建一个templates文件，将html文件放入其中
        # 该文件夹需要被标记为模板文件夹，且模板语言设置为jinja2
    return render_template("hello.html", name=name)
        # 此处欲发送post请求，需要在对应html文件的form表单中设置method为post
    elif request.method == 'POST':
        pass

# 启动 Flask 应用
if __name__ == '__main__':
    app.run(debug=False)

render_template() 函数用于渲染 hello.html 模板，并将变量 name 的值传递到模板中，从而动态生成最终的 HTML 页面.

3、构建完整的 CRUD 应用

一个典型的 Web 应用需要对数据进行创建（Create）、读取（Read）、更新（Update）和删除（Delete），这被称为 CRUD 操作。借助 Flask可以很方便地构建一个支持 CRUD 操作的应用：

#用 Flask 构建一个完整的 CRUD API。通过 HTTP 的 POST、GET、PUT 和 DELETE 方法，客户端可以实现对 Item 对象的创建、读取、更新和删除操作

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///crud.db"
db = SQLAlchemy(app)

class Item(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), unique=True, nullable=False)
    def to_dict(self):
        return {"id": self.id, "name": self.name}
    
@app.route("/items", methods=["POST"])
def create_item():
    name = request.json.get("name")
    item = Item(name=name)
    db.session.add(item)
    db.session.commit()
    return jsonify(item.to_dict()), 201

@app.route("/items", methods=["GET"])
def get_items():
    items = Item.query.all()
    return jsonify([item.to_dict() for item in items])

@app.route("/items/<int:item_id>", methods=["PUT"])
def update_item(item_id):
    item = Item.query.get_or_404(item_id)
    item.name = request.json.get("name")
    db.session.commit()
    return jsonify(item.to_dict())

@app.route("/items/<int:item_id>", methods=["DELETE"])
def delete_item(item_id):
    item = Item.query.get_or_404(item_id)
    db.session.delete(item)
    db.session.commit()
    return "", 204

if __name__ == "__main__":
    with app.app_context():
        db.create_all()
    app.run(debug=False)

运行上行脚本，打开postman新建一个文件执行操作：

点击send如果成功则会出现：

在浏览器里面输入对应路由显示为：

表明数据新建成功。

如果想要更新或者是删除操作则分别选择put和DELETE选项，发送给对应请求，请求体路由输入对应ID即可，请求体输入对应新的名称：

4、Flask 中间件与蓝图

4.1. 中间件

中间件（Middleware）是位于请求与响应之间的代码，用于对请求或响应进行处理。Flask 的中间件可以用来做很多事情，例如：记录日志、修改请求、或在响应中增加自定义的 HTTP 头等：

from flask import Flask, jsonify

app = Flask(__name__)

# 定义在请求处理之前执行的函数
@app.before_request
def before_request_func():
    print("Before request is called")

# 定义在响应返回之前执行的函数
@app.after_request
def after_request_func(response):
    print("After request is called")
    # 在这里你可以修改响应（如添加头信息等）
    response.headers["X-Custom-Header"] = "Hello World"
    return response

# 一个简单的路由
@app.route('/')
def home():
    return jsonify(message="Hello, Flask!")

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

@app.before_request 装饰器用于定义在每个请求处理之前执行的函数。这通常用于一些预处理任务，比如验证用户身份、记录日志、设置请求上下文等。

该函数表示每一次请求之前，可以执行某个特定功能的函数；
可以存在多个before_request装饰器，执行顺序是由上到下，先绑定的先执行；
并且先执行 flask app 的 before_request, 再执行 blueprint 的 before_request；
执行时机：当客户端发出请求时，Flask 在请求被处理之前调用 before_request_func。
常见用途：检查用户是否已经登录，处理跨域请求（CORS），或者设置数据库连接等。

@app.after_request 装饰器用于定义在每个请求处理完成后、响应返回给客户端之前执行的函数。这个函数接受一个 response 参数，可以对响应进行修改或者执行一些收尾操作。

执行时机：当请求处理完成并且响应准备返回时，Flask 会调用 after_request_func。你可以对 response 进行修改（例如添加 headers、修改响应内容等）。
常见用途：设置响应头、日志记录、性能监控、跨域处理等。

@app.before_request 和 @app.after_request 是应用级别的装饰器，适用于每一个请求。
@app.after_request 装饰器中的函数必须返回一个响应对象。如果没有对 response 做任何修改，至少需要返回原始的 response 对象。

@app.before_first_request：与before_request的区别是，只在第一次请求之前调用；

该函数表示第一次请求之前可以执行的函数；
执行顺序同样也是先绑定的先执行；

@app.teardown_request：每一次请求之后都会调用；

该函数接收一个参数，该参数是服务器出现的错误信息；
执行顺序也是先绑定的后执行；
只有在请求上下文被 pop 出请求栈的时候才会直接跳转到teardown_request；
所以在被正常调用之前，即使某次请求有抛出错误，该请求也都会被继续执行, 并在执行完后返回 response；

4.2. 蓝图（Blueprint）

当应用规模变大时，代码结构的组织变得至关重要。Flask 提供了蓝图（Blueprint）组件来帮助开发者将应用模块化。在 Flask 中，应用可以通过蓝图将不同的视图函数和路由分离开来，使得代码更加结构化和易于管理。蓝图的作用是将视图函数、错误处理、静态文件、模板等逻辑与主应用程序分开，便于进行模块化开发。在一个大型应用程序中，蓝图可以将不同的功能区域分离，使得不同的功能部分有独立的管理。在 Flask 应用中，可以通过 app.register_blueprint() 将蓝图注册到应用程序中，之后蓝图中的路由和视图就会成为主应用的一部分。

将用户相关的路由逻辑组织在一个独立的蓝图中，并在主应用中注册这个蓝图，从而使代码结构更加清晰和模块化：

# 创建蓝图：user.py
from flask import Blueprint

# 创建一个名为 "user" 的蓝图模块
user_bp = Blueprint("user", __name__)

# 定义蓝图中的一个路由
@user_bp.route("/user/<username>")
def show_user(username):
    return f"User: {username}"

# 注册蓝图
from flask import Flask
from user import user_bp
# from article import article_bp

app = Flask(__name__)

# 注册多个蓝图
app.register_blueprint(user_bp)
# app.register_blueprint(article_bp)

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

5、url与视图函数的绑定

实现url与视图函数的绑定，除了使用路由装饰器@app.route，还可以通过add_url_rule(rule,endpoint=None,view_func=None)方法，其中：
        rule：设置的url
        endpoint：给url设置的名称
        view_func：指定视图函数的名称

from flask import Flask,url_for

app = Flask(__name__)

@app.route('/branch',endpoint='branch')
# 底层其实是用add_url_rule实现的
def check_branch():
    return 'branch Hive is null'

def my_test():
    return '这是测试查询页面'
app.add_url_rule(rule='/test',endpoint='test',view_func=my_test)

# 请求上下文只有在发送request请求时才会被激活，激活后request对象被设置为全局可访问
# 其内部封装了客户端发出的请求数据报文
# 此处是主动生成一个临时的测试请求上下文
with app.test_request_context():
    print(url_for('test')) # 输出结果为/test

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

6、类试图和自定义装饰器

视图还可以由类来实现，即标准类视图：
定义时需要继承flask的views.View这一基类;
每个类视图内必须包含一个dispatch_request方法，每当类视图接收到请求时都会执行该方法，返回值的设定和视图函数相同；
视图函数可以通过@app.route和app.add_url_rule来进行注册（映射到url），但类视图只能通过app.add_url_rule来注册，注册时view_func不能直接使用类名，需要调用基类中的as_view方法来为自己取一个“视图函数名”
采用类视图的最大优势，就是可以把多个视图内相同的东西放在父类中，然后子类去继承父类；而类视图不方便的地方，就是每一个子类都要通过一个add_url_rule来进行注册。