Flask API 日志过滤：通过白名单机制优化请求日志管理

import re
from flask import Flask
from werkzeug import serving

# 假设您的Flask应用实例名为app
app = Flask(__name__)

def restrict_access_logs(app_instance):
    """
    修改WSGIRequestHandler的log_request方法，实现基于白名单的日志过滤。
    """
    # 保存原始的log_request方法
    parent_log_request = serving.WSGIRequestHandler.log_request

    # 动态获取所有已注册的端点名称
    # 注意：这里获取的是端点名称（endpoint），而不是完整的URL路径
    permitted_endpoints = [rule.endpoint for rule in app_instance.url_map.iter_rules()]

    def log_request(self, *args, **kwargs):
        """
        自定义的log_request方法，根据白名单判断是否记录日志。
        """
        # 检查请求路径是否匹配白名单中的任一端点
        # 假设所有API路径都以 /api/v1/ 开头，且端点名称与路径的最后一部分对应
        # 例如，如果端点是 'hello'，则匹配 '/api/v1/hello' 或 '/api/v1/hello/anything'
        is_whitelisted = False
        for endpoint in permitted_endpoints:
            # 排除Flask自带的'static'端点，通常不需要记录其日志
            if endpoint == 'static':
                continue

            # 构建正则表达式来匹配请求路径
            # 这里以 '/api/v1/' 作为前缀示例，请根据您的实际API路径结构调整
            # 确保正则表达式能正确匹配您的URL结构
            pattern = rf"/api/v1/{re.escape(endpoint)}(/.*)?$"
            if re.match(pattern, self.path):
                is_whitelisted = True
                break

        # 如果请求路径在白名单中，则调用原始的log_request方法记录日志
        if is_whitelisted:
            parent_log_request(self, *args, **kwargs)

    # 将WSGIRequestHandler的log_request方法替换为我们自定义的函数
    serving.WSGIRequestHandler.log_request = log_request

# 示例API路由
@app.route('/api/v1/hello', methods=['GET'])
def hello():
    return "Hello, Flask!"

@app.route('/api/v1/getEvidencesByProductID/<int:product_id>', methods=['GET'])
def getEvidencesByProductID(product_id):
    return f"Fetching evidences for product ID: {product_id}"

@app.route('/api/v1/testpoint', methods=['GET'])
def testpoint():
    ep_list = [rule.endpoint for rule in app.url_map.iter_rules()]
    ep_str = ", ".join(ep_list)
    return f"Available Endpoints: {ep_str}"

@app.route('/api/v1/unlisted', methods=['GET'])
def unlisted_endpoint():
    return "This endpoint should not be logged."

if __name__ == '__main__':
    # ... (上面定义的 app 实例和路由) ...

    # 在所有路由定义完成后，调用日志限制函数
    restrict_access_logs(app) 

    # 运行Flask应用
    app.run(debug=True)

import re
from flask import Flask
from werkzeug import serving

app = Flask(__name__)

def restrict_access_logs(app_instance):
    """
    修改WSGIRequestHandler的log_request方法，实现基于白名单的日志过滤。
    """
    parent_log_request = serving.WSGIRequestHandler.log_request

    # 动态获取所有已注册的端点名称
    permitted_endpoints = [rule.endpoint for rule in app_instance.url_map.iter_rules()]
    print(f"Whitelisted Endpoints: {permitted_endpoints}") # 调试输出

    def log_request(self, *args, **kwargs):
        """
        自定义的log_request方法，根据白名单判断是否记录日志。
        """
        is_whitelisted = False
        for endpoint in permitted_endpoints:
            if endpoint == 'static': # 排除Flask自带的'static'端点
                continue

            # 根据您的API路径结构调整正则表达式
            # 例如，如果您的API前缀是/api/v1/
            pattern = rf"/api/v1/{re.escape(endpoint)}(/.*)?$"
            if re.match(pattern, self.path):
                is_whitelisted = True
                break

        if is_whitelisted:
            parent_log_request(self, *args, **kwargs)

    serving.WSGIRequestHandler.log_request = log_request

# 示例API路由定义
@app.route('/api/v1/hello', methods=['GET'])
def hello():
    return "Hello, Flask!"

@app.route('/api/v1/getEvidencesByProductID/<int:product_id>', methods=['GET'])
def getEvidencesByProductID(product_id):
    return f"Fetching evidences for product ID: {product_id}"

@app.route('/api/v1/testpoint', methods=['GET'])
def testpoint():
    ep_list = [rule.endpoint for rule in app.url_map.iter_rules()]
    ep_str = ", ".join(ep_list)
    return f"Available Endpoints: {ep_str}"

@app.route('/api/v1/unlisted', methods=['GET'])
def unlisted_endpoint():
    return "This endpoint should not be logged."

@app.route('/no-api-prefix', methods=['GET'])
def no_api_prefix():
    return "This endpoint has no /api/v1/ prefix."

if __name__ == '__main__':
    # 确保在所有路由定义之后调用此函数
    restrict_access_logs(app) 
    app.run(debug=True)