Как использовать asyncio в стандартном for цикле?

Question

[Александр Лебедев]

Утро доброе :) Python 3.6.

Кратко:
Есть список url сайтов. Есть цикл for, который идет по этому списку сайтов. Есть две функции: одна проверяет скорость сайта через Google Pagespeed, вторая часть GTMetrix. Идут по очереди, в каждой есть несколько sleep, для ожидания результатов проверки.

Я бы хотел запускать эти две проверки асинхронно, чтобы не ждать, образно, 2 минуты, а ждать одну, ведь API разные, и запрашивать их можно одновременно.

Примерный исходный код цикла:

for item in ad_items:
    site_url = 'http://'+item

    #Получение статистики по скорости загрузки сайта
    ad_items_dict['keyword_sites'][clear_url]['gtm_results'] = gtm_check_speed(site_url)
    print('- Проверка скорости загрузки сайта завершена!')
    
    #Получение статистики по Pagespeed
    ad_items_dict['keyword_sites'][clear_url]['google_desktop_speed'] = pagespeed_request('desktop', site_url)
    print('- Проверка PageSpeed для ПК завершена!')
    ad_items_dict['keyword_sites'][clear_url]['google_mobile_speed'] = pagespeed_request('mobile', site_url)
    print('- Проверка PageSpeed для мобильных завершена!')

Примерный код функции:

#Проверка данных Google PageSpeed
async def pagespeed_request(strategy, site_url):
    pagespeed_request = requests.get('https://www.googleapis.com/pagespeedonline/v2/runPagespeed?url={0}&filter_third_party_resources=true&locale=en_US&screenshot=false&strategy={1}&key={2}'.format(site_url, strategy, PAGESPEED_API))
    pagespeed_request.encoging = 'utf8'
    pagespeed = json.loads(pagespeed_request.text)
    await asyncio.sleep(15)
    return pagespeed['ruleGroups']['SPEED']['score']

Из того, что я смог понять и прочесть, мне необходимо в начале прохода каждого for цикла создавать loop, а в конце его заканчивать, чтобы он начинался заново, когда я перехожу на следующий пункт for цикла. То есть примерно так:

for item in ad_items:
    site_url = 'http://'+item
    loop = asyncio.get_event_loop()

    #Получение статистики по скорости загрузки сайта
    ad_items_dict['keyword_sites'][clear_url]['gtm_results'] = gtm_check_speed(site_url)
    print('- Проверка скорости загрузки сайта завершена!')
    
    #Получение статистики по Pagespeed
    ad_items_dict['keyword_sites'][clear_url]['google_desktop_speed'] = pagespeed_request('desktop', site_url)
    print('- Проверка PageSpeed для ПК завершена!')
    ad_items_dict['keyword_sites'][clear_url]['google_mobile_speed'] = pagespeed_request('mobile', site_url)
    print('- Проверка PageSpeed для мобильных завершена!')

    loop.close()

Все функции промечены async, все sleep заменены на await asyncio.sleep, но в таком случае я получаю ошибку coroutine was never awaited, потому что, по логике, я так ни одну функцию и не запустил асинхронно.

Собственно, вопрос:
Если я использую что-то вроде loop.run_until_complete(pagespeed_request('desktop', site_url)), то цикл заканчивается практически моментально. Я не до конца понимаю логику, как мне сделать так, чтобы каждая функция выполнилась асихронно сама по себе, но asyncio цикл выключился только тогда, когда отработали они все и пора переходить на новую итерацию for цикла. Как это можно реализовать? :)

Буду благодарен за любой совет. Спасибо.

Answer 1

[Александр Лебедев]

Ответ от товарища enchantner:

Если тебе надо две корутины запустить “параллельно” - делай два инстанса через asyncio.ensure_future(), и поверх них делай asyncio.gather().

Если у тебя цикл в синхронном коде - то внутри цикла делай то же самое, только с шелухой сверху:

loop.run_until_complete(
  asyncio.gather(
       asyncio.ensure_future(first_cor()),
asyncio.ensure_future(second_cor())
  )
)

Answer 2

[smf002]

Как вариант можно использовать grequests

Comments

[webus]

А она еще жива ? Последний коммит больше года назад

[smf002]

Ну тут от задачи зависит. Мне пригодилось для небольших скриптов. А так конечно вряд ли кто-то будет баги выпиливать.

Answer 3

[Никита Корольков]

import asyncio

import aiohttp


async def get_urls_statistic(urls: list):
    """
    Получение статистики по сайтам

    :param urls: список сайтов для которых вернется статистика
    :return: {url: {'gtm_speed': ..., 'pagespeed_desctop': ..., 'pagespeed_mobile': ...}}
    """

    # одновременно обрабатываем все сайты, которые бросили в asyncio
    # sites_statistics - это список словарей, расположенные по порядку сайтов
    sites_statistics = await asyncio.gather(*[get_site_statistic(url) for url in urls])
    # объединяем url и его статистику
    return {url: statistic for url, statistic in zip(urls, sites_statistics)}


async def get_site_statistic(url) -> dict:
    # так же одновременно делаем запросы на получение разных статистик
    gtm_speed, pagespeed_desctop, pagespeed_mobile = await asyncio.gather(
        get_site_speed(url),
        get_pagespeed_statistic(url, 'desctop'),
        get_pagespeed_statistic(url, 'mobile')
    )
    return {
        'gtm_speed': gtm_speed,
        'pagespeed_desctop': pagespeed_desctop,
        'pagespeed_mobile': pagespeed_mobile,
    }


async def get_site_speed(url):
    """
    тут типо должны использовать сайт для измерения скорости и потом парсить ответ

    """
    gtm_url = ''  # prepare your
    response = await get_response(url)
    # ... parse response
    return response.content


async def get_pagespeed_statistic(site_url, version):
    response = await get_response(site_url)
    # ... parse response
    return response.content


async def get_response(url, base_semaphore=asyncio.Semaphore(5)) -> aiohttp.ClientResponse:
    """
    Получение тела ответа
    
    :param url: куда делаем запрос
    :param base_semaphore: семафор, который не будет давать делать
        больше 5 подключений к сайтам (а то нас забанят)

    в данном примере base_semaphore - переменная, которая один раз инициализировалась,
    и при каждом вызове функции get_response будет использоваться base_semaphore
    """

    # код ниже может быть запущен одновременно не больше 5 раз, если больше
    # то будем ждать, завершения
    async with base_semaphore:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return response


if __name__ == '__main__':
    # # получаем asyncio цикл событий
    loop = asyncio.get_event_loop()
    # говорим циклу событий: работай пока все не сделаешь и верни мне результат!!!
    statistics = loop.run_until_complete(get_urls_statistic([
        'https://pythondigest.ru/',
        'https://yandex.ru',
        # ...
    ]))
    print(statistics)

    # с python3.7 рекумондуют использовать asyncio.run