0

asyncioを使ってリクエストを出している間に他の処理を行ってプログラムの速度改善を行いたいのですが、記述方法が分かりません。
2日ほどasyncioについての下記の記事と睨めっこしているのですが、手も足もでず、イベントループにどのように仕事を渡せば良いか分かりません。

なんとなく理解するasyncio
Pythonの非同期通信(asyncioモジュール)入門を書きました

やりたい事はrequests.get()を出したらawaitしてFutureオブジェクトが完了した順でHTMLをスクレイピングして欲しい商品のURLとマッチしたらreturnでループを抜けるようにしたいです。
これも意味があっているか分からないですが、2日かけて用語のちょっとした理解しか出来ませんでした。

リクエストを次々に投げて、待機している間にリクエストが完了したものからbeatifulsoupで解析を始める。

requestsはasyncioに対応してないので下記の処理コードを追加する必要があるみたいです。

非同期でリクエストを使用するにはどうすればよいですか? - it-swarm.dev

下記の処理を早くしたいです。サイトにある商品一覧のURLを取得して欲しい商品と一致するURLを探すプログラムです。
現状では一個ずつリクエストを出して、サーバからデータのレスポンスがあるまで処理が止まっています。

詳しい方、足場かけをお願いします。一人では歯が立ちません。

import requests
from bs4 import BeautifulSoup

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return links = [url.get('href') for url in items_div]

def want_item_url(links, name, color):
    for link in links:
        url = 'https://www.supremenewyork.com' + link
        #下記のリクエストを非同期にしたい。
        item_page = requests.get(url)
        soup = BeautifulSoup(item_page.content, 'lxml')
        item_name = soup.select('h1[itemprop="name"]')[0].string
        print(item_name)
        item_color = soup.select('#details > p.style')[0].string
        print(item_color)
        if name in item_name and color in item_color:
            return url

links = get_item_urls('accessories')
want_url = want_item_url(links, 'Crew Socks', 'White')

print(want_url)

自分なりに書いて見ました。

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    links = [url.get('href') for url in items_div]
    return links

async def get_html(link):
    url = 'https://www.supremenewyork.com' + link
    item_page = requests.get(url) # ここから非同期
    return item_page

def want_item_url(html, name, color):
    soup = BeautifulSoup(html.content, 'lxml')
    item_name = soup.select('h1[itemprop="name"]')[0].string
    print(item_name)
    item_color = soup.select('#details > p.style')[0].string
    print(item_color)
    if name in item_name and color in item_color:
        return url

async def req(loop, links, name, color):
    for link in links:
        html = await loop.run_in_executor(None, get_html, link)
        url = want_item_url(html, name, color)
    
    return url


def run(links, name, color):
    loop = asyncio.get_event_loop()
    try:
        return loop.run_until_complete(req(loop, links, name, color))
    finally:
        loop.close()

links = get_item_urls('accessories')
want_url = run(links, 'Crew Socks', 'White')

print(want_url)

下記のエラーが出力されます。

Traceback (most recent call last):
  File "non-req.py", line 140, in <module>
    want_url = run(links, 'Crew Socks', 'White')
  File "non-req.py", line 135, in run
    return loop.run_until_complete(req(loop, links, name, color))
  File "/home/vagrant/anaconda3/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "non-req.py", line 127, in req
    url = want_item_url(html, name, color)
  File "non-req.py", line 116, in want_item_url
    soup = BeautifulSoup(html.content, 'lxml')
AttributeError: 'coroutine' object has no attribute 'content'
sys:1: RuntimeWarning: coroutine 'get_html' was never awaited

おそらくhtmlがない状態でBeautifulSoupの解析が始まる事が原因な気がするのですが、どのように処理をHTMLが返ってくるまで止めて置いて、返って来た順に処理を進めて行けば良いのか分かりません。

9/5追記

ほんとにありがとうございます。一人でも組めるようになりたいので、少し教えて下さい。コード内で疑問になった所をコメントで書きました。見て頂けると幸いです。

import asyncio
import requests
from bs4 import BeautifulSoup
import time

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return [url.get('href') for url in items_div]

def search_item(link, name, color):
    url = 'https://www.supremenewyork.com' + link
    item_page = requests.get(url)
    soup = BeautifulSoup(item_page.content, 'lxml')
    item_name = soup.select('h1[itemprop="name"]')[0].string
    print(item_name)
    item_color = soup.select('#details > p.style')[0].string
    print(item_color)
    if name in item_name and color in item_color:
        return url

