Python Asyncio を使用して Web サイトのステータスチェックを実装する方法

asyncio を使用すると、ストリームを開いて HTTP リクエストとレスポンスを読み書きすることで、Web サイトの HTTP ステータスをクエリできます。

その後、asyncio を使用して複数の Web サイトのステータスを同時にクエリし、結果を動的にレポートすることもできます。

1. Asyncio を使用して HTTP ステータスを確認する方法

asyncio モジュールは、ソケット接続のオープンと、ストリームを介したデータの読み書きのサポートを提供します。この機能を使用して、Web ページのステータスを確認できます。

#これには 4 つの手順が必要になる場合があります。

• 接続を開く

• リクエストを書きます

• 応答を読む

• 接続を閉じる

2. HTTP 接続を開く

使用可能asyncio.open_connection() 関数は、asyncio で接続を開きます。多くのパラメータの中で、この関数は文字列のホスト名と整数のポート番号を受け取ります。

これは、ソケットを使用した読み取りと書き込みのために StreamReader と StreamWriter を返す、待機する必要があるコルーチンです。

これは、ポート 80 で HTTP 接続を開くために使用できます。

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 80)

ssl=True パラメータを使用して SSL 接続を開くこともできます。これを使用して、ポート 443 で HTTPS 接続を開くことができます。

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 443)

3. HTTP リクエストの書き込み

開いた後、StreamWriter にクエリを書いて HTTP リクエストを作成できます。たとえば、HTTP バージョン 1.1 リクエストはプレーン テキストです。ファイル パス "/" をリクエストできます。これは次のようになります:

GET / HTTP/1.1
Host: www.google.com

重要なのは、各行の終わりにはキャリッジ リターンとライン フィード (\r\n)、そして最後の空白行。

Python 文字列として、これは次のようになります:

'GET / HTTP/1.1\r\n'
'Host: www.google.com\r\n'
'\r\n'

この文字列は、StreamWriter に書き込まれる前にバイトにエンコードする必要があります。これは、文字列自体に対して encode() メソッドを使用することで実現できます。デフォルトの「utf-8」エンコーディングで十分な場合があります。

...
# encode string as bytes
byte_data = string.encode()

その後、StreamWriter の write() メソッドを介してバイトをソケットに書き込むことができます。

...
# write query to socket
writer.write(byte_data)

リクエストを書いた後は、データのバイトが送信され、ソケットの準備ができるまで待つのが最善です。これは、drain() メソッドを通じて実現できます。これは待機する必要があるコルーチンです。

...
# wait for the socket to be ready.
await writer.drain()

4. HTTP 応答を読み取る

HTTP 要求を作成した後、応答を読み取ることができます。これはソケットの StreamReader を通じて実現できます。応答は、バイトの大きなブロックを読み取る read() メソッド、またはバイト行を読み取る readline() メソッドを使用して読み取ることができます。

一度に 1 行ずつ HTML データを送信するテキストベースの HTTP プロトコルを使用しているため、readline() メソッドを使用することをお勧めします。 readline() メソッドはコルーチンであるため、待機する必要があります。

...
# read one line of response
line_bytes = await reader.readline()

HTTP 1.1 応答は 2 つの部分で構成されます。空行で区切られたヘッダーと、その後に空行で終わる本文です。ヘッダーには、リクエストが成功したかどうか、および送信されるファイルの種類に関する情報が含まれ、本文には HTML Web ページなどのファイルのコンテンツが含まれます。

HTTP ヘッダーの最初の行には、サーバー上の要求されたページの HTTP ステータスが含まれています。各行はバイトから文字列にデコードする必要があります。

これは、バイト データに対して decode() メソッドを使用することで実現できます。繰り返しますが、デフォルトのエンコーディングは「utf_8」です。

...
# decode bytes into a string
line_data = line_bytes.decode()

5. HTTP 接続を閉じる

StreamWriter を閉じることでソケット接続を閉じることができます。これは、close() メソッドを呼び出すことで実現できます。

...
# close the connection
writer.close()

これはブロックされず、ソケットをすぐには閉じない可能性があります。 asyncio を使用して HTTP リクエストを作成し、レスポンスを読み取る方法がわかったので、Web ページのステータスをチェックする例をいくつか見てみましょう。

