蓝本中创建的线程如何操作数据库？

HelloFlask Flask Web 开发实战
1 
emr_bp=Blueprint("emr",__name__,url_prefix='/emr')

def test():
    for i in range(30):
        time.sleep(1)
        print('start update')
    admin = User(
        username='admin',
        name='超级管理员'
    )
    admin.set_password('admin')
    db.session.add(admin)
    db.session.commit()
    flash('end....','info')

@emr_bp.route('/',methods=['GET', 'POST'])
def index():
    form=EmrSearchForm()
    if request.method == 'POST':
        t1 = threading.Thread(target=test)
        t1.start()
        flash(u'提交服务器运行，请稍等...','info')

    page=request.args.get('page',1,type=int)
    per_page=current_app.config['EMR_CHECK_RECODES_PER_PAGE']
    pagination=EmrCheckRecord.query.order_by(EmrCheckRecord.timestamp.desc()).paginate(page,per_page=per_page)
    records=pagination.items
    return render_template('emr/index.html',form=form,pagination=pagination,records=records)

以上代码在线程中插入一条数据到数据库，不成功。请问要怎么处理？谢谢!

2

你没有提供报错信息，我猜测的话是数据库操作需要在程序上下文环境下执行（比如需要从程序上下文获取数据库配置等信息），你需要把程序实例传递到后台线程，然后手动构造程序上下文：

def test(app):
    with app.app_context():
        # ...

@emr_bp.route('/',methods=['GET', 'POST'])
def index():
    # ...  
    app = current_app._get_current_object()
    t1 = threading.Thread(target=test, args=[app])
    t1.start()
    # ...

另外你在后台线程调用的 flash() 不会起作用，因为这个函数是把消息放到 Cookie 里，而你的后台线程处理逻辑不在请求处理流程内，Cookie 没法被正确放到响应里（这里仅猜测，未验证）。你可以在客户端用 AJAX 轮询的方式获取操作完成状态。

下次提问请务必包含完整的文字形式的错误回溯信息。

1 个赞
4

谢谢，按您的方法解决了。:grinning:

5

谢谢，按您的方法解决了
在线程中调用的flash()确实不会起作用。

6

我在app.py中使用manager(app)和manager.run()启动运行的时候，在app.py之外的其他py文件中定义了一个类，然后再类中的函数执行获取sqlalchemy查询的函数，也是拿不到上下文，如果引入app.py中的mananger对象会报错可能存在循环引用怎么办？