Section 11.3.4 : Consommer une API avec Python — Cas pratiques
🎯 Objectif pédagogique
Construire des scripts Python robustes pour interagir avec des APIs réelles. Vous serez capable d'enchaîner des appels API, gérer les erreurs, paginer les résultats, et transformer les réponses en données exploitables.
De la théorie à la pratique
Marc connaît la théorie REST. Maintenant il construit de vrais scripts : récupérer la météo, les offres d'emploi, les statistiques GitHub — tout ce qui connecte son Job Tracker au monde réel.
Pattern de base — Client API
import requests
import os
from dotenv import load_dotenv
load_dotenv()
class APIClient:
"""Client HTTP réutilisable."""
def __init__(self, base_url, api_key=None):
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
if api_key:
self.session.headers['Authorization'] = f'Bearer \{api_key\}'
self.session.headers['Content-Type'] = 'application/json'
def get(self, endpoint, params=None):
url = f"\{self.base_url\}/\{endpoint.lstrip('/')\}"
response = self.session.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json()
def post(self, endpoint, data=None):
url = f"\{self.base_url\}/\{endpoint.lstrip('/')\}"
response = self.session.post(url, json=data, timeout=30)
response.raise_for_status()
return response.json()
Cas 1 : Météo (Open Meteo — sans clé API)
import requests
def get_weather(city_lat, city_lon, city_name=""):
"""Récupère la météo actuelle."""
response = requests.get(
"https://api.open-meteo.com/v1/forecast",
params=\{
"latitude": city_lat,
"longitude": city_lon,
"current_weather": True,
"timezone": "Europe/Paris"
\},
timeout=10
)
response.raise_for_status()
data = response.json()
weather = data['current_weather']
print(f"🌤️ Météo \{city_name\}: \{weather['temperature']\}°C, vent \{weather['windspeed']\} km/h")
return weather
# Paris
get_weather(48.86, 2.35, "Paris")
Cas 2 : GitHub — Analyser un profil
import requests
import pandas as pd
def analyze_github_profile(username):
"""Analyse un profil GitHub."""
# Infos du profil
user = requests.get(f"https://api.github.com/users/\{username\}", timeout=10).json()
# Repos (avec pagination)
repos = []
page = 1
while True:
response = requests.get(
f"https://api.github.com/users/\{username\}/repos",
params=\{'page': page, 'per_page': 100, 'sort': 'updated'\},
timeout=10
)
batch = response.json()
if not batch:
break
repos.extend(batch)
page += 1
df = pd.DataFrame(repos)
print(f"👤 \{user.get('name', username)\} (@\{username\})")
print(f"📊 \{user['public_repos']\} repos publics | \{user['followers']\} followers")
print(f"⭐ Total stars: \{df['stargazers_count'].sum()\}")
print(f"🔤 Langages: \{df['language'].dropna().value_counts().head(5).to_dict()\}")
return df
# Exemple
df_repos = analyze_github_profile("torvalds")
Cas 3 : Enchaîner des APIs
import requests
import json
def job_search_enriched(query, location="Paris"):
"""Recherche d'emploi enrichie : offres + météo + stats."""
results = \{\}
# 1. Météo du lieu
geo = requests.get(
"https://geocoding-api.open-meteo.com/v1/search",
params=\{"name": location, "count": 1\},
timeout=10
).json()
if geo.get('results'):
city = geo['results'][0]
weather = requests.get(
"https://api.open-meteo.com/v1/forecast",
params=\{
"latitude": city['latitude'],
"longitude": city['longitude'],
"current_weather": True
\},
timeout=10
).json()
results['weather'] = weather['current_weather']
results['location'] = f"\{city['name']\}, \{city.get('country', '')\}"
# 2. On pourrait ajouter d'autres APIs ici
# (API emploi, statistiques, etc.)
print(json.dumps(results, indent=2, ensure_ascii=False))
return results
job_search_enriched("data analyst", "Lyon")
Gestion d'erreurs robuste
import requests
from requests.exceptions import RequestException, Timeout, HTTPError
def safe_api_call(url, params=None, retries=3):
"""Appel API avec gestion complète des erreurs."""
for attempt in range(retries):
try:
response = requests.get(url, params=params, timeout=15)
response.raise_for_status()
return response.json()
except Timeout:
print(f"⏳ Timeout (tentative \{attempt+1\}/\{retries\})")
except HTTPError as e:
if e.response.status_code == 429:
wait = int(e.response.headers.get('Retry-After', 30))
print(f"⚠️ Rate limited. Attente \{wait\}s...")
import time; time.sleep(wait)
elif e.response.status_code >= 500:
print(f"🔥 Erreur serveur \{e.response.status_code\}")
else:
raise # 4xx = erreur client, pas de retry
except RequestException as e:
print(f"🌐 Erreur réseau : \{e\}")
return None # Toutes les tentatives échouées
Session vs requests.get()
Utilisez requests.Session() quand vous faites plusieurs appels au même serveur. La session réutilise la connexion TCP (plus rapide), garde les cookies, et permet de configurer les headers une seule fois.
🏋️ Exercice pratique (25 minutes)
# Construisez un script qui :
# 1. Récupère la météo de 3 villes (Paris, Lyon, Marseille)
# 2. Analyse un profil GitHub
# 3. Combine les résultats dans un DataFrame pandas
# 4. Sauvegarde en CSV
# Bonus : ajoutez la gestion d'erreurs et le rate limiting
Pourquoi utiliser response.raise_for_status() après un appel API ?
Pourquoi paginer les résultats d'une API au lieu de tout demander d'un coup ?