6. HTTP ステータスを順番にチェックする例

asyncio を使用して複数の Web サイトの HTTP ステータスをチェックするサンプルを開発できます。

この例では、まず、指定された URL のステータスを確認するコルーチンを開発します。次に、上位 10 の Web サイトごとにこのコルーチンを 1 回呼び出します。

まず、URL 文字列を受け入れて HTTP ステータスを返すコルーチンを定義できます。

# get the HTTP/S status of a webpage
async def get_status(url):
	# ...

URL はそのコンポーネント部分に解析される必要があります。 HTTP リクエストを行うときは、ホスト名とファイル パスが必要です。 SSL が必要かどうかを判断するには、URL スキーム (HTTP または HTTPS) も知る必要があります。

これは、urllib.parse.urlsplit() 関数を使用して実現できます。この関数は、URL 文字列を受け取り、すべての URL 要素の名前付きタプルを返します。

...
# split the url into components
url_parsed = urlsplit(url)

次に、URL スキームに基づいて HTTP 接続を開き、URL ホスト名を使用します。

...
# open the connection
if url_parsed.scheme == 'https':
    reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
else:
    reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)

次に、ホスト名とファイル パスを使用して HTTP GET リクエストを作成し、StreamWriter を使用してエンコードされたバイトをソケットに書き込みます。

...
# send GET request
query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
# write query to socket
writer.write(query.encode())
# wait for the bytes to be written to the socket
await writer.drain()

次に、HTTP 応答を読み取ります。必要なのは、HTTP ステータスを含む応答の最初の行だけです。

...
# read the single line response
response = await reader.readline()

その後、接続を閉じることができます。

...
# close the connection
writer.close()

最後に、サーバーから読み取ったバイト、リモートの末尾の空白をデコードして、HTTP ステータスを返すことができます。

...
# decode and strip white space
status = response.decode().strip()
# return the response
return status

これらを組み合わせると、完全な get_status() コルーチンが以下にリストされます。ホストにアクセスできない、応答が遅いなどのエラー処理はありません。これらの追加は、読者に優れた拡張機能を提供します。

# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status

次に、チェックしたい複数の Web ページまたは Web サイトに対して get_status() コルーチンを呼び出すことができます。この場合、世界のトップ 10 の Web ページのリストを定義します。

...
# list of top 10 websites to check
sites = ['https://www.google.com/',
    'https://www.youtube.com/',
    'https://www.facebook.com/',
    'https://twitter.com/',
    'https://www.instagram.com/',
    'https://www.baidu.com/',
    'https://www.wikipedia.org/',
    'https://yandex.ru/',
    'https://yahoo.com/',
    'https://www.whatsapp.com/'
    ]

然后我们可以使用我们的 get_status() 协程依次查询每个。在这种情况下，我们将在一个循环中按顺序这样做，并依次报告每个状态。

...
# check the status of all websites
for url in sites:
    # get the status for the url
    status = await get_status(url)
    # report the url and its status
    print(f'{url:30}:\t{status}')

在使用 asyncio 时，我们可以做得比顺序更好，但这提供了一个很好的起点，我们可以在以后进行改进。将它们结合在一起，main() 协程查询前 10 个网站的状态。

# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # check the status of all websites
    for url in sites:
        # get the status for the url
        status = await get_status(url)
        # report the url and its status
        print(f'{url:30}:\t{status}')

最后，我们可以创建 main() 协程并将其用作 asyncio 程序的入口点。

...
# run the asyncio program
asyncio.run(main())

将它们结合在一起，下面列出了完整的示例。

# SuperFastPython.com
# check the status of many webpages
import asyncio
from urllib.parse import urlsplit
 
# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status
 
# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # check the status of all websites
    for url in sites:
        # get the status for the url
        status = await get_status(url)
        # report the url and its status
        print(f'{url:30}:\t{status}')
 
# run the asyncio program
asyncio.run(main())

运行示例首先创建 main() 协程并将其用作程序的入口点。main() 协程运行，定义前 10 个网站的列表。然后顺序遍历网站列表。 main()协程挂起调用get_status()协程查询一个网站的状态。