async def want_item_url(loop, links, name, color):
    # これがなぜ必要なのか分からない。
    async def async_ex(i):
        # これでスレッドを制限している。なしでも可能か?
        async with asyncio.Semaphore(20):
            # ここの関数には非同期処理をしたい関数を代入する。search_itemの関数自体をたくさんのスレッドで動作させている。
            # search_itemの一連の処理が済んだ順でこのコードだとurlがあったら処理を終了するのではなく、とりあえず全部のurlを巡回する仕様
            # 仕様になっている。もし該当urlが見つかった瞬間にイベントループを抜けるにはどうしたら良いでしょうか?
            return await loop.run_in_executor(None, search_item, links[i], name, color)
    # 再帰的になっているなぜ[]なのか返る値はリスト形式に入れられるのか?Futureがここに返ってくるの?
    tasks = [async_ex(i) for i in range(len(links))]
    # gtather渡したシーケンスの順番を保ってくれる。
    return await asyncio.gather(*tasks)

links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
data = [s for s in data if s]

#データの前にある*はなんなのか?
print(*data)
print(time.time() - start)

質問の抜粋

質問
async_exで処理を囲うのが必要な理由

loop.run_in_exexutorについて

個人的解釈
関数には非同期処理をしたい関数を代入する。search_itemの関数自体をたくさんのスレッドで動作させている。
search_itemrequestsでブロックが入り、HTMLが返ってきたら処理を進める。取得したURLを全て巡回してその結果を返す。

質問
もし該当urlが見つかった瞬間にイベントループを抜ける(より早く動作する)にはどうしたら良いでしょうか?それをするのは難しくなりますか?

tasks = [async_ex(i) for i in range(len(links))]

質問
Pythonの内包表記と呼ばれる物でしょうか?
再帰的になっているなぜ[]なのか返る値はリスト形式に入れられるのか?
Futureがここに返ってくるの?

return await asyncio.gather(*tasks)
個人的解釈
gtather渡したシーケンスの順番を保ってくれる。

質問
*dataの前にある*はなんなのか?

非同期処理結果をすべて受け取れるため使いやすいですね。

質問
これはtasks = [async_ex(i) for i in range(len(links))]に未完のfutureオブジェクト入って全て完了したら次の処理に移れる事を指しているのでしょうか?

9/6の追記質問

回答ありがとうございます。

勘違いされていませんか?再帰的にはなっていませんよ。

勘違いしていました。for文で回しているだけでした。

gtatherのシーケンスを保つについてはこちらの記事に書かれていました。

イベントループを途中で抜けたくて

if name in item_name and color in item_color:
     # ここにループを抜ける処理を書く
     # task.cancel()を入れたら上手く動作する。
        return url

task.cancel()を挟めばいいと思うのですが、肝心のtaskをどこから持ってきたら良いか分かりません。
もしご存知でしたら、教えて下さいお願いします。

7
  • 余談になりますが、it-swarm.dev は StackOverflow のコンテンツを機械翻訳しているサイトです。 / "日本語で読める" という利点はあるのかもしれませんが、複製した時点で情報が止まっているので、元ページ も参照することをオススメします。
    – cubick
    Commented 2020年9月3日 16:34
  • ありがとうございます。
    – wataru
    Commented 2020年9月4日 1:19
  • 1
    ただし、1つのサイトにそのようにアクセスするのは、(今の連続アクセスでさえも) DoS攻撃を仕掛けているのと変わらないような気がしますが。
    – kunif
    Commented 2020年9月4日 1:38
  • 1
    カテゴリで絞ってあるので、商品の数が多くても20いかないくらいなので、同時アクセスは20を下回るのでDoS攻撃のようにサーバに負荷を与える処理にはならないと考えています。
    – wataru
    Commented 2020年9月4日 1:51

1 件の回答 1

2

asyncio ではありませんが threading.Thread を使って want_item_url の実装をマルチスレッドにしたサンプルです。

import threading

# 省略

