Python aiohttp timeout?

Question

[Кирилл Горелов]

Парни, как поставить таймоут на текущий url, а не на все.
Два кусочка кода.

async with session.get(url, headers=headers) as response:
            date = response.headers.get("DATE")

async with aiohttp.ClientSession() as session:
        for url in check_site:
            task = asyncio.ensure_future(bound_fetch(sem, url, session))
            tasks.append(task)
        responses = asyncio.gather(*tasks)
        await responses

Вот так, работает очень долго если много url, т.к. все сайты разные и отвечают с разным временем.
Если я ставлю timeout

async with session.get(url, timeout=30, headers=headers) as response:

то скрипт работает всего 30 секунд, а не на каждый адрес ставиться отдельный timeout.

То есть примеру даю на вход 1000 адресов, те что успели открыть за 30 секунд получаю ответ, а остальные, идут в исключение, т.к. не успел их проверить.
Точно такая же ситуация с read_timeout и conn_timeout.

Comments

[Ivan Yakushenko]

session.get(url, timeout=30, headers=headers)
Это и есть правильный вариант. Скиньте полный traceback ошибки, когда вы пытаетесь запустить этот скрипт.
Вероятнее всего дело не в том, что вы timeout неправильно настраиваете, а в том что пытаетесь установить 1000 подключений.

[Кирилл Горелов]

Ivan Yakushenko, условно говоря вот такая ошибка

http://yandex.ru : exeption / class <class 'concurrent.futures._base.TimeoutError'>

А это успешное выволнение

http://yandex.ru -> Status Code: 302
Wed, 04 Sep 2019 10:23:46 GMT:https://yandex.ru/ with delay

[Ivan Yakushenko]

Кирилл Горелов, я так понимаю bound_fetch(sem, url, session) где sem - это Semaphore?
Как вы его используете? Так?

async with sem:
    async with session.get():
        ...

[Кирилл Горелов]

Ivan Yakushenko, sem - это Semaphore.

sem = asyncio.Semaphore(1000)

    async with aiohttp.ClientSession() as session:
        for url in check_site:
            task = asyncio.ensure_future(bound_fetch(sem, url, session))
            tasks.append(task)

        responses = asyncio.gather(*tasks)
        await responses

loop = asyncio.get_event_loop()
global_start_time = time.time()

future = asyncio.ensure_future(run())
loop.run_until_complete(future)

[Ivan Yakushenko]

Кирилл Горелов, да, и передавать его в качестве аргумента не нужно. Он должен быть инициализирован примерно так:

sem = asyncio.Semaphore(100)


async def bound_fetch(url, session):
    async with sem:
        async with session.get():
            ...


async def main():
    async with aiohttp.ClientSession() as session:
        for url in check_site:
            task = asyncio.ensure_future(bound_fetch(sem, url, session))
            tasks.append(task)
        responses = asyncio.gather(*tasks)
        await responses

И убедитесь что у вас bound_fetch возвращает date.

[Кирилл Горелов]

Ivan Yakushenko, эээээм

Вот мой полный код

from  domainlist import site_pages, real_site,site_pages_mini, site_pages_react, site_pages_all,site_pages_random_3000,site_pages_random_1000, site_pages_random_2000, list_5000
from memory_profiler import memory_usage
import random
import aiohttp
import asyncio
import time

async def fetch(url, session):
    headers = {
        "Accept":"application/json, text/plain, */*",
        "User-Agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36",
        "Accept-Language":"zh-CN,zh;q=0.9",
        "accept-encoding":"gzip, deflate"}

    try:
        async with session.get(url, timeout=0.1, headers=headers) as response:
            date = response.headers.get("DATE")
            code_status = response.history[0].status if response.history else response.status
            print('%s -> Status Code: %s' % (url, code_status))
            return await response.read()
    except Exception as e:
        print("{} : exeption {}/ class {}".format(url, e, e.__class__))
        pass
        
    


async def bound_fetch(sem, url, session):
    async with sem:
        await fetch(url, session)


