flask异步任务

flask程序有三种状态

1.程序设置状态

当flask应用实例被创建后，应用处于程序设置状态，此时所有的全局对象都没有被绑定。就像下面第二行代码，app被创建，但是配置类还没加载，蓝图还没注册，数据库扩展以及其他的各种扩展也还没来得及初始化，此时应用对象是一个“干净”的app

def create_app(config_name=None):
    app = Flask('test_flask')
    if config_name is None:
        config_name = os.getenv('FLASK_CONFIG', 'development')
    app.config.from_object(config[config_name])
    app.register_blueprint(test_bp)
    db.init_app(app)
    async_task.init_app(app)
    return app

2.程序运行状态

当完成启动，但API还没被访问时，flask应用处于程序运行状态。此时程序上下文对象current_app和g都绑定了各自的对象，之后才可以正常使用。下图为程序运行状态的日志

E:\python_code\test_flask\.venv\Scripts\python.exe -m flask run
 * Serving Flask app "test_flask" (lazy loading)
 * Environment: development
 * Debug mode: on
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 108-954-734
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

3.请求运行状态

当API被访问，也就是我们写的视图函数被执行时，flask应用处于请求运行状态，请求上下文对象request和session会才会被绑定。而请求运行状态被激活时，程序运行状态必定会被激活，所以我们才可以在视图中正常使用程序上下文和请求上下文

无上下文的异常

上下文其实很好理解，就像小时候写作文一样：一个句子只有放到它的上下文环境中，表达的意思才通顺。同样的道理，flask应用中某个函数也需要它的上下文环境。句子离开上下文可能语义不清，函数离开上下文环境就会报错，如下就是我碰到的一个无上下文环境报错的日志

 No application found. Either work inside a view function or push an application context. See http://flask-sqlalchemy.pocoo.org/contexts/.

场景

我需要在API视图中执行一段比较耗时的代码，最后将结果写入到数据库。为了不让API超时，就需要引入异步任务的机制：将API执行的任务封装出来，然后异步执行。

处理办法

在网上搜了一下，码友的方法大致分为两派：1.使用线程池，2.使用celery（实现代码很简单我就不写了）。两种方法其实都可以实现python异步任务，但是在flask有些场景需要上下文环境，异步出去后上下文就随着视图的return走了，再执行任务直接就被干掉，最后被add_done_callback捕获错误。

带上下文的异步任务的扩展

在flask项目合适的地方实例化扩展

from Flask_Threadpool import FlaskThreadPool
 
async_task = FlaskThreadPool()

在flask项目合适的地方初始化扩展实例

def create_app(config_name=None):
    app = Flask('test_flask')
    ...
    async_task.init_app(app)
    return app

在需要异步执行的方法上添加装饰器

def callback(future):
    e = future.exception()
    print("异步任务的错误：" + str(e))
 
 
@async_task.submit(callback)
def long_task():
    current_app.logger.info("任务启动...")
    ...
    current_app.logger.info("任务结束")
 
 
@test_bp.route('/starttask', methods=['GET', 'POST'])
def start_task():
    long_task()
    return "SUCCESS"
 
 

前面用到的Flask_Threadpool扩展就是下面这个

from flask import Flask
from concurrent.futures import ThreadPoolExecutor
 
 
class FlaskThreadPool:
    """
    1.在项目合适的地方实例化
        ft_pool = FlaskThreadPool()
    2.在项目合适的地方初始化
        ft_pool.init_app(app)
    3.装饰一个预期异步执行的任务函数
        @ft_pool.submit(callback)
        def func(args1, args2):
            ...
    """
    def __init__(self, app=None):
        if app is not None:
            self.init_app(app)
 
    def init_app(self, app: Flask):
        """ 完成初始化 """
        # 将扩展加入到扩展字典
        self.app = app
        if not hasattr(app, "extensions"):
            app.extensions = {}
        app.extensions["async_task"] = self
        size = app.config.setdefault('THREAD_POOL_SIZE', 4)
        self.executor = ThreadPoolExecutor(size)
 
    def submit(self, callback=None):
        """ 装饰器：将任务异步执行
        1.先将任务加入上下文环境
        2.将任务变成异步任务
        """
        def async_task(func):
            def inner(*args, **kwargs):
                self.executor.submit(func, *args, **kwargs).add_done_callback(callback)
            return inner
 
        def decorator(func):
            @async_task
            def inner(*args, **kwargs):
                with self.app.app_context():
                    func(*args, **kwargs)
            return inner
        return decorator

that's all

如果有更好的解决方案麻烦告诉我，谢谢~