johanleroy/ENI-PythonAdvanced_03
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
Johan
2025-12-15 15:58:02 +01:00
commit f042f8e3e2
13 changed files with 1507 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/tp_asyncio/__init__.py Normal file
View File

126
src/tp_asyncio/async_downloader.py Normal file
View File

@@ -0,0 +1,126 @@
"""
Script pour télécharger une liste d'images de manière asynchrone.
Ce script utilise `aiohttp` et `aiofiles` pour effectuer des téléchargements
parallèles, ce qui est beaucoup plus rapide qu'une approche séquentielle.
Il inclut :
- L'utilisation de sessions `aiohttp` pour optimiser les performances réseau.
- Une gestion des erreurs robuste pour les opérations réseau et disque.
"""
import asyncio
import logging
import sys
import time
from pathlib import Path
from typing import List
import aiofiles
import aiohttp
from aiohttp import ClientError
# --- Configuration ---
# Une liste d'URLs d'images à télécharger
IMG_URLS: List[str] = [
"https://images.pexels.com/photos/842711/pexels-photo-842711.jpeg",
"https://images.pexels.com/photos/3408744/pexels-photo-3408744.jpeg",
"https://images.pexels.com/photos/3244513/pexels-photo-3244513.jpeg",
"https://images.pexels.com/photos/210186/pexels-photo-210186.jpeg",
"https://images.pexels.com/photos/1261728/pexels-photo-1261728.jpeg",
"https://images.pexels.com/photos/414144/pexels-photo-414144.jpeg",
"https://images.pexels.com/photos/110854/pexels-photo-110854.jpeg",
"https://images.pexels.com/photos/546819/pexels-photo-546819.jpeg",
"https://images.pexels.com/photos/1640777/pexels-photo-1640777.jpeg",
"https://images.pexels.com/photos/885880/pexels-photo-885880.jpeg",
"https://images.pexels.com/photos/5318967/pexels-photo-5318967.jpeg",
"https://images.pexels.com/photos/3464632/pexels-photo-3464632.jpeg",
"https://images.pexels.com/photos/2110951/pexels-photo-2110951.jpeg",
"https://images.pexels.com/photos/774909/pexels-photo-774909.jpeg",
"https://images.pexels.com/photos/168927/pexels-photo-168927.jpeg",
]
# Le répertoire de sortie pour les images téléchargées
OUTPUT_DIR: Path = Path("../../images")
# --- Fonctions ---
def setup_logging() -> None:
"""
Configure le système de logging pour le script.
"""
# Sous un autre système d'exploitation que Windows, il est possible d'utiliser une blibliothèque de logging asynchrone telle que aiologger
# Lorsqu'il existe notamment des Handler de type fichiers, cela évite de ralentir la boucle asyncio
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[
logging.StreamHandler(sys.stdout),
]
)
async def download_and_save(
session: aiohttp.ClientSession, url: str, output_dir: Path
) -> None:
"""
Télécharge une image et la sauvegarde de manière asynchrone.
Paramètres:
session (aiohttp.ClientSession): la session client `aiohttp` à utiliser.
url (str): l'URL de l'image à télécharger.
output_dir (Path): le répertoire où sauvegarder l'image.
"""
# TODO : votre code ici. Similaire à la version synchrone, mais avec `async`/`await`.
# 1. Utilisez un bloc `try...except` pour gérer les erreurs (ClientError, asyncio.TimeoutError, IOError).
#
# 2. Dans le bloc `try` :
# a. Extrayez le nom du fichier et construisez le chemin de sauvegarde (comme avant).
# b. Logguez le début du téléchargement.
# c. Utilisez `async with session.get(url) as response:` pour effectuer la requête.
# d. Vérifiez le statut de la réponse avec `response.raise_for_status()`.
# e. Lisez le contenu de la réponse de manière asynchrone : `data = await response.read()`.
# f. Utilisez `async with aiofiles.open(path, "wb") as f:` pour ouvrir le fichier
# sans bloquer la boucle événementielle.
# g. Écrivez les données dans le fichier : `await f.write(data)`.
# h. Logguez la confirmation de la sauvegarde.
#
# 3. Dans les blocs `except`, logguez les erreurs spécifiques.
async def main() -> None:
"""
Fonction principale asynchrone.
Orchestre la création du répertoire de sortie et le lancement
des tâches de téléchargement en parallèle.
"""
setup_logging()
logging.info("Début du script de téléchargement asynchrone.")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
logging.info(f"Le répertoire de sortie est : {OUTPUT_DIR}")
# TODO : votre code ici. C'est ici que la magie de l'asynchronisme opère.
# 1. Créez une session client avec `async with aiohttp.ClientSession() as session:`.
#
# 2. À l'intérieur du `async with`, créez une liste de "tâches". Chaque tâche
# est un appel à votre coroutine `download_and_save`.
# Astuce : une compréhension de liste est parfaite pour ça.
# Exemple : `tasks = [download_and_save(session, url, OUTPUT_DIR) for url in IMG_URLS]`
#
# 3. Utilisez `asyncio.gather()` pour lancer toutes les tâches en parallèle et
# attendre qu'elles se terminent toutes.
# La syntaxe est : `await asyncio.gather(*tasks)`
# (L'étoile `*` dépaquette la liste de tâches en arguments pour `gather`).
if __name__ == "__main__":
# `asyncio.run()` est la manière moderne de lancer une application asyncio
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nScript interrompu par l'utilisateur.")

