communauté Apprendre Bibliothèque d'outils Loisirs
recherche
Français
Maison > développement back-end > Tutoriel Python > Comment implémenter la vérification de l'état du site Web à l'aide de Python Asyncio

Comment implémenter la vérification de l'état du site Web à l'aide de Python Asyncio

PHPz
Libérer： 2023-04-21 14:10:18
avant
825 Les gens l'ont consulté

Nous pouvons utiliser asyncio pour interroger l'état HTTP d'un site Web en ouvrant un flux et en écrivant et en lisant des requêtes et des réponses HTTP.

Ensuite, nous pouvons utiliser asyncio pour interroger simultanément l'état de plusieurs sites Web et même rapporter les résultats de manière dynamique.

1. Comment utiliser Asyncio pour vérifier l'état HTTP

Le module asyncio prend en charge l'ouverture de connexions socket et la lecture et l'écriture de données via des flux. Nous pouvons utiliser cette fonctionnalité pour vérifier l'état d'une page Web.

Cela peut impliquer quatre étapes, elles sont :

  • Ouvrir une connexion

  • Rédiger une demande

  • Lire la réponse

  • Fermer la connexion

2. Ouvrir la connexion HTTP

Les connexions peuvent être ouvertes dans asyncio à l'aide de la fonction asyncio.open_connection(). Parmi de nombreux paramètres, la fonction prend un nom d'hôte sous forme de chaîne et un numéro de port entier.

Il s'agit d'une coroutine incontournable qui renvoie un StreamReader et un StreamWriter pour la lecture et l'écriture à l'aide du socket.

Cela peut être utilisé pour ouvrir une connexion HTTP sur le port 80. 

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 80)
Copier après la connexion

Nous pouvons également ouvrir une connexion SSL en utilisant le paramètre ssl=True. Cela peut être utilisé pour ouvrir une connexion HTTPS sur le port 443. 

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 443)
Copier après la connexion

3. Écrire une requête HTTP

Après l'ouverture, nous pouvons écrire des requêtes sur StreamWriter pour faire des requêtes HTTP. Par exemple, les requêtes HTTP version 1.1 sont en texte brut. Nous pouvons demander le chemin du fichier "/", qui peut ressembler à ceci : 

GET / HTTP/1.1
Host: www.google.com
Copier après la connexion

Surtout, chaque ligne doit avoir un retour chariot et un saut de ligne (rn) à la fin, et une ligne vide à la fin.

En tant que chaîne Python, cela pourrait ressembler à ceci : 

'GET / HTTP/1.1\r\n'
'Host: www.google.com\r\n'
'\r\n'
Copier après la connexion

Cette chaîne doit être codée en octets avant d'être écrite dans StreamWriter. Ceci peut être réalisé en utilisant la méthode encode() sur la chaîne elle-même. L'encodage par défaut "utf-8" peut être suffisant. 

...
# encode string as bytes
byte_data = string.encode()
Copier après la connexion

Les octets peuvent ensuite être écrits sur le socket via la méthode write() de StreamWriter. 

...
# write query to socket
writer.write(byte_data)
Copier après la connexion

Après avoir rédigé une requête, il est préférable d'attendre que les octets de données soient envoyés et que le socket soit prêt. Ceci peut être réalisé grâce à la méthode drain(). C'est une coroutine qui doit attendre. 

...
# wait for the socket to be ready.
await writer.drain()
Copier après la connexion

4. Lire la réponse HTTP

Après avoir fait une requête HTTP, nous pouvons lire la réponse. Ceci peut être réalisé via le StreamReader du socket. La réponse peut être lue à l'aide de la méthode read(), qui lit un gros bloc d'octets, ou de la méthode readline(), qui lit une ligne d'octets.

Nous préférerons peut-être la méthode readline() car nous utilisons le protocole HTTP basé sur le texte, qui envoie les données HTML une ligne à la fois. La méthode readline() est une coroutine et doit attendre. 

...
# read one line of response
line_bytes = await reader.readline()
Copier après la connexion

Les réponses HTTP 1.1 se composent de deux parties, un en-tête séparé par une ligne vide, suivi d'un corps terminé par une ligne vide. L'en-tête contient des informations indiquant si la demande a réussi et quel type de fichier sera envoyé, et le corps contient le contenu du fichier, comme une page Web HTML.

La première ligne de l'en-tête HTTP contient le statut HTTP de la page demandée sur le serveur. Chaque ligne doit être décodée des octets en chaîne.

Cela peut être réalisé en utilisant la méthode decode() sur les données d'octets. Encore une fois, l'encodage par défaut est "utf_8". 

...
# decode bytes into a string
line_data = line_bytes.decode()
Copier après la connexion

5. Fermez la connexion HTTP

Nous pouvons fermer la connexion socket en fermant le StreamWriter. Ceci peut être réalisé en appelant la méthode close(). 

...
# close the connection
writer.close()
Copier après la connexion
Copier après la connexion

Cela ne bloque pas et peut ne pas fermer la prise immédiatement. Maintenant que nous savons comment effectuer des requêtes HTTP et lire des réponses à l'aide d'asyncio, examinons quelques exemples de vérification de l'état d'une page Web.

