Pythonの非同期な関数を同期的に実行できるライブラリ、syncerを作りました
Python 3.5 から使える async/await 記法ですが、テストの際などは非同期な関数を実行するのにひと手間必要で面倒でした。 そこで、非同期な関数を同期的に実行するライブラリ、syncer を作りました。
syncerは @sync
というデコレータ用の関数をひとつだけ提供する非常に小さなライブラリです。
主にテストでの利用を想定していますが、テスト以外でも汎用的に使えるライブラリになっています。
syncerはPython 3.4以上に対応しています。
3.4では async/await は使えませんが、@asyncio.coroutine
で装飾された関数を同期的に実行することができます。
インストール
pip
でインストールできます。
python3 -m pip install syncer
使い方
以下の例ではテストはunittestを使って書いていますが、noseやpytestにも対応しています。 ですがそれらのライブラリは似た機能を提供するプラグインもあるのでそちらを使ってもいいかもしれません。
デコレータとして使う
以下のような関数があった時:
async def async_add(i, j, wait_time=0): await asyncio.sleep(wait_time) return i + j
以下のようにテスト関数を async def
で定義して実行してもテストは実行されません。
失敗するはずでも成功になってしまいます。
import unittest class TestAsync(unittest.TestCase): async def test_async(self): self.assertEqual(await async_add(1, 1), 3)
一応以下のような警告が出るはずです。
RuntimeWarning: coroutine 'TestAsync.test_async' was never awaited
そこで、テスト関数をsyncer.sync
で装飾し、同期的に実行されるようにします。
import unittest import syncer class TestAsync(unittest.TestCase): @syncer.sync async def test_async(self): self.assertEqual(await async_add(1, 1), 3)
今度は無事にテストが実行され、きちんとテストが失敗になっているはずです。
関数として使う
syncer.sync
はコルーチンやフューチャーを引数にとる関数としてawait
の代わりに通常の関数の中で使用することもできます。
先ほどのテストは以下のように書くこともできます。
import unittest import syncer class TestAsync(unittest.TestCase): def test_async(self): self.assertEqual(syncer.sync(async_add(1, 1)), 3)
先ほどとの違いは以下の2点です。
- テスト関数の定義が
async def
ではなくdef
await
の代わりにsyncer.sync
でラップしている
鋭い方は「asyncio.get_event_loop().run_until_complete()
と何が違うの?」と思われたかもしれませんが、違いはありません。
同じです。
タイプ数が少ないだけです。
正直デコレータだけで十分な気はしたのですが、せっかくなので関数としても使えるようにしただけです。
まとめ
非同期な関数を同期的に実行できるライブラリ、syncer を紹介しました。
試してみてバグやおかしい部分があれば Issue にお知らせいただけると助かります。