如何将flask上下文与concurrent.futures.ThreadPoolExecutor一起使用

Question

我正在尝试使多个请求异步并获取响应，我使用

concurrent.futures

来执行此操作，但在我的函数中使用来自

current_app

的

flask

，我总是收到此错误：

RuntimeError: Working outside of application context.

我不知道如何解决这个问题。有人可以帮忙吗？

以下是我的代码：

运行.py：

import concurrent.futures
from flask import current_app
from http_calls import get_price, get_items

def init():
    with current_app._get_current_object().test_request_context():
        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
            futs = []
            futs.append(executor.submit(get_price))
            futs.append(executor.submit(get_items))

            print([fut.result() for fut in concurrent.futures.as_completed(futs)])

init()

http_calls.py

from flask import current_app

def get_price():
    url = current_app.config['get_price_url']
    return requests.get(url).json()

def get_items():
    url = current_app.config['get_items_url']
    return requests.get(url).json()

Answer 1

我在使用并发.futures 和 Flask 时遇到了类似的问题。我编写了 Flask-Executor 作为并发的 Flask 友好包装器。futures 来解决这个问题。对您来说，与这两者一起工作可能是一种更简单的方法。

Answer 2

您应该在脚本中导入 Flask 实例。在应用程序上下文下使用

current_app

。

import concurrent.futures
from your_application import your_app  # or create_app function to return a Flask instance
from flask import current_app
from http_calls import get_price, get_items

def init():
    with your_app.app_context():
        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
            ...

Answer 3

def with_app_context(cls, app, func):
    def wrapper(*args, **kwargs):
        with app.app_context():
            return func(*args, **kwargs)
    return wrapper

您可以使用包装器来实现这一点。您可以从线程池中调用您的函数，如下所示

with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
                # download using wkhtmltopdf (helps to address access denied issues)
                res = executor.submit(with_app_context(app, cls.some_func))