六角レンチです
作ってるプログラムで非同期処理を使おうと思ってちょっと沼に入りかけてるので気晴らしに記事を書こうと思います

環境

linux 6.6.9
python 3.10.13

最近ノーパソ買ってlinux入れて使ってます
windowsよりすごい軽くて感動した

同期的に動いちゃうプログラム

例として下のようなプログラムを書きます

import asyncio
import time
import timeit

# IOバウンド(あるいは同期的な)処理
def blocking(t):
    time.sleep(t)
    print("sleeped", t)

async def main():
    print("main start")
    # ここの部分を非同期的に並行に処理したい
    # 〜〜〜〜〜〜〜〜
    for i in range(3):
        blocking(i)
    # 〜〜〜〜〜〜〜〜
    print("main finish")

# 時間の計測
print(timeit.timeit("asyncio.run(main())", number=1, globals=globals()))

このプログラムの

    for i in range(3):
        blocking(i)

この部分を非同期的に処理します(というかコメントとして書いてあるけど)

実行結果

main start
sleeped 0
sleeped 1
sleeped 2
main finish
3.009273783005483

一回一回実行して待ってるので3秒かかってます

イベントループを持ってきて実行する方法

多分これが一般的な方だと思います

import asyncio
import time
import timeit

# IOバウンド(あるいは同期的な)処理
def blocking(t):
    time.sleep(t)
    print("sleeped", t)

async def main():
    print("main start")
    # ここの部分を非同期的に並行に処理したい
    # 〜〜〜〜〜〜〜〜
    loop = asyncio.get_running_loop()
    await asyncio.gather(*(loop.run_in_executor(None, blocking, i) for i in range(3)))
    # 〜〜〜〜〜〜〜〜
    print("main finish")

# 時間の計測
print(timeit.timeit("asyncio.run(main())", number=1, globals=globals()))

asyncio.get_running_loop()で実行中のループを持ってきてrun_in_executorで走らせる感じ
run_in_executorの第一引数はNoneにすることでデフォルトのエクセキュータ(何もしない場合ThreadPoolExecutor)を使用します

asyncio.gatherはまとめて実行するために使ってます

実行結果

main start
sleeped 0
sleeped 1
sleeped 2
main finish
2.007849339999666

asyncio.to_threadを使う方法

python 3.9で追加されたto_threadを使う方法です
明確な違いはto_threadは高レベルAPIであるということです(get_running_loopは低レベルAPI)

import asyncio
import time
import timeit

# IOバウンド(あるいは同期的な)処理
def blocking(t):
    time.sleep(t)
    print("sleeped", t)

async def main():
    print("main start")
    # ここの部分を非同期的に並行に処理したい
    # 〜〜〜〜〜〜〜〜
    await asyncio.gather(*(asyncio.to_thread(blocking, i) for i in range(3)))
    # 〜〜〜〜〜〜〜〜
    print("main finish")

# 時間の計測
print(timeit.timeit("asyncio.run(main())", number=1, globals=globals()))

一行でかけるのでスッキリします
ただしエクセキュータは指定できません
（実際はできますが...¹）

実行結果

main start
sleeped 0
sleeped 1
sleeped 2
main finish
2.0057736759990803

それぞれのメリットとデメリット

イベントループを持ってきて実行する方法

メリット

エクセキュータを指定できる

デメリット

変数が一個必要

asyncio.to_threadを使う方法

メリット

一行で済む(変数を作らない)

デメリット

エクセキュータを指定できない¹

まとめ

エクセキュータを指定したい場合(CPUバウンドな処理とか)じゃなければto_threadでいいと思います

参考元

おまけ

to_threadの中身を見てみましょう

async def to_thread(func, /, *args, **kwargs):
    #〜〜〜〜〜〜〜〜〜〜〜
    #〜 to_threadの説明　〜
    #〜〜〜〜〜〜〜〜〜〜〜
    loop = events.get_running_loop()
    ctx = contextvars.copy_context()
    func_call = functools.partial(ctx.run, func, *args, **kwargs)
    return await loop.run_in_executor(None, func_call)

つまりto_threadはループを持ってきて実行するやつの糖衣構文です

おまけより、よく見てみるとrun_in_executorの第一引数がNoneなのでget_running_loopでループを取得した後にset_default_executorでデフォルト値を書き込めばエクセキュータを指定できることがわかります。
ただ、ループを持ってきている時点でrun_in_executorを使えばいいしset_default_executorに使えるエクセキュータはThreadPoolExecutor(CPUバウンドな処理とかに向いてない方)でなければならないのであまり使いみちはなさそう ↩ ↩²