def want_item_url(links, name, color):
    lock = threading.Lock()
    target_url = []
    def search_item(url, name, color):
        item_page = requests.get(url)
        soup = BeautifulSoup(item_page.content, 'lxml')
        item_name = soup.select('h1[itemprop="name"]')[0].string
        print(item_name)
        item_color = soup.select('#details > p.style')[0].string
        print(item_color)
        if name in item_name and color in item_color:
            with lock:
                target_url.append(url)
    threads = []
    for i, link in enumerate(links):
        url = 'https://www.supremenewyork.com' + link
        th = threading.Thread(target=search_item, name="search_"+str(i), args=(url, name, color))
        th.daemon = True
        th.start()
        threads.append(th)
    for th in threads:
        th.join()
    return target_url

links = get_item_urls('accessories')
want_url = want_item_url(links, 'Crew Socks', 'White')

print(*want_url)

以下追記

links の 個々の requests.get() までの処理にして速度比較をした結果(単位:秒)では
Thread も asyncio もほぼ同じ(Thread が asyncio よりほんの少し速い)結果でした。
asyncio に関しては使用したことがないため want_url を書き込む際に排他が必要かどうか
から調べないといけませんので他の方の回答をご期待ください。

Thread 版 平均 0.166sec
0.213, 0.205, 0.147, 0.144, 0.157, 0.203, 0.154, 0.138, 0.148, 0.151
asyncio 版 平均 0.218sec
0.221, 0.203, 0.213, 0.246, 0.216, 0.230, 0.219, 0.218, 0.203, 0.210

Thread 版(性能比較用:search_item() は requests.get(url) のみ)

import requests
from bs4 import BeautifulSoup
import threading
import time

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return [url.get('href') for url in items_div]

def want_item_url(links, name, color):
    lock = threading.Lock()
    target_url = []
    def search_item(url, name, color):
        item_page = requests.get(url)
    threads = []
    for i, link in enumerate(links):
        url = 'https://www.supremenewyork.com' + link
        th = threading.Thread(target=search_item, name="search_"+str(i), args=(url, name, color))
        th.daemon = True
        th.start()
        threads.append(th)
    for th in threads:
        th.join()
    return target_url

daemon = True
links = get_item_urls('accessories')
start = time.time()
want_url = want_item_url(links, 'Crew Socks', 'White')
print(time.time() - start)

asyncio 版

import asyncio
import requests
from bs4 import BeautifulSoup
import time

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return [url.get('href') for url in items_div]

links = get_item_urls('accessories')

def ex(i):
    r = requests.get('https://www.supremenewyork.com' + links[i])
    return len(r.content)

async def handler(loop):
    async def async_ex(i):
        async with asyncio.Semaphore(20):   # BUG!  2回目の追記コード参照
            return await loop.run_in_executor(None, ex, i)
    tasks = [async_ex(i) for i in range(len(links))]
    return await asyncio.gather(*tasks)

start = time.time()
loop = asyncio.get_event_loop()
data = loop.run_until_complete(handler(loop))
print(time.time() - start)

以下2回目の追記です。
asyncio を使用して実装してみました。
非同期処理結果をすべて受け取れるため使いやすいですね。
性能を比較するとなぜか Thread 版の方がほんの少し速いですね。
Thread はリソースをそれなりに消費します。
アクセスする URL が数十であれば Thread でいいでしょうが多い場合は
asyncio を使うべきですね。

asyncio 要求機能実装版

import asyncio
import requests
from bs4 import BeautifulSoup
import time

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return [url.get('href') for url in items_div]

def search_item(link, name, color):
    url = 'https://www.supremenewyork.com' + link
    item_page = requests.get(url)
    soup = BeautifulSoup(item_page.content, 'lxml')
    item_name = soup.select('h1[itemprop="name"]')[0].string
    print(item_name)
    item_color = soup.select('#details > p.style')[0].string
    print(item_color)
    if name in item_name and color in item_color:
        return url

async def want_item_url(loop, links, name, color):
    # 2020/9/5 以下4行修正
    sem = asyncio.Semaphore(20)   # 20: concurrency limitation  追加
    async def async_ex(i):
        #async with asyncio.Semaphore(20):   # BUG! 削除
        async with sem:                      # 修正
            return await loop.run_in_executor(None, search_item, links[i], name, color)
    tasks = [async_ex(i) for i in range(len(links))]
    return await asyncio.gather(*tasks)

links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
data = [s for s in data if s]
loop.close()
print(*data)
print(time.time() - start)

9/5質問への回答

これがなぜ必要なのか分からない。
async def async_ex(i):

