Проблемы с requests + threading

Question

Хотел написать чекер для прокси, но столкнулся с проблемой при работе с многопоточностью.

Код выглядит так

Класс чекер

import requests as req


class Checker():
    def __init__(self, normal_status_code=200, check_address="http://google.com"):
        self.check_address = check_address
        self.normal_status_code = normal_status_code

    def check_proxy(self, proxy_dict):
        try:
            r = req.get(self.check_address, proxies=proxy_dict)
        except req.exceptions.ProxyError:
            return False
        if r.status_code == self.normal_status_code:
            return True
        else:
            return False

Соответсвенно если для каждого из наборов прокси порт вызвать метод check прокси то мы получим список нормальных проксей. Но это будет работать очень медленно, поэтому я использую библиотеку threading

from Checker import Checker
import threading


def check(proxy):
    """Функция для работы с многопоточностью"""
    print(checker.check_proxy(proxy))


threads = 10

checker = Checker()
proxy_filename = "proxy.txt"
proxy_list = []

try:
    f = open(proxy_filename, "r")
except FileNotFoundError:
    filename = input("Введите название файла прокси ")
    f = open(filename, "r")

for line in f:
    proxy = "http://" + line
    proxy = proxy.replace("\n", "")
    proxy_list.append({"http": proxy})

print(proxy_list)

for i, proxy_dict in enumerate(proxy_list):
    t = threading.Thread(target=check, name=f"thread {i}", args=proxy_dict)
    t.start()

Этот код выдаёт вот такую ошибку:

[{'http': 'http://130.185.122.203:3128'}]
Exception in thread thread 0:
Traceback (most recent call last):
  File "C:\Users\Alex\AppData\Local\Programs\Python\Python39\lib\threading.py", line 950, in _bootstrap_inner
    self.run()
  File "C:\Users\Alex\AppData\Local\Programs\Python\Python39\lib\threading.py", line 888, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\Alex\PycharmProjects\Proxy\main.py", line 7, in check
    print(checker.check_proxy(proxy))
  File "C:\Users\Alex\PycharmProjects\Proxy\Checker.py", line 21, in check_proxy
    r = req.get(self.check_address, proxies=proxy_dict)
  File "C:\Users\Alex\PycharmProjects\Proxy\venv\lib\site-packages\requests\api.py", line 76, in get
    return request('get', url, params=params, **kwargs)
  File "C:\Users\Alex\PycharmProjects\Proxy\venv\lib\site-packages\requests\api.py", line 61, in request
    return session.request(method=method, url=url, **kwargs)
  File "C:\Users\Alex\PycharmProjects\Proxy\venv\lib\site-packages\requests\sessions.py", line 532, in request
    settings = self.merge_environment_settings(
  File "C:\Users\Alex\PycharmProjects\Proxy\venv\lib\site-packages\requests\sessions.py", line 710, in merge_environment_settings
    no_proxy = proxies.get('no_proxy') if proxies is not None else None
AttributeError: 'str' object has no attribute 'get'

Process finished with exit code 0

Answer 1

Проблема в том, что вы передаете параметры функции потока в конструктор Thread неправильно.

Аргумент args трактуется конструктором Thread как последовательность и, соответственно, аргумент функции, если он единственный, нужно передавать в контейнере, например в кортеже:

t = threading.Thread(target=check, name=f"thread {i}", args=(proxy_dict,))

Этот пример демонстрирует разницу:

def f(*params):
    print(params)

# тут каждый элемент последовательности т.е. каждый символ
# будет передан как отдельный параметр в f
# будет напечатано `('a', 'b', 'c')`
t = threading.Thread(target=f, args='abc')
t.start()
t.join()

# тут 'abc' будет передан как один параметр в f
# будет напечатано `('abc')`
t = threading.Thread(target=f, args=('abc',))
t.start()
t.join()

БЛОГ НА HUSL

Проблемы с requests + threading

Ответы (1 шт):