一、先搞懂为啥要做统一异常处理和日志优化

很多刚写Flask项目的朋友,一开始都是在每个接口里单独处理异常——比如写个用户登录接口,怕数据库连不上、怕密码输错、怕用户不存在,就给每个可能出错的地方套个try-except,出问题就直接return个json。但项目越写越大,接口越来越多,这种做法的问题就冒出来了:每个接口的错误返回格式都不一样,有的是{"code":500,"msg":"出错了"},有的是{"status":-1,"message":"登录失败"},前端对接的时候要适配各种格式,疯得很;而且出问题的时候,日志要么只写“出错了”,要么乱得找不到到底是哪行代码出的错、传的参数是什么,排查问题要花大半天。

还有生产环境的日志,很多人要么把所有日志都开着,硬盘很快被占满;要么只开错误日志,出问题的时候找不到上下文(比如不知道当时用户传的参数、请求的接口路径、关联的用户ID是什么),根本没法定位问题。所以统一封装异常处理、优化生产环境日志,是Flask项目从“能跑”到“好用、好维护”的必经一步。

二、Flask统一异常处理的完整实现

2.1 先明确技术栈

所有示例统一使用:Python 3.9 + Flask 2.2 + Werkzeug 2.2(Flask默认依赖,不用额外装),没有其他第三方库,保证大家直接复制就能跑。

2.2 第一步:定义统一的错误返回格式

首先要定一个所有接口都遵守的错误返回规则,比如我们规定:所有错误返回都包含三个字段——code(状态码,200是成功,400是请求参数错,401是未登录,403是没权限,500是服务器错)、msg(给前端看的提示语)、data(额外的错误信息,比如参数错误的具体字段,没有就给null)。

然后我们可以写一个统一的返回函数,不管是成功还是出错,都用这个函数返回:

# 统一返回函数,处理成功和错误的格式
def make_response(code: int, msg: str, data=None):
    # 把参数转成符合JSON规范的字典
    response_data = {
        "code": code,
        "msg": msg,
        "data": data
    }
    # 用Flask的jsonify转成JSON响应,同时设置状态码(比如400、500这些)
    return jsonify(response_data), code

2.3 第二步:捕获所有Flask默认的异常

Flask本身有一些默认的异常,比如404(找不到接口)、405(请求方法不对,比如用GET访问了只支持POST的接口)、500(服务器内部错误),我们要把这些异常都捕获,然后用我们的统一格式返回。

Flask提供了一个装饰器@errorhandler,可以用来捕获指定的异常,我们可以给它传HTTP状态码,比如@errorhandler(404)就会捕获404异常:

from flask import Flask, jsonify, request

# 初始化Flask应用
app = Flask(__name__)

# 捕获404异常:找不到接口
@app.errorhandler(404)
def handle_404(e):
    # e是Flask抛出的异常对象,里面有异常的描述
    return make_response(404, "接口不存在", str(e))

# 捕获405异常:请求方法不对
@app.errorhandler(405)
def handle_405(e):
    return make_response(405, "请求方法不允许", str(e))

# 捕获500异常:服务器内部错误
@app.errorhandler(500)
def handle_500(e):
    # 这里可以先把异常信息记录到日志,后面讲日志的时候会详细说
    app.logger.error(f"服务器内部错误:{str(e)}")
    return make_response(500, "服务器开小差了,请稍后再试", None)

2.4 第三步:捕获业务自定义的异常

除了Flask默认的异常,我们自己的业务逻辑也会有很多异常,比如“用户不存在”、“密码错误”、“参数格式不对”。如果每个业务异常都单独处理,还是会乱,所以我们可以定义一个业务异常类,所有业务相关的异常都继承这个类:

# 自定义业务异常类
class BusinessException(Exception):
    def __init__(self, code: int, msg: str):
        # 异常的状态码和提示语
        self.code = code
        self.msg = msg
        # 调用父类的构造函数,把异常信息传进去
        super().__init__(msg)

然后我们用@errorhandler捕获这个自定义的异常类:

# 捕获自定义的业务异常
@app.errorhandler(BusinessException)
def handle_business_exception(e):
    # 把异常信息记录到日志
    app.logger.error(f"业务异常:{e.msg},状态码:{e.code}")
    return make_response(e.code, e.msg, None)

这样以后我们写业务逻辑的时候,只要抛出BusinessException就行,比如:

# 示例接口:用户登录
@app.route('/login', methods=['POST'])
def login():
    # 从请求里获取用户名和密码
    username = request.json.get('username')
    password = request.json.get('password')
    
    # 校验参数:用户名或密码为空
    if not username or not password:
        # 抛出业务异常,状态码400,提示语“用户名或密码不能为空”
        raise BusinessException(400, "用户名或密码不能为空")
    
    # 模拟查数据库:假设用户不存在
    user = None  # 实际项目里这里是查数据库的逻辑
    if not user:
        raise BusinessException(404, "用户不存在")
    
    # 模拟密码错误
    if password != "123456":  # 实际项目里是校验密码哈希
        raise BusinessException(401, "密码错误")
    
    # 登录成功
    return make_response(200, "登录成功", {"user_id": 123})

2.5 第四步:捕获未被捕获的异常

有时候我们写代码会不小心漏掉异常,比如调用了一个函数,这个函数抛出了我们没捕获的异常,这时候Flask会返回默认的500错误,格式还是乱的。所以我们要加一个全局的异常捕获,捕获所有未被处理的异常:

# 捕获所有未被处理的异常
@app.errorhandler(Exception)
def handle_uncaught_exception(e):
    # 把异常信息记录到日志
    app.logger.error(f"未捕获的异常:{str(e)}")
    return make_response(500, "服务器开小差了,请稍后再试", None)

这里要注意,这个装饰器要放在所有其他errorhandler的后面,因为Flask会按装饰器的顺序匹配异常,先匹配到的异常会先处理。

三、生产环境日志优化的完整实现

3.1 先搞懂日志的核心需求

生产环境的日志要满足三个要求:第一,能定位问题(要知道什么时候、哪个接口、哪个用户、传了什么参数、出了什么错);第二,能节省空间(不要把没用的日志都存着);第三,能方便排查(日志格式要统一,方便用工具搜索)。

3.2 第二步:配置日志的级别

日志的级别从低到高是:DEBUG(调试用,开发的时候用)、INFO(普通信息,比如接口被调用)、WARNING(警告,比如参数格式不对但不影响正常运行)、ERROR(错误,比如业务异常)、CRITICAL(严重错误,比如数据库连不上)。

生产环境我们要把日志级别设为WARNING或者ERROR,这样只会记录有用的日志,不会存大量的DEBUG和INFO日志,节省空间。

3.3 第三步:配置日志的格式

我们要让每一条日志都包含以下信息:时间(精确到毫秒)、日志级别、日志内容、请求的接口路径、请求的方法、请求的参数、用户ID(如果是登录后的请求)、异常的堆栈信息(如果是错误日志)。

Flask的日志是基于Python的logging模块的,我们可以自己配置日志的格式:

import logging
from logging.handlers import RotatingFileHandler

# 配置日志
def setup_logging(app):
    # 日志级别:生产环境设为WARNING,开发环境设为DEBUG
    log_level = logging.WARNING if app.config.get('ENV') == 'production' else logging.DEBUG
    
    # 日志格式:定义每一条日志的内容
    log_format = logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s - 请求路径:%(path)s - 请求方法:%(method)s - 请求参数:%(params)s - 用户ID:%(user_id)s',
        datefmt='%Y-%m-%d %H:%M:%S.%f'  # 时间格式,精确到毫秒
    )
    
    # 日志处理器:把日志写到文件,并且按大小分割(防止单个日志文件太大)
    # 每个日志文件最大10MB,最多保留5个日志文件,旧的日志会被自动删除
    file_handler = RotatingFileHandler(
        'flask_app.log',  # 日志文件名
        maxBytes=10*1024*1024,  # 10MB
        backupCount=5,  # 最多保留5个日志
        encoding='utf-8'  # 编码,防止中文乱码
    )
    file_handler.setFormatter(log_format)
    file_handler.setLevel(log_level)
    
    # 控制台日志处理器:开发的时候可以在控制台看到日志,生产环境可以关掉
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(log_format)
    console_handler.setLevel(log_level)
    
    # 把处理器添加到Flask的日志对象
    app.logger.addHandler(file_handler)
    app.logger.addHandler(console_handler)
    app.logger.setLevel(log_level)
    
    # 给日志添加自定义的属性:请求路径、请求方法、请求参数、用户ID
    # 这里用Flask的before_request装饰器,在每个请求处理前给日志添加这些属性
    @app.before_request
    def add_log_attributes():
        # 获取请求路径
        request.path = request.path
        # 获取请求方法
        request.method = request.method
        # 获取请求参数:GET请求的参数在args里,POST请求的参数在json里
        if request.method == 'GET':
            request.params = str(request.args)
        else:
            request.params = str(request.json) if request.json else '{}'
        # 获取用户ID:假设登录后把用户ID存在g对象里(Flask的g对象是每个请求的全局对象)
        request.user_id = str(getattr(request, 'user_id', None))
        
        # 把这些属性添加到日志的extra里,这样日志格式里就能用了
        app.logger = logging.LoggerAdapter(app.logger, {
            'path': request.path,
            'method': request.method,
            'params': request.params,
            'user_id': request.user_id
        })