タスクの同時動作数を制限しないのであれば、次のように書けますね。

async def want_item_url(loop, links, name, color):
    tasks = [loop.run_in_executor(None, search_item, links[i], name, color) for i in range(len(links))]
    return await asyncio.gather(*tasks)

これでスレッドを制限している。なしでも可能か?
async with asyncio.Semaphore(20):

スレッドを制限してはいません。同時動作タスク数を制限するつもりのコードでした。
(同時動作タスク数を制限できていないコードでした。)
このコードでは、毎回セマフォオブジェクトを生成していますので意味がありません。
(1つのセマフォオブジェクトを共有して初めて意味がある)
バグのあるコードを提示してすみませんでした。修正内容を下に提示します。
(2回目の追記コードも修正しました)
asyncio はイベントループ上でタスクを切り替えています。スレッドを切り替えているわけではありません。
リファレンスでは、イベントループ関連のほとんどがスレッドアンセーフとなっています。

■修正前

async def want_item_url(loop, links, name, color):
    async def async_ex(i):
        async with asyncio.Semaphore(20):

■修正後

async def want_item_url(loop, links, name, color):
    sem = asyncio.Semaphore(20)   # 20: concurrency limitation
    async def async_ex(i):
        async with sem:

ここの関数には非同期処理をしたい関数を代入する。search_itemの関数自体をたくさんのスレッドで動作させている。
search_itemの一連の処理が済んだ順でこのコードだとurlがあったら処理を終了するのではなく、とりあえず全部のurlを巡回する仕様
仕様になっている。もし該当urlが見つかった瞬間にイベントループを抜けるにはどうしたら良いでしょうか?

もし該当 url が見つかった瞬間にイベントループを抜けるには、すでに動作している他のタスクをキャンセルする or 強制終了する等の面倒な処理が必要になるはずです。
リファレンスでは gather がキャンセルされた場合はキャンセルが伝搬されるとあるため gather のキャンセルで可能かもしれません。

再帰的になっているなぜ[]なのか返る値はリスト形式に入れられるのか?Futureがここに返ってくるの?
tasks = [async_ex(i) for i in range(len(links))]

勘違いされていませんか?再帰的にはなっていませんよ。

gtather渡したシーケンスの順番を保ってくれる。
return await asyncio.gather(*tasks)

リファレンスを見た限りでは、何も書かれていないようです。
https://docs.python.org/ja/3.6/library/asyncio-task.html#asyncio.gather

データの前にある*はなんなのか?
print(*data)

アスタリスクは、イテラブルアンパック演算子です。
この場合、リストをアンパックして print 関数の引数に渡しています。

| 非同期処理結果をすべて受け取れるため使いやすいですね。

これはtasks = [async_ex(i) for i in range(len(links))]に未完のfutureオブジェクト入って全て完了したら次の処理に移れる事を指しているのでしょうか?

「非同期処理結果をすべて受け取れる」と書いたのは
一番上のレベルで
data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
と実行結果の各 URL を「戻り値として」すべて受け取れるの意味です。
Thread の場合は global か nonlocal の変数に各スレッドから排他処理して格納しなければなりません。

9/6 の追加質問の回答

gtatherのシーケンスを保つについてはこちらの記事に書かれていました。

こちらの記事には「シーケンスを保つ」ことのエビデンスが何も提示されていません。
実際の動作上は、確かに保たれていそうですが Python の言語リファレンス上では確認できませんでした。
先の async with asyncio.Semaphore(20): も Web の内容をうっかり使ってしまった結果です。
いつも Web 情報が信頼できるわけではありません。

イベントループを途中で抜けたくて

if name in item_name and color in item_color:
# ここにループを抜ける処理を書く
# task.cancel()を入れたら上手く動作する。
return url

task.cancel()を挟めばいいと思うのですが、肝心のtaskをどこから持ってきたら良いか分かりません。
もしご存知でしたら、教えて下さいお願いします。

そこで task.cancel() をして return url 以降がまともに動くのでしょうか?
どうしてもキャンセルしたいのであれば以下のようなアプローチでしょうかね。
asyncio.gather(*tasks) のオブジェクトとキャンセル用のフラグ(キャンセル未)と結果格納用の ans_url 変数をグローバルで保持し、
return url の代わりに「フラグをキャンセル可にし ans_url = url を実行」する。
search_item 関数内の
item_page = requests.get(url) ここが時間がかかる処理と認識しています
の直後で「フラグがキャンセル可であればフラグをキャンセル済みにし gather 結果オブジェクトを cancel() する」(キャンセル済みにすることで2重キャンセルを防止する)
その上で exception 処理を追加

