Асинхронный python без головной боли (часть 2)

Это продолжение цикла статей про asyncio. Начало здесь.

6. Aiohttp и другие жители асинхронного мира

Продолжаем готовить asyncio. Теперь мы уже знаем достаточно много, чтобы написать модный асинхронный микросервис. Реализуем известный архитектурный паттерн "API-шлюз". Это довольно простая штука. По запросу на свой API-интерфейс приложение собирает данные из других API, обрабатывает и возвращает результат пользователю. При этом пользователь знает только одну точку входа, а все внутренние подробности обработки от него скрыты.

В предыдущей главе мы научились запрашивать погоду у сервиса api.openweathermap.org. Давайте его слегка импортозаместим. Вернее сказать, русифицируем. Пусть пользователь нашего приложения посылает название города на русском языке в параметрах GET-запроса (если вы не знаете что это такое, бегом читать про протокол HTTP) и получает ответ в виде json опять-таки на великом и могучем.

Мы уже освоили http-клиента библиотеки aiohttp, с помощью которого можно обращаться к внешним API. Оказывается, в этой же библиотеке есть и все необходимое для создания полноценного http-сервера. Для начала напишем просто зеркальный прокси:

Пример 6.1

import json
from aiohttp import ClientSession, web


async def get_weather(city):
    async with ClientSession() as session:
        url = f'http://api.openweathermap.org/data/2.5/weather' \
              f'?q={city}&APPID=2a4ff86f9aaa70041ec8e82db64abf56'

        async with session.get(url) as response:
            weather_json = await response.json()
            try:
                return weather_json["weather"][0]["main"]
            except KeyError:
                return 'Нет данных'


async def handle(request):
    city = request.rel_url.query['city']
    weather = await get_weather(city)
    result = {'city': city, 'weather': weather}

    return web.Response(text=json.dumps(result, ensure_ascii=False))


if __name__ == '__main__':
    app = web.Application()
    app.add_routes([web.get('/weather', handle)])
    web.run_app(app)

В асинхронной функции get_weather ничего нового нет, мы ее лишь слегка "облагородили", чтобы запрос погоды для несуществующего города не приводил к трагическим последствиям для всего нашего приложения. За обработку запроса отвечает функция handle ("ручка" на сленге бэкендеров). Из запроса извлекается параметр city и передается в get_weather. Далее формируется результирующий ответ в виде json. Адрес нашего сервиса и тип запроса задается следующим образом: app.add_routes([web.get('/weather', handle)]).

Стоп! А где же наш старый друг asyncio.run? Не переживайте. Когда мы имеем дело с асинхронными веб-фреймворками (а aiohttp — это именно фреймворк, хоть и супер-минималистический), вся работа по созданию и запуску задач asyncio происходит у фреймворка "под капотом". Где-то глубоко в недрах 'web.run_app' создаются задачи, организуется бесконечный цикл и запускается в asyncio.run. Нам, как разработчикам, теперь нет нужды беспокоится об этих низменных деталях. Приложение мирно спит в бесконечном цикле, пока не придет запрос GET на определенный URL. Как только это произойдет, отработает логика в ручке. И снова баю-бай до следующего запроса. Но если первый запрос еще не успел обработаться, как поступил следующий, фреймворк отработает его в отдельной задаче, не дожидаясь (по возможности) окончания обработки первого. В этом сама суть асинхронности.

Заходим браузером на адрес: localhost:8080/weather?city=Sochi и получаем симпатичный json:

{"city": "Sochi", "weather": "Clouds"}

Кстати, если вы всерьез решили заняться бэкэнд-разработкой, одним браузером вам никак не обойтись. Потребуется инструмент, позволяющий залезать вглубь HTTP. Стандарт де-факто здесь Postman, но в природе есть и альтернативные решения.

Скелет нашего приложения готов, теперь начинаем наращивать на него мышцы. Воспользуемся бесплатным API-переводчиком libretranslate.de:

Пример 6.2

import json
from aiohttp import ClientSession, web


async def get_weather(city):
    async with ClientSession() as session:
        url = f'http://api.openweathermap.org/data/2.5/weather' \
              f'?q={city}&APPID=2a4ff86f9aaa70041ec8e82db64abf56'

        async with session.get(url) as response:
            weather_json = await response.json()
            try:
                return weather_json["weather"][0]["main"]
            except KeyError:
                return 'Нет данных'


async def get_translation(text, source, target):
    async with ClientSession() as session:
        url = 'https://libretranslate.de/translate'

        data = {'q': text, 'source': source, 'target': target, 'format': 'text'}

        async with session.post(url, json=data) as response:
            translate_json = await response.json()

            try:
                return translate_json['translatedText']
            except KeyError:
                return text


async def handle(request):
    city_ru = request.rel_url.query['city']
    city_en = await get_translation(city_ru, 'ru', 'en')

    weather_en = await get_weather(city_en)
    weather_ru = await get_translation(weather_en, 'en', 'ru')

    result = {'city': city_ru, 'weather': weather_ru}

    return web.Response(text=json.dumps(result, ensure_ascii=False))


if __name__ == '__main__':
    app = web.Application()
    app.add_routes([web.get('/weather', handle)])
    web.run_app(app)

Теперь в ручке дважды вызывается асинхронная функция get_translation и, вуаля:

localhost:8080/weather?city=Сочи

{"city": "Сочи", "weather": "Облака"}

Что это за микросервис без логгера? Однако использовать в насквозь асинхронном приложении привычную синхронную (а значит блокирующую) библиотеку logging — это непрофессионально. Воспользуемся асинхронной библиотекой логгирования aiologger:

Пример 6.3

import json
from aiohttp import ClientSession, web
from aiologger.loggers.json import JsonLogger


logger = JsonLogger.with_default_handlers(
            level='DEBUG',
            serializer_kwargs={'ensure_ascii': False},
        )


async def get_weather(city):
    async with ClientSession() as session:
        url = f'http://api.openweathermap.org/data/2.5/weather' \
              f'?q={city}&APPID=2a4ff86f9aaa70041ec8e82db64abf56'

        async with session.get(url) as response:
            weather_json = await response.json()
            try:
                return weather_json["weather"][0]["main"]
            except KeyError:
                logger.error(f'Невозможно получить погоду для города: {city}')
                return 'Нет данных'


async def get_translation(text, source, target):
    await logger.info(f'Поступил запрос на на перевод слова: {text}')

    async with ClientSession() as session:
        url = 'https://libretranslate.de/translate'

        data = {'q': text, 'source': source, 'target': target, 'format': 'text'}

        async with session.post(url, json=data) as response:
            translate_json = await response.json()

            try:
                return translate_json['translatedText']
            except KeyError:
                logger.error(f'Невозможно получить перевод для слова: {text}')
                return text


async def handle(request):
    city_ru = request.rel_url.query['city']

    await logger.info(f'Поступил запрос на город: {city_ru}')

    city_en = await get_translation(city_ru, 'ru', 'en')
    weather_en = await get_weather(city_en)
    weather_ru = await get_translation(weather_en, 'en', 'ru')

    result = {'city': city_ru, 'weather': weather_ru}

    return web.Response(text=json.dumps(result, ensure_ascii=False))


if __name__ == '__main__':
    app = web.Application()
    app.add_routes([web.get('/weather', handle)])
    web.run_app(app)

Обратите внимание, асинхронный логгер ни на миллисекунду не задержит работу нашего приложения в целом. Он будет использовать паузы, пока мы ожидаем запроса от пользователя или ответа от внешнего сервиса.

Ну а теперь вишенка на торте — асинхронный доступ к базе данных. Предположим, в процессе работы нашего приложения нам надо что-то писать в БД, ну, например, сохранять поступившие запросы (ничего более умного мне в голову не пришло). Самая простая БД в мире, как известно, — это SQLite. И для нее, к счастью, есть асинхронный драйвер aiosqlite. Пробуем:

