Pythonでasyncなexecを実現する方法

この記事は、Python Advent Calendar 2023の23日目です。

exec を使うべきじゃないといった話は一旦置いておくとして。

実現したいこと

引数で指定した文字列をPythonコードとして実行する exec 関数。
しかし、その中で await を使いたい！ と思っても、単純にはできません。

次のコードはエラーになります。

exec.py

import asyncio

async def run():
    print('start')
    await asyncio.sleep(3)
    print('end')

async def main():
    script: str = 'await run()'
    exec(script)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

awayume@assam:~/programming/playground$ python exec.py
/home/awayume/programming/playground/exec.py:13: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
Traceback (most recent call last):
  File "/home/awayume/programming/playground/exec.py", line 14, in <module>
    loop.run_until_complete(main())
  File "/usr/lib/python3.11/asyncio/base_events.py", line 650, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/awayume/programming/playground/exec.py", line 10, in main
    exec(script)
  File "<string>", line 1
SyntaxError: 'await' outside function

await を非同期関数外から使っていると怒られます。
というわけで、結論。

ならば非同期関数を作ってしまおう

つまり、実行したいコードを非同期関数の中に入れて定義し、exec() の外からその非同期関数を呼び出してやればいいのです。

def get_script(code: str) -> str:
    return 'async def _exec():\n ' + ' \n'.join(code.split('\n'))
    #       ^ 非同期関数を定義        ^ インデントを追加

async def main():
    exec(get_script('await run()'))

非同期関数を定義することができました。
では、それを呼び出してみましょう。しかし、_exec は exec のローカルスコープで定義されているので、直接呼び出すことはできません。呼び出そうとすると、NameError: name '_exec' is not defined. が発生します。
それを解決する方法は複数あります。

locals() を使う

await locals()['_exec']()

locals() でローカル変数の辞書を取得し、定義した関数を取り出して実行しています。

exec() に名前空間として辞書を渡す

よく知られているように、exec は第一引数に実行するコードのオブジェクトを取りますが、Optionalな第二引数、第三引数も存在します。
第二引数は exec() 内でコードを実行するときのグローバルスコープ、第三引数はローカルスコープをそれぞれ取ります。

dic = {}
exec(get_script('await run()'), None, dic)
await dic['_exec']()

ローカルスコープとしての辞書を定義し、そこから定義した関数を取り出して実行しています。

個人的には前者の方が好みです。

まとめ

パンが無いなら作ればいいじゃないの。