Асинхронное программирование. Библиотека asyncio.

asyncio абстракции

При работе с асинхронностью мы встретились с понятием цикл событий (event loop). Это программная конструкция, которая управляет выполнением различных задач: регистрирует поступление и запускает в подходящий момент.

C помощью синтаксиса await мы определяем места, где можно переключиться на другие ожидающие выполнения задачи.

Рассмотрим подробнее следующий пример:

import asyncio

async def compute(a, b):
    print('Compute...')
    await asyncio.sleep(1.0)
    return a + b

async def print_sum(a, b):
    result = await compute(a, b)
    print('{} + {} = {}'.format(a, b, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

Начиная с версии Python 3.7 синтаксис работы с библиотекой (а именно создание цикла событий) был упрощен. Подробнее про библиотеку можно узнать здесь.

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({i})...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")

async def main():
    await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Waiting & timeouts

Иногда выполнение операции может занять очень длительное время. Например, вы до сих пор не получили ответ от сервера. В случае отсутствия соединения ваша операция может висеть бесконечно долго. В таком случае на асинхронные операции имеет смысл выставлять timeout. Пример на выставление timeout:

import asyncio

async def eternity():
    # Sleep for one hour
    await asyncio.sleep(3600)
    print('yay!')

async def main():
    # Wait for at most 1 second
    try:
        await asyncio.wait_for(eternity(), timeout=1.0)
    except asyncio.TimeoutError:
        print('timeout!')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

async with

Асинхронный контекстный менеджер - это контекстный менджер, который умеет приостанавливать выполнение в методах входа и выхода: __aenter__(), __aexit__()

lock = asyncio.Lock()

# ... later
await lock.acquire()
try:
    # access shared state
finally:
    lock.release()

lock = asyncio.Lock()

# ... later
async with lock:
    # access shared state

Futures

Футуры (futures) — объекты, в которых хранится текущий результат выполнения какой-либо задачи. Это может быть информация о том, что задача ещё не обработана или уже полученный результат; а может быть вообще исключение.

Одна из особенностей футур, что мы можем запустить задачу на исполнение в одной корутине, а получить результат выполнения в другой. У футур есть 4 возможных состояния: + ожидание (pending) + выполнение (running) + выполнено (done) + отменено (cancelled)

Когда футура находится в состояние done, у неё можно получить результат выполнения. В состояниях pending и running такая операция приведёт к исключению InvalidStateError, а в случае canelled будет CancelledError, и наконец, если исключение произошло в самой корутине, оно будет сгенерировано снова при попытке получить результат.

Узнать состояние футуры можно с помощью методов done() или cancelled(), Вызов result() возвращает ожидаемый результат. Для получения исключения есть метод exception(). Для отмены выполнения футуры есть метод cancel(). И result() и exception() выбросят CancelledError, если футура была остановлена в процессе работы.

Ожидание окончания футуры можно сделать при помощи функции wait_for(). Первый аргумент - футура, второй - таймаут (None, если таймаут не нужен).

import asyncio

async def set_after(delay, value):
    # Sleep for *delay* seconds.
    await asyncio.sleep(delay)

    # Set *value* as a result of *fut* Future.
    return value

async def main():
    # Run "set_after()" coroutine in a parallel Task.
    fut = asyncio.ensure_future(
        set_after(1, '... world'))

    print('hello ...')

    # Wait until *fut* has a result (1 second)
    await asyncio.wait_for(fut, None)
    # and print it.
    print(fut.result())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Возможен запуск футуры при помощи await.

async def main():
     # Run "set_after()" coroutine in a parallel Task.
    fut = asyncio.ensure_future(
        set_after(1, '... world'))

    print('hello ...')

    # Wait until *fut* has a result (1 second) and print it.
    # Alternative way to get a result, just use it.
    print(await fut)

aiohttp

Рядом с asyncio создано огромное количество асинхронных модулей для решения всевозможных задач. aiohttp - лишь одна из них. Это асинхронный HTTP Клиент/Сервер

В следующем примере получаем содержимое страницы google.com: (при отсутствии доступа в интернет, cs.mipt.ru)

import aiohttp

async with aiohttp.ClientSession() as session:
    async with session.get('http://google.com') as resp:
        text = await resp.text()
        print('{:.70}...'.format(text))

Реализация простого сервера:

from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', 'Anonymous')
    text = 'Hello, ' + name
    # ...
    # здесь идет некоторая дополнительная логика с async/await
    #
    return web.Response(text=text)

app = web.Application()
app.add_routes([web.get('/', handle),
                web.get('/{name}', handle)])

web.run_app(app)

from collections import namedtuple
import time
import asyncio
from concurrent.futures import FIRST_COMPLETED
import aiohttp

Service = namedtuple('Service', ('name', 'url', 'ip_attr'))

SERVICES = (
    Service('ipify', 'https://api.ipify.org?format=json', 'ip'),
    Service('ip-api', 'http://ip-api.com/json', 'query')
)

async def fetch_ip(service):
    # получение ip


async def asynchronous():
    # TODO:
    # создание футур для сервисов
    # используйте FIRST_COMPLETED

ioloop = asyncio.get_event_loop()
ioloop.run_until_complete(asynchronous())

aiogram

Это библиотека для написания асинхронного Telegram бота.

from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor

PROXY_URL = 'socks5://xxx.xxx.xxx.xxx' # вставить здесь подходящий ip

secret_token = 'XXX'  # строка вида: 123456789:ABCDEFGHJABCDEFGHJABCDEFGHJABCDEFGHJ

bot = Bot(token=secret_token, proxy=PROXY_URL)
dp = Dispatcher(bot)


@dp.message_handler(commands=['start', 'help'])
async def send_welcome(message: types.Message):
    await message.reply("Hi!\nI'm EchoBot!\nPowered by aiogram.")


@dp.message_handler()
async def echo(message: types.Message):
    await message.reply(message.text)


if __name__ == '__main__':
    executor.start_polling(dp)