6. Exemple de vérification séquentielle du statut HTTP

Nous pouvons développer un exemple pour vérifier le statut HTTP de plusieurs sites Web à l'aide d'asyncio.

Dans cet exemple, nous allons d'abord développer une coroutine pour vérifier l'état d'une URL donnée. Nous appellerons ensuite cette coroutine une fois pour chacun des 10 meilleurs sites Web.

Tout d'abord, nous pouvons définir une coroutine qui acceptera une chaîne d'URL et renverra le statut HTTP. 

# get the HTTP/S status of a webpage
async def get_status(url):
	# ...
Copier après la connexion

Doit analyser l'URL dans ses composants. Nous avons besoin du nom d'hôte et du chemin du fichier lors d'une requête HTTP. Nous devons également connaître le schéma d'URL (HTTP ou HTTPS) pour déterminer si SSL est requis.

Cela peut être réalisé en utilisant la fonction urllib.parse.urlsplit(), qui accepte une chaîne d'URL et renvoie un tuple nommé de tous les éléments d'URL. 

...
# split the url into components
url_parsed = urlsplit(url)
Copier après la connexion

Ensuite, nous pouvons ouvrir une connexion HTTP basée sur le schéma d'URL et utiliser le nom d'hôte de l'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)
Copier après la connexion

Ensuite, nous pouvons créer une requête HTTP GET en utilisant le nom d'hôte et le chemin du fichier et utiliser un StreamWriter pour écrire les octets codés sur le socket. 

...
# 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()
Copier après la connexion

Ensuite, nous pouvons lire la réponse HTTP. Nous n'avons besoin que de la première ligne de la réponse contenant le statut HTTP. 

...
# read the single line response
response = await reader.readline()
Copier après la connexion

La connexion peut alors être fermée. 

...
# close the connection
writer.close()
Copier après la connexion
Copier après la connexion

Enfin, nous pouvons décoder les octets lus sur le serveur, les espaces de fin à distance et renvoyer l'état HTTP. 

...
# decode and strip white space
status = response.decode().strip()
# return the response
return status
Copier après la connexion

Pour tout relier, la coroutine complète get_status() est répertoriée ci-dessous. Il n'a aucune gestion des erreurs, telles qu'un hôte inaccessible ou une réponse lente. Ces ajouts fourniront une belle extension aux lecteurs. 

# 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
Copier après la connexion

Ensuite, nous pouvons appeler la coroutine get_status() pour plusieurs pages Web ou sites Web que nous souhaitons vérifier. Dans ce cas, nous définirons une liste des 10 meilleures pages Web au monde. 

...
# 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/'
    ]
Copier après la connexion

然后我们可以使用我们的 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}')
Copier après la connexion

在使用 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}')
Copier après la connexion

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

...
# run the asyncio program
asyncio.run(main())
Copier après la connexion

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

# 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())
Copier après la connexion

运行示例首先创建 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]
Copier après la connexion

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

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

...
# execute all coroutines and wait
results = await asyncio.gather(*coros)
Copier après la connexion

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

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

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

# 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())
Copier après la connexion

运行该示例会像以前一样执行 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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!

Étiquettes associées：
python asyncio
source：yisu.com
Article précédent：Comment supprimer ou déplacer des images spécifiées par lots avec Python ? Article suivant：Chaînes Python : utilisation courante et analyse du code source des chaînes F
Déclaration de ce site Web
Le contenu de cet article est volontairement contribué par les internautes et les droits d'auteur appartiennent à l'auteur original. Ce site n'assume aucune responsabilité légale correspondante. Si vous trouvez un contenu suspecté de plagiat ou de contrefaçon, veuillez contacter admin@php.cn
Derniers articles par auteur
Derniers numéros
Python/MySQL ne peut pas conserver correctement les données entières Aucun code n'est requis ici. Je souhaite enregistrer un numéro très long car je crée un je...
Depuis 2024-04-04 19:09:44
0
1
367
En utilisant Selenium, vous voulez cliquer et définir une URL en classe J'ai besoin d'un autre conseil aujourd'hui. J'essaie de créer du code Python/Selenium et l...
Depuis 2024-04-04 14:14:44
0
1
3492
Selenium + Python - inspecter l'image viaexecute_script Je dois vérifier qu'une image est affichée sur la page en utilisant du sélénium en python....
Depuis 2024-04-03 09:32:15
0
1
375
Comment conserver les X premières lignes et supprimer les lignes du tableau J'ai une grande table avec des millions d'enregistrements dans MySQLincident_archive, je s...
Depuis 2024-04-01 18:32:54
0
1
347
Comment récupérer un texte Google Weather spécifique à l'aide de BeautifulSoup ? Comment trouver le texte du cours « New York City, USA » en Python à l'aide de BeautifulSo...
Depuis 2024-04-01 14:06:14
0
1
308
Rubriques connexes
Plus>
Recommandations populaires
Tutoriels populaires
Plus>
Tutoriels associés
Recommandations populaires
Derniers cours
Derniers téléchargements
Plus>
effets Web
Code source du site Web
Matériel du site Web
Modèle frontal