Flask应用中请求进入与第三方请求发送前的拦截处理问题

针对你需要在发送第三方请求前修改请求头和参数的需求，@app.after_request确实不适用（它是在Flask应用生成响应后触发），这里提供几种实用方案：

方案1：封装自定义请求函数 #

不用直接调用requests.get，自己封装一个函数统一处理请求头和参数：

import requests

def custom_requests_get(url, **kwargs):
    # 添加自定义请求头
    headers = kwargs.get('headers', {})
    headers['X-Custom-Header'] = 'your-custom-value'
    kwargs['headers'] = headers

    # 添加自定义URL参数
    params = kwargs.get('params', {})
    params['custom_param'] = 'custom-param-value'
    kwargs['params'] = params

    # 发送处理后的请求
    return requests.get(url, **kwargs)

之后在视图函数里替换原调用：

@app.route('/foo')
def get_foo():
    # 用封装的函数替代原生requests.get
    response = custom_requests_get('http://third-party-service/v1/api')
    # 后续响应处理逻辑...
    return 'processed result'

方案2：装饰器包装视图函数 #

如果仅需要针对特定视图处理第三方请求，可以写一个装饰器临时替换requests.get：

def intercept_third_party(func):
    def wrapper(*args, **kwargs):
        # 保存原生requests.get方法
        original_get = requests.get

        def wrapped_get(url, **kw):
            # 注入自定义请求头和参数
            headers = kw.get('headers', {})
            headers['X-Custom-Header'] = 'your-custom-value'
            kw['headers'] = headers

            params = kw.get('params', {})
            params['custom_param'] = 'custom-param-value'
            kw['params'] = params

            return original_get(url, **kw)

        # 临时替换requests.get
        requests.get = wrapped_get
        try:
            # 执行视图函数
            result = func(*args, **kwargs)
        finally:
            # 恢复原生方法，避免影响其他请求
            requests.get = original_get
        return result
    return wrapper

给目标视图添加装饰器即可：

@app.route('/foo')
@intercept_third_party
def get_foo():
    response = requests.get('http://third-party-service/v1/api')
    # 后续响应处理逻辑...
    return 'processed result'

方案3：全局猴子补丁（全应用统一处理） #

如果所有第三方GET请求都需要统一修改，可以在Flask应用启动时给requests.get打补丁：

import requests
from flask import Flask

app = Flask(__name__)

# 保存原生方法
original_get = requests.get

def patched_get(url, **kwargs):
    # 注入自定义请求头
    headers = kwargs.get('headers', {})
    headers['X-Custom-Header'] = 'your-custom-value'
    kwargs['headers'] = headers

    # 注入自定义参数
    params = kwargs.get('params', {})
    params['custom_param'] = 'custom-param-value'
    kwargs['params'] = params

    return original_get(url, **kwargs)

# 替换原生requests.get
requests.get = patched_get

@app.before_request
def before_request_handler():
    # 你已实现的请求进入视图前的处理逻辑
    pass

@app.route('/foo')
def get_foo():
    response = requests.get('http://third-party-service/v1/api')
    # 后续响应处理逻辑...
    return 'processed result'

内容的提问来源于stack exchange，提问作者xan