get_status() 协程运行、解析 URL 并打开连接。它构造一个 HTTP GET 查询并将其写入主机。读取、解码并返回响应。main() 协程恢复并报告 URL 的 HTTP 状态。

对列表中的每个 URL 重复此操作。该程序大约需要 5.6 秒才能完成，或者平均每个 URL 大约需要半秒。这突出了我们如何使用 asyncio 来查询网页的 HTTP 状态。

尽管如此，它并没有充分利用 asyncio 来并发执行任务。

https://www.google.com/ : HTTP/1.1 200 OK
https://www.youtube.com/ : HTTP/1.1 200 OK
https://www.facebook.com/ : HTTP/1.1 302 Found
https://twitter.com/ : HTTP/1.1 200 OK
https://www.instagram.com/ : HTTP/1.1 200 OK
https://www.baidu.com/ : HTTP/1.1 200 OK
https://www.wikipedia.org/ : HTTP/1.1 200 OK
https://yandex.ru/ : HTTP/1.1 302 Moved temporarily
https://yahoo.com/ : HTTP/1.1 301 Moved Permanently
https://www.whatsapp.com/ : HTTP/1.1 302 Found

7. 并发查看网站状态示例

asyncio 的一个好处是我们可以同时执行许多协程。我们可以使用 asyncio.gather() 函数在 asyncio 中并发查询网站的状态。

此函数采用一个或多个协程，暂停执行提供的协程，并将每个协程的结果作为可迭代对象返回。然后我们可以遍历 URL 列表和可迭代的协程返回值并报告结果。

这可能是比上述方法更简单的方法。首先，我们可以创建一个协程列表。

...
# create all coroutine requests
coros = [get_status(url) for url in sites]

接下来，我们可以执行协程并使用 asyncio.gather() 获取可迭代的结果。

请注意，我们不能直接提供协程列表，而是必须将列表解压缩为单独的表达式，这些表达式作为位置参数提供给函数。

...
# execute all coroutines and wait
results = await asyncio.gather(*coros)

这将同时执行所有协程并检索它们的结果。然后我们可以遍历 URL 列表和返回状态并依次报告每个。

...
# process all results
for url, status in zip(sites, results):
    # report status
    print(f'{url:30}:\t{status}')

将它们结合在一起，下面列出了完整的示例。

# SuperFastPython.com
# check the status of many webpages
import asyncio
from urllib.parse import urlsplit
 
# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status
 
# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # create all coroutine requests
    coros = [get_status(url) for url in sites]
    # execute all coroutines and wait
    results = await asyncio.gather(*coros)
    # process all results
    for url, status in zip(sites, results):
        # report status
        print(f'{url:30}:\t{status}')
 
# run the asyncio program
asyncio.run(main())

运行该示例会像以前一样执行 main() 协程。在这种情况下，协程列表是在列表理解中创建的。

然后调用 asyncio.gather() 函数，传递协程并挂起 main() 协程，直到它们全部完成。协程执行，同时查询每个网站并返回它们的状态。

main() 协程恢复并接收可迭代的状态值。然后使用 zip() 内置函数遍历此可迭代对象和 URL 列表，并报告状态。

这突出了一种更简单的方法来同时执行协程并在所有任务完成后报告结果。它也比上面的顺序版本更快，在我的系统上完成大约 1.4 秒。

https://www.google.com/       :    HTTP/1.1 200 OK
https://www.youtube.com/      :    HTTP/1.1 200 OK
https://www.facebook.com/     :    HTTP/1.1 302 Found
https://twitter.com/          :    HTTP/1.1 200 OK
https://www.instagram.com/    :    HTTP/1.1 200 OK
https://www.baidu.com/        :    HTTP/1.1 200 OK
https://www.wikipedia.org/    :    HTTP/1.1 200 OK
https://yandex.ru/            :    HTTP/1.1 302 Moved temporarily
https://yahoo.com/            :    HTTP/1.1 301 Moved Permanently
https://www.whatsapp.com/     :    HTTP/1.1 302 Found

以上がPython Asyncio を使用して Web サイトのステータスチェックを実装する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。