跳到主要内容

错误处理

错误处理是 Web 应用开发中至关重要的环节。无论代码多么完美，生产环境中总会遇到各种意外情况：数据库连接失败、外部服务不可用、用户输入错误等。良好的错误处理机制不仅能提升用户体验，还能帮助开发者快速定位和修复问题。

理解 Flask 的错误处理机制​

HTTP 状态码​

HTTP 协议定义了一系列状态码来表示请求的处理结果：

状态码范围类别常见状态码
2xx成功200 OK, 201 Created
3xx重定向301 Moved Permanently, 302 Found
4xx客户端错误400 Bad Request, 401 Unauthorized, 403 Forbidden, 404 Not Found
5xx服务器错误500 Internal Server Error, 502 Bad Gateway, 503 Service Unavailable

在 Flask 中，当发生错误时，会自动返回相应的 HTTP 状态码。例如：

  • 访问不存在的路由返回 404
  • 未处理的异常返回 500
  • 请求方法不允许返回 405

Werkzeug 异常​

Flask 使用 Werkzeug 库提供的 HTTP 异常类。这些异常类可以直接抛出或使用 abort() 函数：

from flask import Flask, abort
from werkzeug.exceptions import BadRequest, NotFound, Forbidden

app = Flask(__name__)

# 方式一：直接抛出异常
@app.route('/admin')
def admin():
    if not is_admin():
        raise Forbidden()  # 返回 403
    return 'Admin Panel'

# 方式二：使用 abort 函数
@app.route('/user/<int:id>')
def get_user(id):
    user = User.query.get(id)
    if user is None:
        abort(404)  # 返回 404
    return render_template('user.html', user=user)

# 方式三：abort 带描述
@app.route('/resource/<int:id>')
def get_resource(id):
    resource = find_resource(id)
    if resource is None:
        abort(404, description='资源不存在')
    return jsonify(resource.to_dict())

常用的 Werkzeug 异常类：

异常类状态码说明
BadRequest400请求格式错误
Unauthorized401未认证
Forbidden403禁止访问
NotFound404资源不存在
MethodNotAllowed405方法不允许
NotAcceptable406无法接受
RequestTimeout408请求超时
Conflict409冲突
Gone410资源已删除
PayloadTooLarge413请求体过大
UnprocessableEntity422无法处理
TooManyRequests429请求过多
InternalServerError500服务器内部错误
NotImplemented501未实现
BadGateway502网关错误
ServiceUnavailable503服务不可用

注册错误处理器​

错误处理器是一个函数，当特定类型的错误发生时被调用，用于生成错误响应。

使用装饰器注册​

from flask import Flask, render_template, jsonify

app = Flask(__name__)

@app.errorhandler(404)
def page_not_found(error):
    """处理 404 错误"""
    return render_template('errors/404.html'), 404

@app.errorhandler(500)
def internal_server_error(error):
    """处理 500 错误"""
    return render_template('errors/500.html'), 500

@app.errorhandler(403)
def forbidden(error):
    """处理 403 错误"""
    return render_template('errors/403.html'), 403

@app.errorhandler(400)
def bad_request(error):
    """处理 400 错误"""
    return render_template('errors/400.html'), 400

使用函数注册​

在应用工厂模式中，使用 register_error_handler 函数：

from flask import Flask, render_template

def page_not_found(error):
    return render_template('errors/404.html'), 404

def internal_server_error(error):
    return render_template('errors/500.html'), 500

def create_app():
    app = Flask(__name__)
    
    # 注册错误处理器
    app.register_error_handler(404, page_not_found)
    app.register_error_handler(500, internal_server_error)
    
    return app

注册异常类处理器​

除了 HTTP 状态码，还可以注册特定异常类的处理器：

from werkzeug.exceptions import HTTPException
from sqlalchemy.exc import SQLAlchemyError

# 处理所有 HTTP 异常
@app.errorhandler(HTTPException)
def handle_http_exception(error):
    """将所有 HTTP 错误转换为 JSON 响应"""
    return jsonify({
        'code': error.code,
        'name': error.name,
        'description': error.description
    }), error.code

# 处理数据库错误
@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
    """处理数据库错误"""
    app.logger.error(f'Database error: {error}')
    return jsonify({
        'error': 'Database error',
        'message': 'An error occurred while accessing the database'
    }), 500

自定义错误页面​

基本错误页面​

创建友好的错误页面模板：

<!-- templates/errors/404.html -->
{% extends "base.html" %}

{% block title %}页面未找到 - {{ super() }}{% endblock %}