Пример 6.4

import json
import aiosqlite
import asyncio
from aiohttp import ClientSession, web
from aiologger.loggers.json import JsonLogger
from datetime import datetime


logger = JsonLogger.with_default_handlers(
            level='DEBUG',
            serializer_kwargs={'ensure_ascii': False},
        )


async def create_table():
    async with aiosqlite.connect('weather.db') as db:
        await db.execute('CREATE TABLE IF NOT EXISTS requests '
                         '(date text, city text, weather text)')
        await db.commit()


async def save_to_db(city, weather):
    async with aiosqlite.connect('weather.db') as db:
        sql = f'INSERT INTO requests VALUES ("{datetime.now()}", "{city}", "{weather}")'

        await db.execute(sql)
        await db.commit()


async def get_weather(city):
    async with ClientSession() as session:
        url = f'http://api.openweathermap.org/data/2.5/weather' \
              f'?q={city}&APPID=2a4ff86f9aaa70041ec8e82db64abf56'

        async with session.get(url) as response:
            weather_json = await response.json()
            try:
                return weather_json["weather"][0]["main"]
            except KeyError:
                logger.error(f'Невозможно получить погоду для города: {city}')
                return 'Нет данных'


async def get_translation(text, source, target):
    await logger.info(f'Поступил запрос на на перевод слова: {text}')

    async with ClientSession() as session:
        url = 'https://libretranslate.de/translate'

        data = {'q': text, 'source': source, 'target': target, 'format': 'text'}

        async with session.post(url, json=data) as response:
            translate_json = await response.json()

            try:
                return translate_json['translatedText']
            except KeyError:
                logger.error(f'Невозможно получить перевод для слова: {text}')
                return text


async def handle(request):
    city_ru = request.rel_url.query['city']

    await logger.info(f'Поступил запрос на город: {city_ru}')

    city_en = await get_translation(city_ru, 'ru', 'en')
    weather_en = await get_weather(city_en)
    weather_ru = await get_translation(weather_en, 'en', 'ru')

    result = {'city': city_ru, 'weather': weather_ru}

    await save_to_db(city_ru, weather_ru)

    return web.Response(text=json.dumps(result, ensure_ascii=False))


if __name__ == '__main__':
    asyncio.run(create_table())

    app = web.Application()
    app.add_routes([web.get('/weather', handle)])
    web.run_app(app)

Можете заглянуть внутрь созданного на лету файла weather.db (только используйте не текстовый просмотрщик, а какую-нибудь приблуду для работы с БД, например, DBeaver). Для каждого запроса создается соответствующая запись в таблице requests. И снова никаких блокировок, мы ведь живем в асинхронном мире.

Смотрите-ка! Неожиданно появился наш забытый друг asyncio.run, к чему бы это? А это к тому, что операция создания таблицы requests (если таковой еще нет) должна выполняться лишь однажды при старте приложения, еще до запуска бесконечного цикла внутри web.run_app. Но функция-то create_table у нас асинхронная, так просто ее не запустишь (попробуйте сами, если не верите). Вот и приходится единожды запускать ее "вручную" при помощи asyncio.run и лишь затем передавать бразды правления web.run_app.

В заключение этого раздела хочу вас поздравить. Теперь вы имеете в руках все необходимое для создания собственных асинхронных веб-приложений. Неважно какой фреймфорк вы будете использовать: FastAPI, Tornado, Falcon или какой-нибудь еще. Принцип останется тот же самый как в старом добром aiohttp: создаем ручку и в ней нанизываем "шашлык" из вызовов асинхронных функций. Главное за чем необходимо следить — это чтобы в эти функции не затесалось что-нибудь блокирующее из скучного пыльного синхронного мира.

На этом мы временно прощаемся.

Продолжение следует...