Работа с ZeroMQ и PostgreSQL в asyncio

Наверное, все уже слышали о чудесной библиотеке asyncio?

Если нет, то коротко: эта либа стала стандартом на асинхронное сетевое программирование на Python. tornado и twisted или научатся с ней работать или постепенно уйдут в маргиналы.

asyncio прекрасна, но это просто библиотека для работы с TCP, UDP, UNIX сокетами, PIPES и асинхронным запуском subprocess.

Чтобы всем было здорово нужны сторонние библиотеки, умеющие работать с asyncio. Кое-что уже есть, но мало.

Вот я с коллегами и сделал пару: одну для ZeroMQ и другую для PostgreSQL

Если вы прочно сидите на Python 2 и не интересуетесь Python 3 — не ходите под кат во избежание глупых вопросов и прочих недоразумений.

Зачем оно мне было надо?

Потому что мы начинаем новый проект в котором без ZeroMQ и Postgress не обойтись. Очень хочется делать на asyncio. Пришлось запилить недостающие кусочки.

Сразу скажу, обе либы добротные. Я Python Core Developer и в коде asyncio тоже есть немало моего труда. Думаю, я понял как хорошо писать для системы, которую сам же помогал создавать.

Плюс (почти) 100% покрытие кода обоих библиотек тестами и внятная (надеюсь) документация.

aiozmq — позволяет использовать ZeroMQ сокеты с asyncio.

Документация — здесь.

aiozmq работает с низкоуровневыми ZeroMQ сокетами и, главное, даёт из коробки механизм Remote Procedure Call (плюс PubSub и Notify).

Короткий пример:

    import asyncio
    import aiozmq
    import aiozmq.rpc

    class ServerHandler(aiozmq.rpc.AttrHandler):
        @aiozmq.rpc.method
        def remote_func(self, a:int, b:int) -> int:
            return a + b

    @asyncio.coroutine
    def go():
        server = yield from aiozmq.rpc.start_server(
            ServerHandler(), bind='tcp://127.0.0.1:5555')
        client = yield from aiozmq.rpc.open_client(
            connect='tcp://127.0.0.1:5555')

        ret = yield from client.rpc.remote_func(1, 2)
        assert 3 == ret

        server.close()
        client.close()

    asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())
    asyncio.get_event_loop().run_until_complete(go())

Теперь перейдем к aiopg

Эта либа умеет работать с psycopg2 в асинхронном режиме и в качестве приятного бонуса даёт connection pool

Документация — здесь.

    @asyncio.coroutine
    def test_select():
        pool = yield from aiopg.create_pool(dsn)

        with (yield from pool.cursor()) as cur:
            yield from cur.execute('SELECT 1')
            ret = yield from cur.fetchone()
            assert ret == (1,), ret

Обе библиотеки выложены на GitHub: aiozmq и aiopg

Пробуйте, если понравилось — используйте.

Найдете баги — пишите на гитхабе в issues, а ещё лучше сразу делайте pull request

Автор: svetlov

Источник