Flask2.0による非同期処理の実装方法

Question

Flask 2.0で非同期実装をしようと当サイトのコード(下記に抜粋)を参考に実装してみています。
非同期処理の実装に慣れていないというのもあり、色々ためしてはいますが、なぜ下記のようなエラーメッセージがでるのでしょうか？
また、非同期処理で４秒以内にすべての処理がおわるように実装したいと思っています。
もし、お気づきの点ありましたらご指摘いただけると助かります！

実行したコード

from flask import Blueprint
import asyncio
import time


app = Blueprint("app", __name__)

async def async_get_data(name, sec):
    print(f"start {name}")
    await asyncio.sleep(sec)
    print(f"end {name}")
    return f"{name}/{sec}"


@app.route("/data")
async def get_data():
    await async_get_data("A", 2)
    await async_get_data("B", 1)
    await async_get_data("C", 3)
    await async_get_data("D", 1)
    await async_get_data("E", 4)
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete()
    return result

エラーメッセージ

$ flask run
 * Serving Flask app 'app.py' (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
start A
end A
start B
end B
start C
end C
start D
end D
start E
end E
[xxx] ERROR in app: Exception on /data [GET]
...
...
...
    result = loop.run_until_complete()
TypeError: run_until_complete() missing 1 required positional argument: 'future'

参考にしたコード

import asyncio


async def async_get_data():
    await asyncio.sleep(1)
    return 'Done!'


@app.route("/data")
async def get_data():
    data = await async_get_data()
    return data

(参照URL)

Answer 1

下のようにしたらエラーはなくなるはずです。

from flask import Blueprint
import asyncio
import time


app = Blueprint("app", __name__)

async def async_get_data(name, sec):
    print(f"start {name}")
    await asyncio.sleep(sec)
    print(f"end {name}")
    return f"{name}/{sec}"


@app.route("/data")
async def get_data():
    await async_get_data("A", 2)
    await async_get_data("B", 1)
    await async_get_data("C", 3)
    await async_get_data("D", 1)
    await async_get_data("E", 4)
    return "SUCCESS!!"

・TypeError: run_until_complete() missing 1 required positional argument: 'future'が出る直
接の原因は、エラー内容の通り、run_until_complete()関数に引数がないからです。

・しかし、下記のように引数を指定しても、今度は
RuntimeError: This event loop is already running
というエラーが出ます。

(略）
async def foo():
   await asyncio.sleep(3)
   return True

@app.route("/data")
async def get_data():
    await async_get_data("A", 2)
    await async_get_data("B", 1)
    await async_get_data("C", 3)
    await async_get_data("D", 1)
    await async_get_data("E", 4)
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(foo())   # エラー
    return result

実行結果：
Traceback (most recent call last):
(略）
  File "***\lib\asyncio\base_events.py", line ***, in _check_running
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running

これは、Flask内部でイベントループが既に駆動している状態で、重ねて駆動させようとしたからです。

get_data()コルーチンはFlaskによって自動的にイベントループに割り当てられて駆動されるため、その内部でloop.run_until_completeを呼び出すことはできません。

---

追記

（編集後の追加質問について）

また、非同期処理で４秒以内にすべての処理がおわるように実装したいと思っています。

いくつかやり方があります。
下記のコードを実行し、/data1 /data2 /data3 と異なるURLにアクセスして挙動を見てみてください。

/data1 と/data2　は同じ挙動（data2はdata1を短縮した書き方）ですが、
/data3は結果を表示するタイミングが/data1、/data2と異なります。

いずれも4秒以内に処理が完了しています。

from flask import Blueprint, Flask
import asyncio
import time

app = Blueprint("app", __name__)

async def async_get_data(name, sec):
    print(f"start {name}")
    await asyncio.sleep(sec)
    print(f"end {name}")
    return f"{name}/{sec}"

@app.route("/data1")
async def get_data1():
    results = await asyncio.gather(        
       async_get_data("A", 2),
       async_get_data("B", 1),
       async_get_data("C", 3),
       async_get_data("D", 1),
       async_get_data("E", 4)
    )
    print(results)
    return "data1 SUCCESS!"

@app.route("/data2")
async def get_data2():
    # get_data1を短縮したもの。
    args = [("A", 2),("B", 1),("C", 3),("D", 1),("E", 4)]
    results = await asyncio.gather(*[async_get_data(*arg) for arg in args])
    print(results)
    return "data2 SUCCESS!"


@app.route("/data3")
async def get_data3():
    # コルーチンが完了するごとに結果を取得して表示。
    coros =  [     
       async_get_data("A", 2),
       async_get_data("B", 1),
       async_get_data("C", 3),
       async_get_data("D", 1),
       async_get_data("E", 4)
    ]
    # 注）python 3.7より前のバージョンだとcreate_taskが使えません。その場合は代わりに「ensure_future」を使います。
    for data in asyncio.as_completed([asyncio.create_task(coro) for coro in coros]):
       res = await data
       print(res)
    return "data3 SUCCESS!"

Comments

[sequelanonymous]

ありがとうございます！qnoirさんの書き方だ非同期実行されず、処理実行時間が11秒かかってしまいます。
(ただ、順番に実行されて終わるまでまつという実装になっているため)
本来なら非同期処理がされ、`await async_get_data("E", 4)`の4秒以内ですべての`async_get_data`
の処理が終わるように実装したいと思っています。

[退会済みユーザー]

後半にコードを追加しました。

[sequelanonymous]

ありがとうございます！とてもたすかります。

以前にこういう書き方をして試行錯誤していて成功していませんでした。
このコードをFlaskのルーティングを付与した関数内にコピーした途端に失敗するようになりました。
この原因は、Flask2.0でコルーチン関数内だとFlaskが自動で`loop.run_until_complete`を実行してくれるため、すでにループがまわっているというエラーがでる、という理解であっていますでしょうか？

```python
if __name__ == "__main__":
loop = asyncio.get_event_loop()
result = loop.run_until_complete(
asyncio.gather(
async_get_data("A", 2),
async_get_data("B", 1),
async_get_data("C", 3),
async_get_data("D", 1),
async_get_data("E", 4),
)
)
print(result)
```

[退会済みユーザー]

その理解であっていると思います。