首页 > tutorial > Flasktutorial > Flasktutoriallist > Flaskin间件 and hookfunction

Flaskin间件 and hookfunction

overview

in WebapplicationDevelopmentin, in间件 and hookfunction is 非常 important concepts. 它们允许我们 in requestprocessing 不同阶段插入自定义code, implementation诸such aslog记录, authenticationauthorization, errorprocessingetc.functions. Flaskproviding了丰富 in间件 and hookfunctionmechanism, 本章节将详细介绍这些functions usingmethod.

1. Flask requestprocessing流程

in Understandin间件 and hookfunction之 before , 我们需要先UnderstandFlask requestprocessing流程:

  1. 客户端发送HTTPrequest to Flaskapplication
  2. Flaskapplicationcreationrequest on under 文 and application on under 文
  3. 执行request before hookfunction (before_request)
  4. 匹配URLrouting, 调用 for 应 视graphfunction
  5. 执行视graphfunction, 生成response
  6. 执行request after hookfunction (after_request)
  7. 返回response给客户端
  8. 执行request结束hookfunction (teardown_request)
  9. 销毁request on under 文 and application on under 文

such as果 in requestprocessing过程in发生error, Flask会调用 for 应 errorprocessingfunction.

2. Flaskhookfunction

Flaskproviding了 many 种hookfunction, 允许我们 in requestprocessing 不同阶段插入自定义code.

2.1 request before processing (before_request)

before_requesthookfunction in 每个requestprocessing之 before 执行, 无论request匹配哪个routing. 我们可以using它来implementation诸such asauthenticationauthorization, log记录, request预processingetc.functions. 

from flask import Flask, request, g, redirect, url_for

app = Flask(__name__)

@app.before_request
def before_request():
    # 记录request开始时间
    g.start_time = time.time()
    
    # log记录
    app.logger.info(f"Request started: {request.method} {request.url}")
    
    # authenticationcheck
    if request.endpoint != 'login' and not g.user:
        return redirect(url_for('login'))

in on 面 例子in, 我们usingbefore_requesthookfunction来记录request开始时间, 记录log, 并checkuser is 否已authentication.

2.2 request before processing (before_first_request)

before_first_requesthookfunction只 in application启动 after 第一个requestprocessing之 before 执行, 之 after 不会再执行. 我们可以using它来初始化datalibrary连接, 加载configurationetc.. 

@app.before_first_request
def before_first_request():
    # 初始化datalibrary连接
    init_db()
    
    # 加载configuration
    load_config()
    
    app.logger.info("Application initialized")

2.3 request after processing (after_request)

after_requesthookfunction in requestprocessing成功 after 执行, 即视graphfunction成功返回response时. 它接收responseobjectserving asparameter, 并返回modify after responseobject. 我们可以using它来添加response头, 记录response时间, modifyresponse in 容etc.. 

@app.after_request
def after_request(response):
    # 计算requestprocessing时间
    if hasattr(g, 'start_time'):
        processing_time = time.time() - g.start_time
        response.headers['X-Processing-Time'] = str(processing_time)
    
    # 添加securityresponse头
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    
    # log记录
    app.logger.info(f"Request completed: {response.status}")
    
    return response

in on 面 例子in, 我们usingafter_requesthookfunction来添加responseprocessing时间头, securityresponse头, 并记录responsestatus.

2.4 request结束processing (teardown_request)

teardown_requesthookfunction in requestprocessing结束 after 执行, 无论request is 否成功. 它接收一个errorobjectserving asparameter, such as果requestprocessing成功, 该parameter for None. 我们可以using它来释放resource, 关闭datalibrary连接etc.. 

@app.teardown_request
def teardown_request(error):
    # 关闭datalibrary连接
    if hasattr(g, 'db_conn'):
        g.db_conn.close()
    
    # log记录
    if error:
        app.logger.error(f"Request error: {error}")
    app.logger.info("Request teardown completed")

2.5 request结束processing (teardown_appcontext)

teardown_appcontexthookfunction in application on under 文销毁时执行, 无论request is 否成功. 它 and teardown_requestclass似, 但它 is 针 for application on under 文 , 而不 is request on under 文. 

@app.teardown_appcontext
def teardown_appcontext(error):
    # 关闭application级别 resource
    if hasattr(app, 'app_resource'):
        app.app_resource.close()
    
    app.logger.info("App context teardown completed")

