0

Программа проверяет на доступность различные url'ы. Через цикл это очень медленно из-за того, что сервера отвечают не сразу и все последовательно. Была мысль завернуть их все в потоки, но кажется это не правильно. Помогите пожалуйста решить это через асинхронность.

У меня сейчас такой код (упрощенная версия):

def chk_url(url):
    request = get(url,timeout=5)
    if request.status_code != 200: return False
    if not 'status:ok' in request.text.lower(): return False
    return True

urls = [...]
avalible = []
for url in urls:
    if chk_url(url): avalible.append(url)
6
  • 1
    Вы смотрели мой ответ? 1 мар 2022 в 14:43
  • @Александр Смотрел, но сделал по-другому, через потоки, но с собственной реализацией "пулов", чтобы код был короче и понятнее
    – AlgiZ
    3 мар 2022 в 23:52
  • Извините а я для кого своё время убивал? Вы тогда опубликуйте свой ответ. Иначе вам ни кто не будет здесь помогать. 4 мар 2022 в 4:19
  • И если вы думаете что я не могу написать короткий код вы вы заблуждаетесь. Но в ряд ли коротко и понятно совместимо. 4 мар 2022 в 4:21
  • @Александр К чему столько агрессии? Я ведь не говорил, что вы чего-то не можете. Ваш ответ, натолкнул на мысль о пуле потоков, спасибо. Или вы тут галочки и голоса зарабатываете?
    – AlgiZ
    4 мар 2022 в 22:57

1 ответ 1

1

Ну допустим так в два раза сокращается время работы функции.

Это на URL чем их больше, тем ощутимее разница.

import time
from multiprocessing import Pool
from functools import wraps

import requests

results = []


def timing(function):
    @wraps(function)
    def wrapper(*args, **kw):
        time_start = time.time()
        result = function(*args, **kw)
        time_end = time.time()
        print(f'function: {function.__name__} arguments: [{args}] took: {time_end-time_start} seconds')
        return result
    return wrapper


def save_result(list_data):
    # print(data)
    for tuple_data in list_data:
        url, status, error = tuple_data
        print(f"URL: {url}, STATUS: {status}, ERROR: {error}")


@timing
def check_url(url):
    try:
        response = requests.get(url, timeout=30)
    except requests.exceptions.ReadTimeout:
        return url, None, requests.exceptions.ReadTimeout
    except requests.exceptions.ConnectionError:
        return url, None, requests.exceptions.ConnectionError
    if response.status_code != 200:
        return url, response.status_code, None
    else:
        return url, 200, None


@timing
def test_check_url(urls):
    for url in urls:
        check_url(url)


@timing
def test_check_url_async(urls):
    with Pool(processes=2) as pool:
        pool.map_async(check_url, urls, callback=save_result).wait()



if __name__ == '__main__':
    urls = ["https://www.google.com",
           "https://www.yandex.ru",
           "https://ru.wikipedia.org",
           "https://www.python.org",
           "https://www.microsoft.com"]

    test_check_url(urls)
    test_check_url_async(urls)
    time.sleep(5)

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.