try:
    data = loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
except asyncio.exceptions.CancelledError as e:
    print("CancelledError", e)

等々が必要になるでしょうね。
その上で cancel() 後に CancelledError exception が発生する場合はタスクのリソースが正常に解放されるか心配になります。
キャンセルはコードが汚くなる上に気を付けることがいろいろでてきます。
また、キャンセルしたらすぐに loop.run_until_complete を中断してくれるかも気になります。(通信のレスポンスを待っているタスクをレスポンスが来ても問題ないようにしてすぐにキャンセルできるか?)
排他制御が不要なだけマルチスレッドのキャンセルよりは楽でしょう。
頑張ってみてください。

9/7 追記

グローバル変数でflagを作成したのですが実行すると見つからないとエラーが出ます。

global 宣言がされていないだけではありませんか?
Python における変数へのオブジェクトの代入文は、オブジェクトへの参照情報の代入です。
(リファレンス上は「束縛」という言葉を使っています)
つまり変数への代入は、オブジェクトへの参照先を変更することになります。
global スコープ以外で global 変数に代入以外のアクセスをするのは単にアクセス式を書く
だけですが、参照先を変更する代入を行う場合は global 宣言が必要です。
(代入前に代入以外のアクセスをする場合はその前に global 宣言が必要です。)
global と nonlocal の使い方は基本的な話です、把握しておきましょう。

以下、提示したアプローチによるサンプルコードです。
キャンセル処理に動きがわかるように print を入れています。
実際に動作させると時間のかかる requests.get(url) は並行して動作しているため一斉に終了します。
そのため requests.get(url) 直後に cancel() 処理を実行してもキャンセルすることはありませんでした。
soup.select() 後に入れることでキャンセルできますが、キャンセル処理を入れる効果は少ないですね。
こちらの環境では Thread 版の方が速いですよ。
20程度の URL アクセスで少しでも早くしたいのであれば Thread 版を検討されてはいかがでしょうか?
asincio はアプリケーションレベルでイベントループ内でタスク切り替えを行うため結構処理が重いようです。
(Thread は OS レベルで切り替えが行われます。)
なお、これまでのコードには loop.close() がなくお行儀が悪いコードでしたので、前の asyncio 要求機能実装版のコードも含め追記しました。

asyncio 要求機能実装版(Cancel 実装) 改善前

import asyncio
import requests
from bs4 import BeautifulSoup
import time
from enum import Enum

class TaskStatus(Enum):
    NonCancelable = 0; Cancelable = 1; Canceled = 2

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return [url.get('href') for url in items_div]

taskflag = TaskStatus.NonCancelable
ans_url = ""

def search_item(link, name, color):
    global taskflag, ans_url
    url = 'https://www.supremenewyork.com' + link
    if taskflag != TaskStatus.NonCancelable:
        if taskflag == TaskStatus.Cancelable:
            gtask.cancel()
            taskflag = TaskStatus.Canceled
            print('*** gtask.cancel() prev requests.get(url) ***')
        return
    print('prev: ', time.time() - start)
    item_page = requests.get(url)
    print('after: ', time.time() - start)
    if taskflag != TaskStatus.NonCancelable:
        if taskflag == TaskStatus.Cancelable:
            gtask.cancel()
            taskflag = TaskStatus.Canceled
            print('*** gtask.cancel() after requests.get(url) ***')
        return
    soup = BeautifulSoup(item_page.content, 'lxml')
    item_name = soup.select('h1[itemprop="name"]')[0].string
    print(item_name)
    if taskflag != TaskStatus.NonCancelable:
        if taskflag == TaskStatus.Cancelable:
            gtask.cancel()
            taskflag = TaskStatus.Canceled
            print('*** gtask.cancel() after soup.select(h1) ***')
        return
    item_color = soup.select('#details > p.style')[0].string
    print(item_color)
    if taskflag != TaskStatus.NonCancelable:
        if taskflag == TaskStatus.Cancelable:
            gtask.cancel()
            taskflag = TaskStatus.Canceled
            print('*** gtask.cancel() after soup.select(details) ***')
        return
    if name in item_name and color in item_color:
        if taskflag == TaskStatus.NonCancelable:
            taskflag = TaskStatus.Cancelable
            ans_url = url
            print('*** set ans_url ***')