3. Flaskin间件

in间件 is a in Webserver and application之间 软件层, 它可以拦截 and processingHTTPrequest and response. in Flaskin, 我们可以throughcreation自定义in间件来scaleapplication functions.

3.1 自定义in间件

要creation自定义in间件, 我们需要creation一个class, 该classimplementation__call__method, 该method接收environmentvariable and 启动response functionserving asparameter, 并返回response. 

class CustomMiddleware:
    def __init__(self, app):
        self.app = app
    
    def __call__(self, environ, start_response):
        # request before processing
        print("Middleware: Before request")
        
        # 调用applicationprocessingrequest
        response = self.app(environ, start_response)
        
        # request after processing
        print("Middleware: After request")
        
        return response

# applicationin间件
app.wsgi_app = CustomMiddleware(app.wsgi_app)

3.2 usingin间件装饰器

我们还可以using装饰器来creationin间件: 

def custom_middleware(app):
    def middleware(environ, start_response):
        # request before processing
        print("Middleware: Before request")
        
        # 调用applicationprocessingrequest
        response = app(environ, start_response)
        
        # request after processing
        print("Middleware: After request")
        
        return response
    return middleware

# applicationin间件
app.wsgi_app = custom_middleware(app.wsgi_app)

3.3 常用in间件

Flaskcommunityproviding了许 many 常用 in间件, 例such as:

  • Flask-CORS - processing跨域resource共享 (CORS)
  • Flask-Gzip - 压缩response in 容
  • Flask-SSLify - 强制usingHTTPS
  • Flask-WTF - 表单processing and CSRF保护
  • Flask-Login - userauthentication

我们可以throughinstallation这些scale来using它们providing in间件functions.

4. errorprocessing

Flask允许我们自定义errorprocessingfunction, 当发生specificerror时调用. 我们可以usingerrorhandler装饰器来registererrorprocessingfunction.

4.1 HTTPerrorprocessing

我们可以 for specific HTTPstatus码registererrorprocessingfunction: 

@app.errorhandler(404)
def not_found_error(error):
    return render_template('errors/404.html'), 404

@app.errorhandler(403)
def forbidden_error(error):
    return render_template('errors/403.html'), 403

@app.errorhandler(500)
def internal_server_error(error):
    # 记录errorinformation
    app.logger.error(f"Internal Server Error: {error}")
    return render_template('errors/500.html'), 500

4.2 exceptionclass型processing

我们还可以 for specific exceptionclass型registererrorprocessingfunction: 

@app.errorhandler(Exception)
def handle_exception(error):
    # 记录exceptioninformation
    app.logger.exception(f"Unhandled exception: {error}")
    # 返回500error
    return render_template('errors/500.html'), 500

# processingdatalibraryexception
from sqlalchemy.exc import SQLAlchemyError

@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
    # rollbackdatalibrarytransaction
    db.session.rollback()
    app.logger.error(f"Database error: {error}")
    return render_template('errors/db_error.html'), 500

4.3 蓝graph级别 errorprocessing

我们还可以 in 蓝graph级别registererrorprocessingfunction, 这些function只会processing蓝graph in 部 error: 

@main.errorhandler(404)
def main_not_found_error(error):
    return render_template('main/404.html'), 404

@auth.errorhandler(401)
def auth_unauthorized_error(error):
    return render_template('auth/401.html'), 401

5. on under 文processing器

on under 文processing器允许我们向所 has 模板注入variable, 而不需要 in 每个视graphfunctionin显式传递. 我们可以usingcontext_processor装饰器来register on under 文processing器.

5.1 application级别 on under 文processing器

@app.context_processor
def inject_global_vars():
    return {
        'app_name': 'My Flask App',
        'current_year': datetime.now().year,
        'current_user': current_user,
        'is_admin': current_user.is_authenticated and current_user.is_admin
    }

in on 面 例子in, 我们向所 has 模板注入了app_name, current_year, current_user and is_adminvariable.

5.2 蓝graph级别 on under 文processing器

我们还可以 in 蓝graph级别register on under 文processing器, 这些processing器只会向蓝graph in 部 模板注入variable: 

@main.context_processor
def main_context_processor():
    return {
        'sidebar_items': get_sidebar_items(),
        'recent_posts': get_recent_posts()
    }

5.3 模板全局function

除了向模板注入variable, 我们还可以向模板注入function: 