{% block content %}
<div class="error-page">
    <h1>404</h1>
    <h2>页面未找到</h2>
    <p>抱歉，您访问的页面不存在。</p>
    <p>可能的原因：</p>
    <ul>
        <li>页面已被删除或移动</li>
        <li>网址输入错误</li>
        <li>链接已过期</li>
    </ul>
    <div class="actions">
        <a href="{{ url_for('main.index') }}" class="btn btn-primary">返回首页</a>
        <a href="javascript:history.back()" class="btn btn-secondary">返回上一页</a>
    </div>
</div>
{% endblock %}
<!-- templates/errors/500.html -->
{% extends "base.html" %}

{% block title %}服务器错误 - {{ super() }}{% endblock %}

{% block content %}
<div class="error-page">
    <h1>500</h1>
    <h2>服务器内部错误</h2>
    <p>抱歉，服务器遇到了一个问题，无法完成您的请求。</p>
    <p>我们已经记录了这个错误，将尽快修复。</p>
    <div class="actions">
        <a href="{{ url_for('main.index') }}" class="btn btn-primary">返回首页</a>
        <a href="mailto:[email protected]" class="btn btn-secondary">联系支持</a>
    </div>
</div>
{% endblock %}

通用错误处理器​

创建一个通用的错误处理器来统一处理所有异常：

from flask import render_template, request, jsonify
from werkzeug.exceptions import HTTPException

@app.errorhandler(Exception)
def handle_exception(error):
    """处理所有未捕获的异常"""
    # 记录错误日志
    app.logger.error(f'Unhandled exception: {error}', exc_info=True)
    
    # 如果是 HTTP 异常，返回相应的错误页面
    if isinstance(error, HTTPException):
        # API 请求返回 JSON
        if request.path.startswith('/api/'):
            return jsonify({
                'code': error.code,
                'name': error.name,
                'description': error.description
            }), error.code
        
        # 普通请求返回 HTML 页面
        return render_template('errors/error.html', error=error), error.code
    
    # 非 HTTP 异常，返回 500
    if request.path.startswith('/api/'):
        return jsonify({
            'error': 'Internal Server Error',
            'message': 'An unexpected error occurred'
        }), 500
    
    return render_template('errors/500.html'), 500

API 错误处理​

当构建 RESTful API 时，应该返回 JSON 格式的错误响应，而不是 HTML 页面。

API 错误处理器​

from flask import jsonify, request
from werkzeug.exceptions import HTTPException

@app.errorhandler(HTTPException)
def handle_api_error(error):
    """API 错误处理器"""
    if request.path.startswith('/api/'):
        return jsonify({
            'success': False,
            'error': {
                'code': error.code,
                'name': error.name,
                'message': error.description
            }
        }), error.code
    
    # 非 API 请求返回 HTML
    return render_template('errors/error.html', error=error), error.code

自定义 API 异常​

创建自定义异常类可以更好地控制 API 错误响应：

from flask import jsonify, request

class APIError(Exception):
    """API 错误基类"""
    status_code = 400
    
    def __init__(self, message, status_code=None, payload=None):
        super().__init__()
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload
    
    def to_dict(self):
        result = dict(self.payload or {})
        result['message'] = self.message
        return result

class ValidationError(APIError):
    """验证错误"""
    status_code = 422

class AuthenticationError(APIError):
    """认证错误"""
    status_code = 401

class AuthorizationError(APIError):
    """授权错误"""
    status_code = 403

class NotFoundError(APIError):
    """资源不存在"""
    status_code = 404

class ConflictError(APIError):
    """资源冲突"""
    status_code = 409

class RateLimitError(APIError):
    """请求过多"""
    status_code = 429

# 注册 API 错误处理器
@app.errorhandler(APIError)
def handle_api_error(error):
    """处理自定义 API 错误"""
    response = jsonify({
        'success': False,
        'error': error.to_dict()
    })
    response.status_code = error.status_code
    return response

使用自定义异常：

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    user = User.query.get(user_id)
    if user is None:
        raise NotFoundError('用户不存在')
    return jsonify(user.to_dict())

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    
    # 验证输入
    if not data.get('username'):
        raise ValidationError('用户名不能为空')
    
    if not data.get('email'):
        raise ValidationError('邮箱不能为空')
    
    # 检查是否已存在
    if User.query.filter_by(username=data['username']).first():
        raise ConflictError('用户名已存在', payload={'field': 'username'})
    
    user = User(**data)
    db.session.add(user)
    db.session.commit()
    
    return jsonify(user.to_dict()), 201

@app.route('/api/admin/users')
@admin_required
def list_all_users():
    """管理员接口"""
    if not current_user.is_admin:
        raise AuthorizationError('需要管理员权限')
    
    users = User.query.all()
    return jsonify([u.to_dict() for u in users])