async def want_item_url(loop, links, name, color):
    sem = asyncio.Semaphore(20)   # 20: concurrency limitation
    async def async_ex(i):
        async with sem:
            return await loop.run_in_executor(None, search_item, links[i], name, color)
    tasks = [async_ex(i) for i in range(len(links))]
    global gtask
    gtask = asyncio.gather(*tasks)
    return await gtask

links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
except asyncio.exceptions.CancelledError as e:
    print("*** CancelledError ***", e)
finally:
    loop.close()

print(ans_url)
print(time.time() - start)

item_page = requests.get(url) の前にもキャンセル処理追加 2020/09/07 12:21

上のコードでは gtask.cancel() を 3 ~ 4 回実行してしまっていました。
gtask.cancel() はタスク切り替えのイベントとなっているようです。

        if taskflag == TaskStatus.Cancelable:
            gtask.cancel()
            taskflag = TaskStatus.Canceled

そのため本来であれば

        if taskflag == TaskStatus.Cancelable:
            taskflag = TaskStatus.Canceled
            gtask.cancel()

と先にフラグを更新しないと多重キャンセルの対策になっていませんでした。
ただ、よく考えると
return url
をなくし
ans_url = url
とグローバルの ans_url に格納したので
ans_url = url
gtask.cancel()
とすることができました。この対応をした改善版を以下に提示します。

asyncio 要求機能実装版(Cancel 実装) 改善版

import asyncio
import requests
from bs4 import BeautifulSoup
import time

def get_item_urls(category):
    url = 'https://www.supremenewyork.com/shop/all/' + category 
    category_page = requests.get(url)
    soup = BeautifulSoup(category_page.content, 'lxml')
    items_div = soup.select('article > .inner-article > a')
    return [url.get('href') for url in items_div]

canceled = False
ans_url = ""

def search_item(link, name, color):
    global canceled, ans_url
    url = 'https://www.supremenewyork.com' + link
    if canceled: return
    item_page = requests.get(url)
    if canceled: return
    soup = BeautifulSoup(item_page.content, 'lxml')
    item_name = soup.select('h1[itemprop="name"]')[0].string
    print(item_name)
    if canceled: return
    item_color = soup.select('#details > p.style')[0].string
    print(item_color)
    if canceled: return
    if name in item_name and color in item_color:
        if not canceled:
            canceled = True
            ans_url = url
            print('*** set ans_url ***')
            gtask.cancel()

async def want_item_url(loop, links, name, color):
    sem = asyncio.Semaphore(20)   # 20: concurrency limitation
    async def async_ex(i):
        async with sem:
            return await loop.run_in_executor(None, search_item, links[i], name, color)
    tasks = [async_ex(i) for i in range(len(links))]
    global gtask
    gtask = asyncio.gather(*tasks)
    return await gtask

links = get_item_urls('accessories')
start = time.time()
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(want_item_url(loop, links, 'Crew Socks', 'White'))
except asyncio.exceptions.CancelledError as e:
    print("*** CancelledError ***", e)
finally:
    loop.close()

print(ans_url)
print(time.time() - start)

上記コードの if canceled: return は必須ではありません。
gtask.cancel() 後、少しでも無駄な処理を省くために入れています。

18
  • とてもありがたいのですが、下記の記事でthreadでは速度があまり出ないと記述されているので可能であればasyncioを使用したいです。note.com/hagure_melon/n/ndf2f05bc791c
    – wataru
    Commented 2020年9月4日 7:40
  • asyncio 使用例を追加しました。
    – lehshell
    Commented 2020年9月4日 13:56
  • ありがとうございます。上記のコードは性能比較のみで目的の動作をするものではないですよね。
    – wataru
    Commented 2020年9月4日 14:03
  • あと上記のコード実行すると0.4711768627166748と表示されるのですが、明らかに0.4秒以上かかっています。
    – wataru
    Commented 2020年9月4日 14:09
  • 2回目の追記は「目的の動作をするもの」ですよ。性能値は、こちらの環境での値です。
    – lehshell
    Commented 2020年9月4日 14:58

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。