@app.template_global(name='format_datetime')
def format_datetime(value, format='%Y-%m-%d %H:%M:%S'):
    if value is None:
        return ''
    return value.strftime(format)

@app.template_global(name='truncate')
def truncate(text, length=100, suffix='...'):
    if len(text) <= length:
        return text
    return text[:length] + suffix

in 模板in, 我们可以直接调用这些function: 

<p>release时间: {{ format_datetime(post.created_at) }}</p>
<p>{{ truncate(post.content, 200) }}</p>

5.4 模板filter器

模板filter器允许我们 in 模板in for variableforprocessing. 我们可以usingtemplate_filter装饰器来register自定义filter器: 

@app.template_filter(name='capitalize_words')
def capitalize_words(text):
    return ' '.join(word.capitalize() for word in text.split())

@app.template_filter(name='pluralize')
def pluralize(n, singular='', plural='s'):
    return singular if n == 1 else plural

in 模板in, 我们可以using管道符号 (|) 来applicationfilter器: 

<p>{{ 'hello world' | capitalize_words }}</p>
<p>{{ 1 }} item{{ 1 | pluralize }}</p>
<p>{{ 2 }} item{{ 2 | pluralize }}</p>

6. Flask on under 文

Flaskin has 两种 on under 文: application on under 文 and request on under 文. 它们providing了访问Flaskcorefunctions 方式, 例such asconfiguration, datalibrary连接, requestdataetc..

6.1 application on under 文

application on under 文 (app context) package含application级别 data, 例such asconfiguration, datalibrary连接etc.. 它 in application启动时creation, in application关闭时销毁. 我们可以usingapp_context()method来creationapplication on under 文: 

with app.app_context():
    # 访问applicationconfiguration
    secret_key = app.config['SECRET_KEY']
    
    # 执行datalibraryoperation
    db.create_all()

6.2 request on under 文

request on under 文 (request context) package含request级别 data, 例such asrequest头, requestparameter, sessiondataetc.. 它 in request to 达时creation, in requestprocessing结束时销毁. 我们可以usingtest_request_context()method来creationtestrequest on under 文: 

with app.test_request_context('/?name=test'):
    # 访问requestdata
    name = request.args.get('name')
    path = request.path
    
    # 访问sessiondata
    session['user_id'] = 1
    user_id = session.get('user_id')

6.3 on under 文局部variable

Flaskproviding了一些 on under 文局部variable, 这些variable只能 in specific on under 文in访问:

  • current_app - 当 before applicationinstance, 只能 in application on under 文in访问
  • g - 用于storerequest级别 data, 只能 in request on under 文in访问
  • request - 当 before requestobject, 只能 in request on under 文in访问
  • session - 当 before sessionobject, 只能 in request on under 文in访问

我们可以 in 视graphfunction, hookfunction and on under 文processing器inusing这些variable.

7. practicalapplicationexample

让我们through一个完整 example来演示such as何usingFlask in间件 and hookfunction.

7.1 log记录in间件

import time
from flask import Flask, request, g

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

# request before processing: 记录request开始时间 and log
@app.before_request
def before_request():
    # 记录request开始时间
    g.start_time = time.time()
    
    # 记录requestlog
    app.logger.info(f"Request received: {request.method} {request.url}")
    app.logger.debug(f"Request headers: {dict(request.headers)}")
    if request.method in ['POST', 'PUT', 'PATCH']:
        app.logger.debug(f"Request body: {request.get_data(as_text=True)}")

# request after processing: 记录response时间 and log
@app.after_request
def after_request(response):
    # 计算requestprocessing时间
    processing_time = time.time() - g.start_time
    
    # 添加response头
    response.headers['X-Processing-Time'] = str(processing_time)
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    
    # 记录responselog
    app.logger.info(f"Request completed: {response.status} ({processing_time:.3f}s)")
    app.logger.debug(f"Response headers: {dict(response.headers)}")
    
    return response

# request结束processing: 释放resource
@app.teardown_request
def teardown_request(error):
    # such as果发生error, 记录errorlog
    if error:
        app.logger.error(f"Request error: {error}")
    
    app.logger.info("Request teardown completed")

# errorprocessing
@app.errorhandler(404)
def not_found_error(error):
    app.logger.warning(f"404 Not Found: {request.url}")
    return {
        'error': 'Not Found',
        'message': f'The requested URL {request.url} was not found on the server.',
        'status_code': 404
    }, 404