async def run():
    tasks = []
    check_site = real_site
    sem = asyncio.Semaphore(1000)

    async with aiohttp.ClientSession() as session:
        for url in check_site:
            task = asyncio.ensure_future(bound_fetch(sem, url, session))
            tasks.append(task)

        responses = asyncio.gather(*tasks)
        await responses

loop = asyncio.get_event_loop()
global_start_time = time.time()

future = asyncio.ensure_future(run())
loop.run_until_complete(future)

print("--- %s seconds ---" % (time.time() - global_start_time))

[Ivan Yakushenko]

Кирилл Горелов,

from  domainlist import site_pages, real_site,site_pages_mini, site_pages_react, site_pages_all,site_pages_random_3000,site_pages_random_1000, site_pages_random_2000, list_5000
from memory_profiler import memory_usage
import random
import aiohttp
import asyncio
import time

sem = asyncio.Semaphore(100)


async def fetch(url, session):
    headers = {
        "Accept":"application/json, text/plain, */*",
        "User-Agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36",
        "Accept-Language":"zh-CN,zh;q=0.9",
        "accept-encoding":"gzip, deflate"}

    try:
        async with session.get(url, timeout=0.1, headers=headers) as response:
            date = response.headers.get("DATE")
            code_status = response.history[0].status if response.history else response.status
            print('%s -> Status Code: %s' % (url, code_status))
            return await response.read()
    except Exception as e:
        print("{} : exeption {}/ class {}".format(url, e, e.__class__))
        pass
        


async def bound_fetch(url, session):
    async with sem:
        await fetch(url, session)


async def run():
    tasks = []
    check_site = real_site

    async with aiohttp.ClientSession() as session:
        for url in check_site:
            task = asyncio.ensure_future(bound_fetch(url, session))
            tasks.append(task)

        responses = asyncio.gather(*tasks)
        await responses

loop = asyncio.get_event_loop()
global_start_time = time.time()

future = asyncio.ensure_future(run())
loop.run_until_complete(future)

print("--- %s seconds ---" % (time.time() - global_start_time))

[Ivan Yakushenko]

Кирилл Горелов, не запускайте одновременно 1000 тасков. Если работаете под Windows, то он просто не осилит столько.

[Кирилл Горелов]

Ivan Yakushenko, не у меня linux.
А сколько тогда оптимально?

[Ivan Yakushenko]

Кирилл Горелов, пробуйте разные значения. Начинайте со 100.

[Ivan Yakushenko]

Кирилл Горелов, и установите ulimit -n 65535 на всякий случай, если планируете таки много тасков запускать.

[Кирилл Горелов]

Ivan Yakushenko, Спасибо))
Давай плюсану тебе))
Пиши как ответ!

[Ivan Yakushenko]

Кирилл Горелов, это вам помогло?

[Кирилл Горелов]

Ivan Yakushenko, ну вроде как стало лучше)))
То есть проверяет все и не выбрасывает по было раньше.

Answer 1

[Ivan Yakushenko]

sem = asyncio.Semaphore(100)
Объявите в корне файла, а не в функции и передавать его в качестве аргумента не нужно.
Не используйте большое кол-во тасков, ошибки могут быть из-за этого. И если все-же хотите попробовать, то установите ulimit -n 65535

Comments

[Кирилл Горелов]

Еще вопрос назрел.
А как получить скорость полной загрузки сайта? И вес страницы сайта?
С помощью aiohttp, что-то ищу по документации да и в интернете не нахожу что-то ничего.

[Ivan Yakushenko]

Кирилл Горелов,
вес страницы как вариант:

import sys

text = await response.text()
sys.getsizeof(text) # размер в байтах

Время загрузки - только засекать текущее время непосредственно перед отправкой запроса и вычитать его из текущего времени сразу после. Примерно так:

start_time = time.time()
async with session.get(url, headers=headers) as response:
    response_time = time.time() - start_time
    date = response.headers.get("DATE")

[Кирилл Горелов]

Ivan Yakushenko, у меня так же получилось. НО время загрузки показывает нереальное.
По 3-6 секунды, хотя открываю сайт, смотрю "инспектор", там за 1-2 секунды.

Ну ладно, спасибо.
Я просто думал я может в документации что пропустил.