1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

PythonAdvent Calendar 2023

Day 23

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

Last updated at Posted at 2023-12-22

この記事は、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()'))

非同期関数を定義することができました。
では、それを呼び出してみましょう。しかし、_execexec のローカルスコープで定義されているので、直接呼び出すことはできません。呼び出そうとすると、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']()

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

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

まとめ

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

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?