Gérer les réponses et les erreurs d’API

L’objet de réponse

import requests

response = requests.get('https://api.example.com/data')

# Key attributes of the response
print(response.status_code)       # e.g. 200
print(response.headers)           # dict of response headers
print(response.headers.get('Content-Type'))  # 'application/json'
print(response.text)              # raw response body as string
print(response.content)           # raw bytes

Analyser du JSON avec response.json()

import requests

response = requests.get(
    'https://api.example.com/users/42',
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

# Parse JSON body
user = response.json()

# Access fields safely with .get()
name = user.get('name', 'Unknown')
email = user.get('email', '')
roles = user.get('roles', [])

print(f'User: {name} ({email})')
print(f'Roles: {roles}')

Vérifier le code d’état avant l’analyse

import requests

response = requests.post(
    'https://api.example.com/tasks',
    json={'title': 'Write report'},
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

if response.status_code == 201:
    task = response.json()
    print('Task created, ID:', task['id'])
elif response.status_code == 400:
    error = response.json()
    print('Validation error:', error.get('message'))
elif response.status_code == 401:
    print('Auth failed — check your token')
else:
    print(f'Unexpected status {response.status_code}: {response.text[:200]}')

raise_for_status() — Lever automatiquement les erreurs

import requests
from requests.exceptions import HTTPError

try:
    response = requests.get(
        'https://api.example.com/users/9999',
        headers={'Authorization': 'Bearer YOUR_KEY'}
    )
    response.raise_for_status()  # raises if status >= 400
    user = response.json()
    print('Found user:', user['name'])

except HTTPError as e:
    print(f'HTTP error: {e.response.status_code}')
    print('Details:', e.response.text[:300])

Gérer les erreurs de décodage JSON

import requests
import json

response = requests.get(
    'https://api.example.com/report',
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

try:
    data = response.json()
except json.JSONDecodeError as e:
    print(f'Response is not valid JSON: {e}')
    print('Content-Type:', response.headers.get('Content-Type'))
    print('First 200 chars:', response.text[:200])
    # Decide: is this an HTML error page? A CSV file?
    data = None

if data is None:
    print('Falling back to text processing')

ConnectionError — Problèmes de réseau

import requests
from requests.exceptions import ConnectionError

try:
    response = requests.get('https://api.example.com/data')
    data = response.json()
except ConnectionError as e:
    print('Cannot reach server. Possible causes:')
    print('- DNS failure (bad hostname)')
    print('- Server is down')
    print('- No internet connection')
    print('- Firewall blocking the port')
    print(f'Error detail: {e}')
    # Consider: queue the request for retry when connectivity returns

Délai d’attente — Éviter que les agents restent bloqués

import requests
from requests.exceptions import Timeout

try:
    response = requests.get(
        'https://api.example.com/slow-endpoint',
        headers={'Authorization': 'Bearer YOUR_KEY'},
        timeout=(5, 30)  # 5s to connect, 30s to read
    )
    data = response.json()
except Timeout:
    print('Request timed out after 30 seconds')
    print('Options: retry, use cached result, or alert operator')

Gestion complète des exceptions

import requests
import json
from requests.exceptions import (
    ConnectionError, Timeout, HTTPError, RequestException
)

def safe_api_call(url, headers):
    try:
        r = requests.get(url, headers=headers, timeout=(5, 30))
        r.raise_for_status()
        return r.json()
    except Timeout:
        print('ERROR: Request timed out')
    except ConnectionError:
        print('ERROR: Cannot reach server')
    except HTTPError as e:
        print(f'ERROR: HTTP {e.response.status_code}')
        try:
            print('API error:', e.response.json().get('message'))
        except json.JSONDecodeError:
            print('Non-JSON error body')
    except RequestException as e:
        print(f'ERROR: Unexpected request error: {e}')
    return None

Consigner les réponses à des fins de débogage

import logging
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('agent.api')

def logged_request(method, url, **kwargs):
    logger.info(f'-> {method.upper()} {url}')
    response = requests.request(method, url, **kwargs)
    logger.info(
        f'<- {response.status_code} '
        f'({len(response.content)} bytes) '
        f'{response.elapsed.total_seconds():.2f}s'
    )
    if response.status_code >= 400:
        logger.error(f'Error body: {response.text[:500]}')
    return response

Gérer les réponses paginées

import requests

def get_all_items(base_url, headers):
    all_items = []
    url = f'{base_url}/items?page=1&limit=100'

    while url:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        data = response.json()

        all_items.extend(data.get('items', []))

        # Follow 'next' link if present
        url = data.get('next_page_url')  # None stops the loop

        print(f'Fetched {len(all_items)} items so far...')

    print(f'Total: {len(all_items)} items')
    return all_items

Diffuser les réponses volumineuses

import requests

response = requests.get(
    'https://api.example.com/large-report',
    headers={'Authorization': 'Bearer YOUR_KEY'},
    stream=True
)

response.raise_for_status()

# Write streamed content to file
with open('report.json', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            f.write(chunk)

print('Download complete')

# For streaming JSON lines (NDJSON):
for line in response.iter_lines():
    if line:
        import json
        record = json.loads(line)
        print(record)

Vérification rapide : raise_for_status

Récapitulatif de la gestion des réponses