更新于：2024-06-20 20:00
一、代码展示
from flask import Flask, Blueprint
app = Flask(__name__)
app.config['DEBUG'] = True
health = Blueprint('health', __name__)
@app.before_request
def before_request_func():
print('before request')
@app.after_request
def after_request_func(response):
print('after request')
return response
@app.route('/', methods=['GET'])
def hello_world():
return 'Hello, World!'
@health.route('/health_check', methods=['GET'])
def health_check():
return 'success'
if __name__ == '__main__':
app.register_blueprint(health)
app.run(host='127.0.0.1', port=8080)
上面代码是通过 flask 框架编写一段简单的 python web 服务，下面我们要讲解一些当启动的时候 flask 框架做了什么，当客户端发送请求时 flask 如何处理。
二、flask启动流程
1. 初始化应用
flask 应用的入口点，通常会创建一个 Flask 应用对象，这个对象是整个应用的核心，负责路由、配置和运行等工作。而 Flask 会初始化应用的一些基本配置，包括日志、静态文件、模板文件等。
app = Flask(__name__)
2. 配置应用
在创建完 Flask 应用对象后，可以配置应用的一些参数，如调试模式、数据库连接、密钥等。
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = 'secretkey'
app.config.from_object()
app.config.from_pyfile()
从配置文件或环境变量中加载配置，可使用from_object和from_pyfile方法设置，或者直接设置config字典中的值来加载配置。
3. 路由和视图函数
路由是 url 路径和视图函数之间的映射，通过@app.route注册视图函数。当某个 url 被访问时，对应的视图函数会被调用。另外蓝图是模块化的应用组件，可以通过蓝图来组织代码，并在启动时注册这些蓝图。
4. 注册请求钩子
请求钩子，是在请求处理的不同阶段可以执行一些自定义代码。主要的钩子包括 before_request、after_request、teardown_request 和 before_first_request。
5. 启动应用
当所有的配置和路由定义完成后，Flask 应用可以通过调用 app.run() 来启动 WSGI 服务器，Flask 会启动一个内置的 Werkzeug 开发服务器（用于开发的服务器）。
三、flask处理请求流程
1. 接收请求
当客户端（例如浏览器）发出 HTTP 请求时，该请求被发送到服务器。在开发环境中，Flask 自带的 Werkzeug 开发服务器会接收请求。在生产环境中，通常用 WSGI 服务器（Gunicorn、uWSGI）进行的，WSGI 服务器会将请求转发给 Flask 应用，Flask 应用是一个 WSGI 应用。
def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
...
from werkzeug.serving import run_simple
try:
run_simple(t.cast(str, host), port, self, **options)
finally:
self._got_first_request = False
2. 创建上下文
Flask 会创建请求上下文和应用上下文，以便在整个请求处理过程中使用。请求上下文：包含与当前请求相关的数据（例如请求方法、URL、表单数据等）。应用上下文：包含与应用相关的全局数据（例如配置、数据库连接等）。
def wsgi_app(self, environ: dict, start_response: t.Callable) -> t.Any:
ctx = self.request_context(environ)
error: t.Optional[BaseException] = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
...
ctx.pop(error)
3. URL 路由匹配
Flask 使用 URL 路由映射（URL Map）将请求路径与视图函数关联起来。当请求到达时，Flask 会查找 URL 映射，确定哪个视图函数应处理该请求。
def full_dispatch_request(self) -> Response:
if not self._got_first_request:
with self._before_request_lock:
if not self._got_first_request:
for func in self.before_first_request_funcs:
self.ensure_sync(func)()
self._got_first_request = True
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
4. 处理请求前的钩子函数
在处理请求之前，Flask 允许注册一些钩子函数，例如 before_request 和 before_first_request，以便在实际处理请求之前执行一些操作。
def preprocess_request(self) -> t.Optional[ft.ResponseReturnValue]:
names = (None, *reversed(request.blueprints))
for name in names:
if name in self.url_value_preprocessors:
for url_func in self.url_value_preprocessors[name]:
url_func(request.endpoint, request.view_args)
for name in names:
if name in self.before_request_funcs:
for before_func in self.before_request_funcs[name]:
rv = self.ensure_sync(before_func)()
if rv is not None:
return rv
return None
5. 调用视图函数
Flask 匹配 URL 路由后，会调用相应的视图函数。视图函数处理请求并返回一个响应对象或响应数据（例如字符串、字典等）。
def dispatch_request(self) -> ft.ResponseReturnValue:
...
return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
6. 处理请求后的钩子函数
在视图函数处理完成后，Flask 会调用 after_request 和 teardown_request 钩子函数，以便进行任何清理操作或修改响应。
def finalize_request(
self,
rv: t.Union[ft.ResponseReturnValue, HTTPException],
from_error_handler: bool = False,
) -> Response:
response = self.make_response(rv)
try:
response = self.process_response(response)
request_finished.send(self, response=response)
except Exception:
if not from_error_handler:
raise
self.logger.exception("Request finalizing failed with an error while handling an error")
return response
def process_response(self, response: Response) -> Response:
ctx = request_ctx._get_current_object()
for func in ctx._after_request_functions:
response = self.ensure_sync(func)(response)
for name in chain(request.blueprints, (None,)):
if name in self.after_request_funcs:
for func in reversed(self.after_request_funcs[name]):
response = self.ensure_sync(func)(response)
if not self.session_interface.is_null_session(ctx.session):
self.session_interface.save_session(self, ctx.session, response)
return response
7. 生成响应对象
Flask 将视图函数的返回值转换为一个 Response 对象。如果视图函数返回的是一个字符串，Flask 会将其封装为响应对象。虽然 Response 对象在执行后处理前就已经生成了，但要经过后置钩子函数处理，因此放在这个步骤讲述。
def make_response(self, rv):
...
rv = self.response_class(rv, status=status, headers=headers)
...
return rv
8. 发送响应
最终 Flask 将响应对象传递回 WSGI 服务器，后者将响应发送回客户端。
def wsgi_app(self, environ: dict, start_response: t.Callable) -> t.Any:
...
response = self.full_dispatch_request()
...
return response(environ, start_response)
四、flask其他组件
下面简单介绍一些在 flask 开发中常用的一些拓展组件。
1. flask-sqlalchemy
sqlalchemy是 Python 的一个对象关系映射（ORM）库，它提供了一种灵活而强大的方式来访问和操作数据库。核心功能：SQL 表达式语言（允许使用 Python 表达式来构造 SQL 查询）和 ORM（可以使用 Python 类来表示数据库表，并通过类实例来操作数据库记录）。flask-sqlalchemy作为 Flask 框架的一个扩展，主要是简化 flask 应用中使用sqlalchemy，它集成了 Flask 应用的配置和上下文管理，提供了更简便的 API 来使用sqlalchemy的功能。
下面展示一段简单的代码，依赖版本如下:
pip3 install Flask==1.1.2
pip install Flask-SQLAlchemy==2.4.1
pip install Jinja2==2.11.2
pip install MarkupSafe==1.1.1
import os
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
class BaseConfig(object):
_DIALCT = os.environ.get('ENV_GLOBAL_DB_DIALCT', 'mysql')
_DRITVER = os.environ.get('ENV_GLOBAL_DB_DRITVER', 'pymysql')
_HOST = os.environ.get('ENV_GLOBAL_DB_HOST', '127.0.0.1')
_PORT = os.environ.get('ENV_GLOBAL_DB_PORT', '3306')
_USERNAME = os.environ.get('ENV_GLOBAL_DB_USERNAME', 'mjtree')
_PASSWORD = os.environ.get('ENV_GLOBAL_DB_PASSWORD', 'mjtree')
_DBNAME = os.environ.get('ENV_GLOBAL_DB_DBNAME', 'test_db')
_CHARSET = os.environ.get('ENV_GLOBAL_DB_CHARSET', 'utf8mb4')
SQLALCHEMY_DATABASE_URI = f"{_DIALCT}+{_DRITVER}://{_USERNAME}:{_PASSWORD}@{_HOST}:{_PORT}/{_DBNAME}?charset={_CHARSET}"
SQLALCHEMY_TRACK_MODIFICATIONS = True
app = Flask(__name__)
app.config.from_object(BaseConfig)
db = SQLAlchemy(app)
class Users(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(32), unique=True, nullable=False)
password = db.Column(db.String(128), nullable=False)
def __repr__(self):
return f'<User {self.username}>'
with app.app_context():
db.drop_all()
db.create_all()
user_1 = Users(username='admin', password='admin@123')
db.session.add(user_1)
db.session.flush()
assert user_1.id == 1
db.session.add(Users(username='guest', password='guest@123'))
db.session.commit()
result_1 = Users.query.filter(Users.id.in_([1])).first()
assert result_1.username == 'admin'
result_2 = db.session.query(Users).filter(Users.username == 'guest').first()
result_2.username = 'mjtree'
db.session.commit()
db.session.delete(result_2)
db.session.commit()
assert Users.query.count() == 1
2. flask-migrate
在项目开发过程中，经常会发生数据库修改的行为，但是不会直接手动的去修改，而是去修改对应的 ORM 模型，然后再把模型映射到数据库中，而flask-migrate负责这种工作。实际上flask-migrate是基于Alembic进行的一个封装，并集成到 flask 中，而所有的迁移操作其实都是Alembic做的（跟踪模型的变化，并将变化映射到数据库中）。
pip3 install Flask-Migrate==2.5.2
'migrate_demo.py'
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://mjtree:mjtree@127.0.0.1:3306/test_db?charset=utf8mb4'
db = SQLAlchemy(app)
migrate = Migrate(app, db)
class Users(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(32), unique=True, nullable=False)
password = db.Column(db.String(128), nullable=False)
def __repr__(self):
return f'<User {self.username}>'
export FLASK_APP=migrate_demo.py
flask db init
flask db migrate
flask db upgrade
3. flask-restful
flask-restful提供了一些方便的工具来构建 RESTful API，使用flask-restful可以快速地定义资源、解析请求参数、格式化响应数据、定义路由和处理异常等，从而提高开发效率并降低出错的风险。
pip3 install Flask-RESTful==0.3.10
from flask import Flask, request
from flask_restful import Resource, Api
app = Flask(__name__)
api = Api(app)
todos = {}
class TodoSimple(Resource):
def get(self, todo_id):
return {todo_id: todos[todo_id]}
def put(self, todo_id):
todos[todo_id] = request.form['data']
return {todo_id: todos[todo_id]}
api.add_resource(TodoSimple, '/<string:todo_id>')
if __name__ == '__main__':
app.run(debug=True)
curl http://localhost:5000/1 -d "data=按时吃饭" -X PUT
curl http://localhost:5000/1
4. flask-jwt-extended
JWT(JSON Web Token)是一种协定，把 JSON 结构的信息进行加密后变成 Token 传递给客户端，然后客户端通过请求中携带 Token 来与服务器进行交互，沒有携带的请求会被拒绝访问，需要重新验证身份获取 Token。flask-jwt-extended是JWT的一个实现，在开发应用时能够更方便地实现基于 Token 的认证过程。
pip3 install Flask-JWT-Extended-3.25.1
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://mjtree:mjtree@127.0.0.1:3306/test_db?charset=utf8mb4'
app.config['JWT_SECRET_KEY'] = 'xS@!a20pw'
db = SQLAlchemy(app)
bcrypt = Bcrypt(app)
jwt = JWTManager(app)
class Users(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(32), unique=True, nullable=False)
password = db.Column(db.String(128), nullable=False)
def __repr__(self):
return f'<User {self.username}>'
with app.app_context():
db.drop_all()
db.create_all()
@app.route('/register', methods=['POST'])
def register():
data = request.get_json()
hashed_password = bcrypt.generate_password_hash(data['password']).decode('utf-8')
new_user = Users(username=data['username'], password=hashed_password)
db.session.add(new_user)
db.session.commit()
return jsonify(message="User created successfully"), 201
@app.route('/login', methods=['POST'])
def login():
data = request.get_json()
user = Users.query.filter_by(username=data['username']).first()
if user and bcrypt.check_password_hash(user.password, data['password']):
access_token = create_access_token(identity=user.id)
return jsonify(access_token=access_token), 200
return jsonify(message="Invalid credentials"), 401
@app.route('/protected', methods=['GET'])
@jwt_required
def protected():
current_user_id = get_jwt_identity()
user = Users.query.get(current_user_id)
return jsonify(username=user.username, email=user.email), 200
if __name__ == '__main__':
app.run(debug=True)
curl -X POST -H "Content-Type: application/json" -d '{"username": "mjtree", "password": "mjtree@123"}' http://127.0.0.1:5000/register
curl -X POST -H "Content-Type: application/json" -d '{"username": "mjtree", "password": "mjtree@123"}' http://127.0.0.1:5000/login
curl -H "Authorization: Bearer ACCESS_TOKEN" http://127.0.0.1:5000/protected
5.other
还有一些组件，例如：Flask-Mail 、Flask-Caching、Flask-Uploads、Flask-Limiter、Flask-WTF 等，具体可以参考官方文档。