然后我们在初始化Flask应用的时候调用这个setup_logging函数:

app = Flask(__name__)
# 配置环境:生产环境设为'production',开发环境设为'development'
app.config['ENV'] = 'production'
# 配置日志
setup_logging(app)

3.4 第四步:配置异常的堆栈信息

出错误的时候,光有错误信息还不够,还要知道异常的堆栈信息(就是哪行代码出的错)。Python的logging模块有一个exc_info参数,把它设为True,就会自动记录异常的堆栈信息:

# 比如在捕获500异常的时候,记录日志
@app.errorhandler(500)
def handle_500(e):
    # exc_info=True表示记录异常的堆栈信息
    app.logger.error(f"服务器内部错误:{str(e)}", exc_info=True)
    return make_response(500, "服务器开小差了,请稍后再试", None)

# 捕获业务异常的时候,也可以记录堆栈信息(如果需要的话)
@app.errorhandler(BusinessException)
def handle_business_exception(e):
    app.logger.error(f"业务异常:{e.msg},状态码:{e.code}", exc_info=True)
    return make_response(e.code, e.msg, None)

3.5 第五步:日志的滚动和清理

我们用的RotatingFileHandler已经实现了日志的滚动和清理,它会在日志文件达到10MB的时候,把旧的日志重命名为flask_app.log.1、flask_app.log.2……,最多保留5个,旧的日志会被自动删除,这样就不会占用太多硬盘空间。

四、应用场景、优缺点、注意事项

4.1 应用场景

这种统一异常处理和日志优化的方案,适合所有需要上线的Flask项目,尤其是:

  • 团队协作的项目:统一的异常格式和日志格式,方便前端对接和后端排查问题;
  • 高访问量的项目:生产环境的日志优化可以节省空间,提高排查问题的效率;
  • 有严格错误处理要求的项目:比如金融、医疗类的项目,需要记录所有错误的详细信息,方便审计和排查。

4.2 技术优缺点

优点

  • 异常格式统一:前端不用适配各种错误格式,对接效率提高;
  • 日志信息全面:出问题的时候能快速定位到具体的代码、参数、用户;
  • 配置灵活:可以根据开发和生产环境切换日志级别,开发的时候能看到详细的调试信息,生产的时候能节省空间;
  • 扩展性强:如果需要添加新的异常类型或者日志字段,只要修改对应的代码就行,不用改每个接口。

缺点

  • 开发的时候可能会觉得麻烦:一开始要写很多配置代码,不能像之前那样随便写个try-except就完事;
  • 日志的大小还是会增加:如果访问量很大,即使只开WARNING级别,日志也会越来越多,需要定期备份和清理;
  • 对新手不友好:需要理解Flask的errorhandler、before_request、logging模块等概念,新手可能需要花时间学习。

4.3 注意事项

  • 不要把敏感信息写到日志里:比如用户的密码、银行卡号、身份证号等,日志里的敏感信息会有安全风险;
  • 不要把异常的详细信息返回给前端:比如500异常的时候,不要把堆栈信息返回给前端,不然黑客会利用这些信息攻击你的系统;
  • 日志的级别不要设得太低:生产环境不要设为DEBUG级别,不然会产生大量的日志,占用硬盘空间;
  • 日志文件的权限要设置好:不要让普通用户能修改或删除日志文件,防止日志被篡改;
  • 定期备份日志:重要的日志要定期备份到其他存储设备,防止硬盘损坏导致日志丢失。

五、文章总结

统一封装Flask项目的异常处理和优化生产环境的日志,是一个看似简单但很重要的工作。通过统一的异常格式,我们可以提高前端对接的效率;通过全面的日志信息,我们可以快速定位和解决问题;通过合理的日志配置,我们可以节省空间,提高系统的稳定性。

这个方案的核心是“统一”和“实用”,统一的规则让团队协作更顺畅,实用的配置让生产环境的维护更轻松。对于新手来说,可能一开始会觉得复杂,但只要跟着示例一步步实现,就能慢慢理解其中的原理,掌握这个技能后,对自己的项目开发能力会有很大的提升。