asyncを使用してflaskで裏で定期的に処理をしたいです。

実現したいこと

前提

pythonでpostリクエストを受け取り、裏で定期的に受け取ったデータを処理したいです。

発生している問題

  app.runのタスクを定義した後の処理がされません。

該当のソースコード

python
1from flask import Flask,request
2import asyncio
3
4
5
6queue = asyncio.Queue()
7
8
9app = Flask(__name__)
10
11    
12@app.route('/',methods=['POST'])
13async def hello():
14    
15    data = request.get_json()
16    await queue.put(data)
17    
18    return '200'
19async def main(): 
20    wstsk=asyncio.create_task(app.run(host='0.0.0.0', port=8000))
21    while True:
22       
23        while not queue.empty():
24            try:
25                item = await queue.get()
26                print(item)
27            finally:
28                queue.task_done()
29        await asyncio.sleep(100)
30            
31        
32
33
34
35if __name__ == '__main__':
36   
37    
38    loop = asyncio.get_event_loop()
39    
40    loop.run_until_complete(main())

試したこと

threddingを使用したところ、データを受け取りつつ処理することができましたが、どうしても使いたいasyncioの機能があり、
なるべくasyncioとthreddingは併用しない方が良いらしいので、asyncを使った処理を実現したいです。
また、app.run()と受け取った情報を処理する関数をどちらもタスクとしてみましたが、proccess_queue内のwhile文が一周目は実行され,"processing"が出力されましたが、app.runのタスクが処理を手放さないのか2周目以降は実行されませんでした。
以下がそのコードです。

python
1from flask import Flask,request
2import asyncio
3
4queue = asyncio.Queue()
5
6app = Flask(__name__)
7
8
9    
10@app.route('/',methods=['POST'])
11async def hello():
12    
13    data = request.get_json()
14    await queue.put(data)
15    
16    return '200'
17
18async def run_app():
19    await app.run(host='0.0.0.0', port=8000)
20    
21
22async def process_queue():
23    
24    while True:
25        
26        print('processing')
27        
28        await asyncio.sleep(2)
29   
30
31    
32async def main():
33    tasks = [
34        asyncio.create_task(process_queue()),
35        asyncio.create_task(run_app())
36
37    ]
38    await asyncio.gather(*tasks)
39if __name__ == '__main__':
40 
41    loop = asyncio.get_event_loop()
42   
43    loop.run_until_complete(main())
44    loop.close()
45

回答1件

ベストアンサー

勘違いしていました。Flaskのasyncは、中で別のevent loopを作るので、外のevent loopには処理は移らないのですね。
そういうことなら、Flaskとは別のスレッドでevent loopを走らせてやるなどの必要があると思います。

thread を使うのはどうしても避けたいということなら、ASGI サーバーとアダプタを使うという方法があるようです。
もしくは、async/await にしっかり対応したフレームワークに移行するのがいいのでは。Quart は Flask とほぼ同じように使えそうです。

元の回答

asyncio.gather() を await しないといけないのでは。

python
1async def main():
2    tasks = [
3        asyncio.create_task(process_queue()),
4        asyncio.create_task(run_app())
5    ]
6
7    await asyncio.gather(*tasks)