蓝图中的错误处理​

蓝图可以定义自己的错误处理器，但有一些限制需要注意。

蓝图错误处理器​

from flask import Blueprint, render_template, request, jsonify

api_bp = Blueprint('api', __name__, url_prefix='/api')

@api_bp.errorhandler(APIError)
def handle_api_error(error):
    """API 蓝图的错误处理器"""
    return jsonify({
        'success': False,
        'error': error.to_dict()
    }), error.status_code

@api_bp.errorhandler(404)
def api_not_found(error):
    """API 404 处理"""
    return jsonify({
        'success': False,
        'error': {
            'code': 404,
            'message': 'API endpoint not found'
        }
    }), 404

404 和 405 的特殊处理​

蓝图无法捕获 404 和 405 路由错误，因为这些错误发生在路由匹配阶段，此时还不知道应该使用哪个蓝图。需要在应用级别处理：

@app.errorhandler(404)
def page_not_found(error):
    """根据 URL 前缀返回不同的错误页面"""
    if request.path.startswith('/api/'):
        return jsonify({
            'success': False,
            'error': {
                'code': 404,
                'message': 'API endpoint not found'
            }
        }), 404
    
    if request.path.startswith('/admin/'):
        return render_template('admin/404.html'), 404
    
    return render_template('errors/404.html'), 404

@app.errorhandler(405)
def method_not_allowed(error):
    """处理方法不允许错误"""
    if request.path.startswith('/api/'):
        return jsonify({
            'success': False,
            'error': {
                'code': 405,
                'message': 'Method not allowed'
            }
        }), 405
    
    return render_template('errors/405.html'), 405

日志记录​

良好的日志记录是错误处理的重要组成部分，它可以帮助你发现问题、调试代码和监控应用健康状况。

Flask 日志系统​

Flask 使用 Python 标准库的 logging 模块：

from flask import Flask

app = Flask(__name__)

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    user = User.query.filter_by(username=username).first()
    
    if user and user.check_password(request.form.get('password')):
        login_user(user)
        app.logger.info(f'User {username} logged in successfully')
        return redirect(url_for('index'))
    else:
        app.logger.warning(f'Failed login attempt for user {username}')
        return render_template('login.html', error='Invalid credentials')

配置日志​

在应用启动时配置日志：

import logging
from logging.config import dictConfig
from flask import Flask

# 配置日志
dictConfig({
    'version': 1,
    'formatters': {
        'default': {
            'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s',
        },
        'detailed': {
            'format': '[%(asctime)s] %(levelname)s in %(module)s [%(pathname)s:%(lineno)d]: %(message)s',
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stdout',
            'formatter': 'default',
            'level': 'INFO'
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'filename': 'logs/app.log',
            'maxBytes': 1024 * 1024 * 10,  # 10MB
            'backupCount': 10,
            'formatter': 'detailed',
            'level': 'DEBUG'
        }
    },
    'root': {
        'level': 'INFO',
        'handlers': ['console', 'file']
    }
})

app = Flask(__name__)

请求信息日志​

在日志中包含请求信息有助于调试：

import logging
from flask import request, has_request_context

class RequestFormatter(logging.Formatter):
    """自定义日志格式化器，包含请求信息"""
    
    def format(self, record):
        if has_request_context():
            record.url = request.url
            record.method = request.method
            record.remote_addr = request.remote_addr
            record.user_id = getattr(current_user, 'id', None) if 'current_user' in globals() else None
        else:
            record.url = None
            record.method = None
            record.remote_addr = None
            record.user_id = None
        
        return super().format(record)

# 配置格式化器
formatter = RequestFormatter(
    '[%(asctime)s] %(remote_addr)s - %(method)s %(url)s - %(levelname)s: %(message)s'
)

# 应用到处理器
for handler in app.logger.handlers:
    handler.setFormatter(formatter)

邮件错误通知​

对于生产环境，可以配置邮件通知：

import logging
from logging.handlers import SMTPHandler

# 邮件处理器
mail_handler = SMTPHandler(
    mailhost='smtp.example.com',
    fromaddr='[email protected]',
    toaddrs=['[email protected]', '[email protected]'],
    subject='Application Error'
)
mail_handler.setLevel(logging.ERROR)
mail_handler.setFormatter(logging.Formatter(
    '''
Time: %(asctime)s
Level: %(levelname)s
URL: %(url)s
Method: %(method)s
IP: %(remote_addr)s
User: %(user_id)s

Message: %(message)s
'''
))

# 仅在生产环境添加
if not app.debug:
    app.logger.addHandler(mail_handler)

使用 Sentry​