@app.errorhandler(500)
def internal_server_error(error):
    app.logger.error(f"500 Internal Server Error: {error}")
    return {
        'error': 'Internal Server Error',
        'message': 'An unexpected error occurred on the server.',
        'status_code': 500
    }, 500

#  on  under 文processing器: 向所 has 模板注入variable
@app.context_processor
def inject_global_vars():
    return {
        'app_name': 'Flask Demo App',
        'current_year': time.localtime().tm_year
    }

# 模板filter器: format时间
@app.template_filter(name='format_time')
def format_time(timestamp, format='%Y-%m-%d %H:%M:%S'):
    return time.strftime(format, time.localtime(timestamp))

# routingexample
@app.route('/')
def home():
    return render_template('home.html', message='Hello, Flask!')

@app.route('/api/data')
def api_data():
    return {
        'message': 'This is a JSON response',
        'timestamp': time.time(),
        'data': [1, 2, 3, 4, 5]
    }

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

7.2 authenticationin间件

from flask import Flask, request, g, redirect, url_for, jsonify
from flask_login import Loginmanagementr, UserMixin, login_user, logout_user, login_required, current_user

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

# configurationLoginmanagementr
login_manager = Loginmanagementr()
login_manager.init_app(app)
login_manager.login_view = 'login'
login_manager.login_message_category = 'info'

# mockuserdata
users = {
    'admin@example.com': {'password': 'admin123', 'id': 1, 'name': 'Admin User', 'is_admin': True},
    'user@example.com': {'password': 'user123', 'id': 2, 'name': 'Regular User', 'is_admin': False}
}

class User(UserMixin):
    def __init__(self, user_data):
        self.id = user_data['id']
        self.name = user_data['name']
        self.email = user_data['email']
        self.is_admin = user_data['is_admin']

@login_manager.user_loader
def load_user(user_id):
    for email, user_data in users.items():
        if user_data['id'] == int(user_id):
            user_data['email'] = email
            return User(user_data)
    return None

