Flask路由同时返回JSON和写数据库响应延迟优化求助

核心思路是将数据库写入操作从请求响应的关键链路中剥离，无需等待写入完成即可向前端返回JSON，可完全消除0.1~0.4s的额外延迟，以下是不同复杂度的实现方案：

方案1：使用Flask内置after_this_request钩子（最轻量，无额外依赖） #

适合新手快速落地，无需安装第三方组件，利用Flask的请求后钩子，将写库逻辑放到响应已经发送给客户端之后执行：

from flask import after_this_request

@app.route('/api/data/<string:from_port>/<string:to_port>/')
def data(from_port,to_port):
    # 原有API调用、数据处理逻辑保持不变，先拿到要返回的api_results
    api_url = helpers.get_api_url(int(un_to_int_dict[from_port]),int(un_to_int_dict[to_port]), api_key=SECRETS.get("big_schedules_api_key"))
    df =  helpers.create_df(requests.get(url = api_url
                                             , proxies=SECRETS.get("proxies")
                                            ),scac_dict = scac_dict)
    
    bigitems = [helpers.create_big_item(df[df['Route ID']==i]) for i in df['Route ID'].unique()]
    api_results = {"data":[x.to_dict() for x in bigitems]}

    # 把写库逻辑注册到请求后钩子，响应发送后才会执行
    @after_this_request
    def save_to_db(response):
        origin_port = from_port
        destination_port = to_port
        current_date = date.today().strftime("%m/%d/%Y")
                           
        new_user_entry = user_data(ip_address = request.remote_addr,
                                   origin_port = origin_port,destination_port = destination_port,date = current_date)
                           
        db.session.add(new_user_entry)
        # 优化：无需全表查最大ID，flush后直接拿自增ID即可，效率提升明显
        db.session.flush()
        request_id = new_user_entry.id
                           
        new_sailing_entry = [cargosmart_sailings(
                         from_port = df['From Port'].iloc[i],to_port= df['To Port'].iloc[i],terminal = df['To Terminal'].iloc[i],
                         etd = convert_date(df['ETD'].iloc[i]),
                         eta = convert_date(df['ETA'].iloc[i]),
                         vessel_name = df['Vessel Name'].iloc[i],service= df['Service'].iloc[i],
                         external_voyage_number = df['External Voyage Number'].iloc[i],
                         direct = df['Direct'].iloc[i],carrier = df['Carrier'].iloc[i],
                         cutoff = convert_date(df['Cutoff'].iloc[i]),
                         transit_time = df['Transit Time'].iloc[i],
                         request_id = request_id
                         
                        ) for i in range(df.shape[0])]

        for x in new_sailing_entry:
            db.session.add(x)
        db.session.commit()
        return response
    
    # 直接返回结果，无需等写库完成
    return jsonify(api_results)

方案2：后台线程执行写库操作 #

如果担心请求上下文销毁带来的异常，可以开启独立守护线程执行写库逻辑：

import threading

@app.route('/api/data/<string:from_port>/<string:to_port>/')
def data(from_port,to_port):
    # 原有API调用、数据处理逻辑保持不变，先拿到api_results
    # ... 省略和上面一致的处理代码 ...
    api_results = {"data":[x.to_dict() for x in bigitems]}

    # 封装写库函数
    def save_to_db(df, remote_addr, from_port, to_port):
        origin_port = from_port
        destination_port = to_port
        current_date = date.today().strftime("%m/%d/%Y")
        new_user_entry = user_data(ip_address = remote_addr,
                                   origin_port = origin_port,destination_port = destination_port,date = current_date)
        db.session.add(new_user_entry)
        db.session.flush()
        request_id = new_user_entry.id
        # 省略后续写库逻辑，和上面一致
        # ...
        db.session.commit()
    
    # 启动守护线程执行写库
    threading.Thread(
        target=save_to_db,
        args=(df, request.remote_addr, from_port, to_port),
        daemon=True
    ).start()
    
    # 直接返回结果
    return jsonify(api_results)

方案3：异步任务队列（适合高流量场景） #

如果后续请求量较大，要求写库成功率高，可使用轻量异步任务队列RQ（Redis Queue），将写库任务丢到队列中异步执行，请求直接返回，相比Celery配置更简单，适合中小型项目。

额外写库性能优化点 #

不管使用上述哪个方案，都可以通过以下优化进一步降低写库耗时：

• 替换原有全表查最大ID的逻辑，db.session.add(new_user_entry)后调用db.session.flush()，直接通过new_user_entry.id获取自增主键，可减少大量查询耗时
• 替换循环add的ORM操作，使用SQLAlchemy Core批量插入语法，速度可提升3~10倍：

  db.session.execute(
      cargosmart_sailings.__table__.insert(),
      [
          # 每个条目转为字典格式
          {"from_port": df['From Port'].iloc[i], "to_port": df['To Port'].iloc[i], ...其余字段} 
          for i in range(df.shape[0])
      ]
  )
  

• 调整SQLite配置参数，执行db.session.execute("PRAGMA synchronous = OFF")，可大幅提升写入速度，适合非核心业务数据场景。

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