Flask API处理大CSV文件返回空响应问题求助

核心问题原因 #

1. Flask配置无效：app.config['TIMEOUT']并非控制请求处理时长的参数，Flask本身不直接管理请求超时，超时由底层WSGI服务器（如默认的Werkzeug）或客户端决定。
2. 客户端主动断开：Postman、浏览器等客户端默认超时远短于10-12分钟，会主动终止连接，导致“空响应”或“socket hang up”，但后台进程不受影响仍会继续运行。
3. 异常处理逻辑错误：原代码中try-except块包裹了整个类方法的定义，无法捕获post方法执行过程中抛出的异常，正确的位置应该是在post方法内部。
4. 同步请求阻塞：长耗时任务会占用Flask工作进程，导致后续请求排队，同时无法及时向客户端反馈进度或结果。

具体解决方案 #

一、修复异常处理逻辑 #

先修正代码中的异常捕获问题，确保处理过程中的错误能被正确捕获并发送Slack告警：

app = Flask(__name__)
CORS(app)
api = Api(app, version='1.0', title='CATALOG-OPS-TECHNOLOGY', description='All catalog processing and images\' download / upload are handled by these APIs')
validate_scraped_file_ns = api.namespace('Validate-Scraped-File', description='Validate the raw scraped file')
validate_scraped_file_request_model = api.model('validateScrapedFileRequest', {
    'index': fields.String(required=True),
    'date_sc': fields.String(required=True),
})
validate_scraped_file_response_model = api.model('validateScrapedFileResponse', {
    'result': fields.Boolean(required=True),
    'validation_url': fields.String(required=False),
    'variation_url': fields.String(required=False),
})

@validate_scraped_file_ns.route('/validate_scraped_file')
class ValidationVariationResource(Resource):
    @validate_scraped_file_ns.expect(validate_scraped_file_request_model)
    @validate_scraped_file_ns.marshal_with(validate_scraped_file_response_model)
    def post(self):
        try:
            request_data = api.payload
            index = request_data['index']
            date_sc = request_data['date_sc']
            response = validate_scraped_file(index, date_sc)
            result = response['response']
            validation_url = response['validation_url']
            variation_url = response['variation_url']
            return {
                'result': result,
                'validation_url': validation_url,
                'variation_url': variation_url
            }
        except Exception as e:
            post_slack_message(str(e))
            validate_scraped_file_ns.abort(500, message=str(e))

二、异步任务处理（生产环境推荐） #

使用Celery实现后台异步处理，避免客户端长时间等待：

1. 安装依赖

pip install celery redis

2. 初始化Celery

from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    return celery

app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379/0',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)
celery = make_celery(app)

# 将耗时函数注册为Celery任务
@celery.task(bind=True)
def validate_scraped_file_task(self, index, date_sc):
    try:
        return validate_scraped_file(index, date_sc)
    except Exception as e:
        post_slack_message(str(e))
        raise self.retry(exc=e, countdown=60)

3. 修改API接口

• 提交任务接口：返回任务ID，客户端轮询状态

# 新增任务状态模型
task_status_model = api.model('TaskStatus', {
    'task_id': fields.String(required=True),
    'status': fields.String(required=True),
    'message': fields.String(required=False)
})

@validate_scraped_file_ns.route('/submit_validation')
class ValidationSubmitResource(Resource):
    @validate_scraped_file_ns.expect(validate_scraped_file_request_model)
    @validate_scraped_file_ns.marshal_with(task_status_model)
    def post(self):
        request_data = api.payload
        index = request_data['index']
        date_sc = request_data['date_sc']
        task = validate_scraped_file_task.delay(index, date_sc)
        return {'task_id': task.id, 'status': 'pending', 'message': '任务已提交，正在处理'}

• 任务状态查询接口：客户端轮询获取结果

@validate_scraped_file_ns.route('/check_task_status/<string:task_id>')
class TaskStatusResource(Resource):
    def get(self, task_id):
        task = validate_scraped_file_task.AsyncResult(task_id)
        if task.state == 'PENDING':
            response = {'task_id': task_id, 'status': task.state, 'message': '任务正在处理中'}
        elif task.state == 'SUCCESS':
            result = task.result
            response = {
                'task_id': task_id,
                'status': task.state,
                'result': result['response'],
                'validation_url': result['validation_url'],
                'variation_url': result['variation_url']
            }
        else:
            response = {'task_id': task_id, 'status': task.state, 'message': str(task.info)}
        return response

三、临时调整超时（仅用于测试，不推荐生产） #

如果必须用同步处理，需要调整WSGI服务器和客户端的超时设置：

• Werkzeug服务器：启动时指定超时参数

flask run --timeout 720  # 12分钟，单位秒

• Gunicorn服务器：启动时设置超时

gunicorn --timeout 720 --workers 4 app:app

• 客户端设置：
  • Postman：在设置中调整“请求超时”为720000毫秒（12分钟）
  • 前端Axios：设置timeout: 720000

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