# authenticationin间件: checkAPIrequest authenticationtoken
@app.before_request
def api_auth_middleware():
    # 跳过login and registerrouting
    if request.endpoint in ['login', 'register']:
        return None
    
    # APIroutingauthentication
    if request.path.startswith('/api/'):
        # 获取authenticationtoken
        auth_header = request.headers.get('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            return jsonify({'error': 'Unauthorized', 'message': 'Missing or invalid authorization header'}), 401
        
        token = auth_header.split(' ')[1]
        #  simple  tokenverification (practicalapplicationin应该usingJWT or othersecurity mechanisms) 
        if token != 'valid-token':
            return jsonify({'error': 'Unauthorized', 'message': 'Invalid token'}), 401
    
    return None

# routing
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        email = request.form['email']
        password = request.form['password']
        
        if email in users and users[email]['password'] == password:
            user_data = users[email].copy()
            user_data['email'] = email
            user = User(user_data)
            login_user(user)
            next_page = request.args.get('next')
            return redirect(next_page) if next_page else redirect(url_for('home'))
        
        return {'error': 'Invalid credentials'}, 401
    
    return render_template('login.html')

@app.route('/logout')
def logout():
    logout_user()
    return redirect(url_for('home'))

@app.route('/')
def home():
    if current_user.is_authenticated:
        return f"Hello, {current_user.name}!"
    return "Hello, Guest! Please login."

@app.route('/admin')
@login_required
def admin():
    if not current_user.is_admin:
        return {'error': 'Forbidden', 'message': 'You do not have permission to access this resource'}, 403
    return "Welcome to the admin panel!"

@app.route('/api/users')
def api_users():
    # 只返回user basicinformation
    return jsonify([{'id': user['id'], 'name': user['name']} for user in users.values()])

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

8. best practices

  • 保持hookfunction简洁 - hookfunction应该只做一件事, 并且保持简洁. such as果需要implementation complex functions, 应该将其拆分 for many 个 small function.
  • usinggobjectstorerequest级别 data - gobject is a 很 good 地方来storerequest级别 data, 例such asdatalibrary连接, userinformationetc..
  • 合理using on under 文processing器 - on under 文processing器应该只注入必要 variable, 避免注入过 many 不常用 variable, 影响performance.
  • 记录详细 log - in hookfunctionin记录详细 log, has 助于debug and monitorapplication.
  • processingerror and exception - 确保 in hookfunctionin正确processingerror and exception, 避免application崩溃.
  • testhookfunction - writingtest用例来testhookfunction functions, 确保它们按照预期工作.
  • using蓝graph组织hookfunction - for 于 big 型application, 应该using蓝graph来组织hookfunction, 避免将所 has hookfunction都放 in 主applicationfilein.

summarized

本章节介绍了Flask in间件 and hookfunction, including:

  • Flask requestprocessing流程
  • 各种hookfunction usingmethod: before_request, before_first_request, after_request, teardown_request, teardown_appcontext
  • such as何creation and using自定义in间件
  • errorprocessingfunction usingmethod
  • on under 文processing器, 模板全局function and 模板filter器
  • Flask on under 文 (application on under 文 and request on under 文)
  • practicalapplicationexample and best practices

in间件 and hookfunction is FlaskapplicationDevelopmentin important tool, 它们允许我们 in requestprocessing 不同阶段插入自定义code, implementation各种functions. through合理using这些functions, 我们可以improvingapplication 可maintenance性, 可scale性 and security性.

codeexample

以 under is a 完整 Flaskin间件 and hookfunctionexample: 

from flask import Flask, request, g, jsonify
import time
import logging

# configurationlog
logging.basicConfig(level=logging.INFO)

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

# request before processing
@app.before_request
def before_request():
    g.start_time = time.time()
    app.logger.info(f"Request: {request.method} {request.url}")
    # 记录requestparameter
    if request.args:
        app.logger.debug(f"Query params: {dict(request.args)}")
    if request.form:
        app.logger.debug(f"Form data: {dict(request.form)}")

# request after processing
@app.after_request
def after_request(response):
    processing_time = time.time() - g.start_time
    response.headers['X-Processing-Time'] = f"{processing_time:.3f}s"
    app.logger.info(f"Response: {response.status} ({processing_time:.3f}s)")
    return response

# request结束processing
@app.teardown_request
def teardown_request(error):
    if error:
        app.logger.error(f"Error: {error}")
    app.logger.info("Request teardown")

# errorprocessing
@app.errorhandler(404)
def not_found(error):
    return jsonify({
        'error': 'Not Found',
        'message': f"The requested URL {request.url} was not found",
        'status_code': 404
    }), 404

@app.errorhandler(500)
def internal_error(error):
    app.logger.exception("Internal Server Error")
    return jsonify({
        'error': 'Internal Server Error',
        'message': 'An unexpected error occurred',
        'status_code': 500
    }), 500

#  on  under 文processing器
@app.context_processor
def inject_global_vars():
    return {
        'app_name': 'Flask Middleware Demo',
        'current_time': time.time()
    }

# 模板filter器
@app.template_filter('format_datetime')
def format_datetime(timestamp):
    return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(timestamp))

# routing
@app.route('/')
def home():
    return "Hello, Flask! This is the home page."

@app.route('/api/data')
def api_data():
    return jsonify({
        'message': 'This is a JSON response',
        'timestamp': time.time(),
        'data': [1, 2, 3, 4, 5]
    })

@app.route('/api/users/')
def api_user(user_id):
    return jsonify({
        'id': user_id,
        'name': f"User {user_id}",
        'email': f"user{user_id}@example.com"
    })

# 自定义in间件class
class CustomMiddleware:
    def __init__(self, app):
        self.app = app
    
    def __call__(self, environ, start_response):
        app.logger.info("Custom middleware: Before request")
        response = self.app(environ, start_response)
        app.logger.info("Custom middleware: After request")
        return response

# application自定义in间件
app.wsgi_app = CustomMiddleware(app.wsgi_app)

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

练习题

  1. creation一个Flaskapplication, usingbefore_requesthookfunctionimplementation simple authenticationfunctions
  2. usingafter_requesthookfunction for 所 has response添加security头
  3. implementation自定义 404 and 500errorprocessing页面
  4. creation一个 on under 文processing器, 向所 has 模板注入当 before userinformation
  5. creation一个自定义in间件, 记录所 has request and response 详细information
  6. usingtemplate_filtercreation一个自定义filter器, 用于format日期时间
on 一页: Flask蓝graph and module化Development under 一页: Flask REST APIDevelopment

Flasktutorial导航

相关resource