Blank File

LinuxとかPythonとかVimとか、趣味でいじる感じで

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点です。

  1. テスト関数の定義がasync defではなくdef
  2. await の代わりに syncer.sync でラップしている

鋭い方は「asyncio.get_event_loop().run_until_complete() と何が違うの?」と思われたかもしれませんが、違いはありません。 同じです。 タイプ数が少ないだけです。 正直デコレータだけで十分な気はしたのですが、せっかくなので関数としても使えるようにしただけです。

まとめ

非同期な関数を同期的に実行できるライブラリ、syncer を紹介しました。

試してみてバグやおかしい部分があれば Issue にお知らせいただけると助かります。