Sentry 是一个专业的错误跟踪平台，可以自动捕获、聚合和通知应用错误：

import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration

# 初始化 Sentry
sentry_sdk.init(
    dsn='https://[email protected]/project-id',
    integrations=[
        FlaskIntegration(),
        SqlalchemyIntegration()
    ],
    traces_sample_rate=1.0,  # 性能监控采样率
    environment='production',
    release='1.0.0'
)

# 可以在请求中添加用户信息
@app.before_request
def set_sentry_user():
    if current_user.is_authenticated:
        sentry_sdk.set_user({
            'id': current_user.id,
            'username': current_user.username,
            'email': current_user.email
        })

调试技巧​

开发环境调试​

在开发环境中，Flask 提供了交互式调试器：

app.run(debug=True)

调试器会在发生错误时显示交互式页面，允许你在浏览器中查看变量和执行代码。

安全警告：永远不要在生产环境中启用调试模式！调试器允许执行任意代码，存在严重的安全风险。

生产环境调试​

在生产环境中，应该：

  1. 记录详细的错误日志
  2. 使用错误跟踪服务（如 Sentry）
  3. 为错误页面提供错误 ID，方便用户报告
import uuid

@app.errorhandler(500)
def internal_error(error):
    error_id = str(uuid.uuid4())[:8]
    app.logger.error(f'Error {error_id}: {error}', exc_info=True)
    
    return render_template('errors/500.html', error_id=error_id), 500
<!-- templates/errors/500.html -->
<div class="error-page">
    <h1>服务器错误</h1>
    <p>错误 ID: {{ error_id }}</p>
    <p>请将此 ID 提供给技术支持</p>
</div>

最佳实践​

1. 区分错误类型​

# 客户端错误 - 用户可修正
@app.errorhandler(400)
def bad_request(error):
    return jsonify({'error': 'Bad request', 'message': str(error)}), 400

# 认证错误 - 需要登录
@app.errorhandler(401)
def unauthorized(error):
    return jsonify({'error': 'Unauthorized', 'message': 'Please log in'}), 401

# 权限错误 - 禁止访问
@app.errorhandler(403)
def forbidden(error):
    return jsonify({'error': 'Forbidden', 'message': 'You do not have permission'}), 403

# 资源不存在
@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Not found', 'message': 'Resource not found'}), 404

# 服务器错误 - 记录日志
@app.errorhandler(500)
def internal_error(error):
    app.logger.error(f'Server error: {error}', exc_info=True)
    return jsonify({'error': 'Internal server error'}), 500

2. 一致的错误格式​

def error_response(status_code, message, details=None):
    """统一的错误响应格式"""
    response = {
        'success': False,
        'error': {
            'code': status_code,
            'message': message
        }
    }
    if details:
        response['error']['details'] = details
    return jsonify(response), status_code

@app.errorhandler(400)
def bad_request(error):
    return error_response(400, 'Bad request', str(error))

@app.errorhandler(422)
def unprocessable_entity(error):
    return error_response(422, 'Validation error', error.data.get('messages'))

3. 错误恢复​

from contextlib import contextmanager

@contextmanager
def db_session():
    """数据库会话上下文管理器"""
    try:
        yield db.session
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        app.logger.error(f'Database error: {e}')
        raise APIError('Database operation failed', status_code=500)

@app.route('/api/users', methods=['POST'])
def create_user():
    with db_session():
        user = User(**request.get_json())
        db.session.add(user)
    return jsonify(user.to_dict()), 201

4. 防止敏感信息泄露​

@app.errorhandler(Exception)
def handle_exception(error):
    """确保错误响应不泄露敏感信息"""
    # 记录完整错误（内部）
    app.logger.error(f'Unhandled exception: {error}', exc_info=True)
    
    # 返回通用错误消息（外部）
    if app.debug:
        # 开发环境显示详细信息
        return jsonify({
            'error': str(error),
            'type': type(error).__name__
        }), 500
    
    # 生产环境隐藏细节
    return jsonify({
        'error': 'An unexpected error occurred'
    }), 500

小结​

本章介绍了 Flask 的错误处理机制：

  1. HTTP 状态码：理解不同状态码的含义和用途
  2. 错误处理器：使用装饰器或函数注册错误处理器
  3. 自定义错误页面：为用户提供友好的错误页面
  4. API 错误处理：返回结构化的 JSON 错误响应
  5. 日志记录：配置日志系统，记录错误信息
  6. 错误跟踪：使用 Sentry 等服务监控生产环境错误

良好的错误处理不仅能提升用户体验，还能帮助开发者快速定位和修复问题，是构建可靠 Web 应用的重要组成部分。

参考资料​