119
src/tp_asyncio/sync_downloader.py Normal file
View File

@@ -0,0 +1,119 @@
"""
Script pour télécharger une liste d'images de manière séquentielle.
Ce script inclut :
- Un système de logging pour suivre les opérations.
- Une gestion des erreurs robuste.
- L'utilisation de sessions `requests` pour optimiser les performances réseau.
"""
import sys
import logging
import time
from pathlib import Path
from typing import List
import requests
from requests.exceptions import RequestException
# --- Configuration ---
# Une liste d'URLs d'images à télécharger.
IMG_URLS: List[str] = [
"https://images.pexels.com/photos/842711/pexels-photo-842711.jpeg",
"https://images.pexels.com/photos/3408744/pexels-photo-3408744.jpeg",
"https://images.pexels.com/photos/3244513/pexels-photo-3244513.jpeg",
"https://images.pexels.com/photos/210186/pexels-photo-210186.jpeg",
"https://images.pexels.com/photos/1261728/pexels-photo-1261728.jpeg",
"https://images.pexels.com/photos/414144/pexels-photo-414144.jpeg",
"https://images.pexels.com/photos/110854/pexels-photo-110854.jpeg",
"https://images.pexels.com/photos/546819/pexels-photo-546819.jpeg",
"https://images.pexels.com/photos/1640777/pexels-photo-1640777.jpeg",
"https://images.pexels.com/photos/885880/pexels-photo-885880.jpeg",
"https://images.pexels.com/photos/5318967/pexels-photo-5318967.jpeg",
"https://images.pexels.com/photos/3464632/pexels-photo-3464632.jpeg",
"https://images.pexels.com/photos/2110951/pexels-photo-2110951.jpeg",
"https://images.pexels.com/photos/774909/pexels-photo-774909.jpeg",
"https://images.pexels.com/photos/168927/pexels-photo-168927.jpeg",
]
# Le répertoire de sortie pour les images téléchargées.
OUTPUT_DIR: Path = Path("../../images")
# --- Fonctions ---
def setup_logging() -> None:
"""
Configure le système de logging pour le script.
"""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[
logging.StreamHandler(sys.stdout),
]
)
def download_and_save(session: requests.Session, url: str, output_dir: Path) -> None:
"""
Télécharge une image depuis une URL et la sauvegarde dans le répertoire spécifié.
Paramètres:
session (requests.Session): La session `requests` à utiliser pour la requête.
url (str): L'URL de l'image à télécharger.
output_dir (Path): Le répertoire où sauvegarder l'image.
"""
# TODO : votre code ici. Suivez ces étapes :
# 1. Utilisez un bloc `try...except` pour gérer les erreurs potentielles
# (RequestException, IOError).
#
# 2. Dans le bloc `try` :
# a. Extrayez le nom du fichier à partir de l'URL.
# Astuce : `url.split("/")[-1]` est une méthode simple.
# b. Construisez le chemin de sauvegarde complet en utilisant `pathlib`.
# Exemple : `output_dir / nom_du_fichier`
# c. Logguez le début du téléchargement.
# d. Effectuez une requête GET avec `session.get(url)`. Pensez à ajouter un timeout.
# e. Vérifiez que la requête a réussi avec `response.raise_for_status()`.
# f. Ouvrez le fichier de destination en mode écriture binaire ('wb')
# en utilisant un `with open(...)`.
# g. Écrivez le contenu de la réponse (`response.content`) dans le fichier.
# h. Logguez la confirmation de la sauvegarde.
#
# 3. Dans les blocs `except`, logguez l'erreur de manière descriptive
# (ex: `logging.error(...)`).
def main() -> None:
"""
Fonction principale du script.
Orchestre la création du répertoire de sortie et le téléchargement
séquentiel des images.
"""
setup_logging()
logging.info("Début du script de téléchargement.")
# Crée le répertoire de sortie s'il n'existe pas.
# `parents=True` crée les répertoires parents si nécessaire.
# `exist_ok=True` évite une erreur si le répertoire existe déjà.
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
logging.info(f"Le répertoire de sortie est : {OUTPUT_DIR}")
start_time = time.perf_counter()
# TODO : votre code ici. Suivez ces étapes :
# 1. Utilisez un gestionnaire de contexte `with requests.Session() as session:`
# pour créer une session qui sera utilisée pour toutes les requêtes.
# Ceci optimise les connexions réseau.
#
# 2. À l'intérieur du `with`, faites une boucle `for` sur la liste `IMG_URLS`.
#
# 3. Dans la boucle, appelez la fonction `download_and_save()` pour chaque URL,
# en lui passant la session, l'URL et le répertoire de sortie.
if __name__ == "__main__":
main()