Selenium异步搜索队列访问self.queue报AttributeError，为何__init__未执行？

问题根本原因 #

你写的Searcher类的构造方法名称拼写错误，导致实例化时构造方法未被执行，所有实例属性都没有初始化：

# 错误写法：init末尾多了1个下划线，不符合Python构造方法的命名规则
def __init___(self):
# 正确写法：init前后各2个下划线
def __init__(self):

Python要求类的构造方法必须严格命名为__init__（前后各两个下划线），你的方法名多了一个下划线，不属于内置构造方法，实例化Searcher()的时候不会自动执行，所以self.queue、self.driver这些属性都没有被创建，调用时就抛出AttributeError。

代码存在的其他问题 #

你修复构造方法后还会遇到其他报错，需要同步修改：

• 队列逻辑错误：你在search方法中刚把任务id加入队列，还没等协程执行就直接pop移除了，队列的顺序控制逻辑完全失效
• 协程未等待：你定义的help协程没有被await就直接返回，而且search方法本身不是异步方法，直接返回协程对象会导致执行异常
• 搜索参数写死：_search方法中你写死了搜索关键词为pixel，没有使用传入的query参数

修复后的参考代码 #

import asyncio
import uuid
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support.expected_conditions import presence_of_element_located
from selenium.common.exceptions import NoSuchElementException


class Searcher:
    def __init__(self):
        self.driver = webdriver.Firefox(executable_path="/home/felix/project/geckodriver")
        self.wait = WebDriverWait(self.driver, 10)
        self.driver.get("https://amazon.com/")
        self.queue = []

    async def _search(self, query):
        # 修复：使用传入的query参数，而非写死pixel
        search_box = self.driver.find_element(By.ID, "twotabsearchtextbox")
        search_box.clear()
        search_box.send_keys(query + Keys.RETURN)
        self.wait.until(presence_of_element_located((By.XPATH,
                "//div[contains(@class, 'sg-col-inner')]/div[contains(@class, 'a-section')]"
                "/div[contains(@class, 'a-section')]")))
        el = self.driver.find_elements(By.XPATH,
                "//div[contains(@class, 'a-section')]/div[h2][not(contains(.,'Sponsored'))]"
                "[contains(@class, 'a-section')]/..")
        out=[]
        for e in el:
            try:
                out.append({
                    "price": e.find_element(By.XPATH, ".//span[contains(@class, 'a-price-whole')]").text + "." +
                    e.find_element(By.XPATH, ".//span[contains(@class, 'a-price-fraction')]").text,
                    "title": e.find_element(By.XPATH, ".//h2").text
                })
            except NoSuchElementException:
                pass
        return out

    async def search(self, query):
        _id = uuid.uuid4()
        self.queue.append(_id)
        try:
            # 修复：等待队列到当前任务再执行
            while self.queue[0] != _id:
                await asyncio.sleep(0.5)
            return await self._search(query)
        finally:
            # 修复：执行完成后再弹出队列
            self.queue.pop(0)

a = Searcher()
print(asyncio.run(a.search("test")))

内容的提问来源于stack exchange，提问作者Felix Farquharson