Richieste HTTP in Python

Le richieste HTTP sono alla base di qualsiasi applicazione che comunica con servizi web, API REST o qualsiasi endpoint remoto. Python offre diversi strumenti per effettuare richieste HTTP, dal modulo standard urllib fino alla libreria di terze parti requests, passando per le soluzioni asincrone come httpx e aiohttp. In questo articolo esploreremo in profondità ciascuno di questi strumenti, analizzandone sintassi, casi d'uso, gestione degli errori e pattern avanzati.

Il modulo urllib della libreria standard

Il modulo urllib fa parte della libreria standard di Python e non richiede alcuna installazione aggiuntiva. È suddiviso in sotto-moduli: urllib.request per aprire URL, urllib.parse per manipolare URL e parametri, urllib.error per la gestione delle eccezioni e urllib.response per i tipi di risposta.

Richiesta GET di base

La forma più semplice di richiesta HTTP con urllib consiste nell'aprire un URL tramite urllib.request.urlopen().

import urllib.request

# apertura della connessione all'endpoint
with urllib.request.urlopen("https://httpbin.org/get") as response:
    # lettura del corpo della risposta come bytes
    body = response.read()
    # decodifica in stringa UTF-8
    text = body.decode("utf-8")
    print(text)

Il metodo urlopen() restituisce un oggetto di tipo http.client.HTTPResponse, che è anche un context manager. La chiamata a read() restituisce i byte grezzi della risposta; la decodifica va effettuata manualmente.

Accesso ai metadati della risposta

L'oggetto risposta espone attributi e metodi utili per ispezionare lo status code e gli header HTTP.

import urllib.request

with urllib.request.urlopen("https://httpbin.org/get") as response:
    # codice di stato HTTP (es. 200, 404)
    status_code = response.status
    # motivo testuale associato al codice
    reason = response.reason
    # dizionario degli header della risposta
    headers = dict(response.getheaders())

    print(f"Status: {status_code} {reason}")
    print(f"Content-Type: {headers.get('Content-Type')}")

Richiesta POST con dati nel corpo

Per inviare una richiesta POST occorre costruire un oggetto Request specificando i dati da inviare come bytes e impostando l'header Content-Type appropriato.

import urllib.request
import urllib.parse
import json

# preparazione del payload come dizionario Python
payload = {
    "username": "mario",
    "email": "mario@example.com"
}

# serializzazione in JSON e codifica in bytes
data = json.dumps(payload).encode("utf-8")

# costruzione della richiesta con metodo, dati e header
request = urllib.request.Request(
    url="https://httpbin.org/post",
    data=data,
    method="POST",
    headers={
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
)

with urllib.request.urlopen(request) as response:
    result = json.loads(response.read().decode("utf-8"))
    print(result["json"])  # corpo inviato, restituito da httpbin

Invio di dati form-encoded

Quando il server attende dati nel formato application/x-www-form-urlencoded, si utilizza urllib.parse.urlencode() per codificare il dizionario.

import urllib.request
import urllib.parse

# codifica dei parametri nel formato chiave=valore&chiave2=valore2
form_data = urllib.parse.urlencode({
    "grant_type": "client_credentials",
    "client_id": "abc123",
    "client_secret": "secret"
}).encode("utf-8")

request = urllib.request.Request(
    url="https://httpbin.org/post",
    data=form_data,
    method="POST",
    headers={"Content-Type": "application/x-www-form-urlencoded"}
)

with urllib.request.urlopen(request) as response:
    print(response.read().decode("utf-8"))

Aggiunta di parametri query string

I parametri GET vanno concatenati all'URL tramite urllib.parse.urlencode() e urllib.parse.urlunparse() o, più semplicemente, con manipolazione diretta della stringa.

import urllib.request
import urllib.parse

# costruzione della query string dai parametri
params = urllib.parse.urlencode({
    "q": "python http",
    "page": 1,
    "per_page": 10
})

# composizione dell'URL finale
base_url = "https://httpbin.org/get"
full_url = f"{base_url}?{params}"

with urllib.request.urlopen(full_url) as response:
    print(response.read().decode("utf-8"))

Gestione degli errori con urllib.error

Le eccezioni di urllib sono raggruppate nel sotto-modulo urllib.error. L'eccezione principale è HTTPError, che viene sollevata per codici di errore HTTP come 404 o 500, mentre URLError gestisce problemi di rete a livello più basso.

import urllib.request
import urllib.error

try:
    with urllib.request.urlopen("https://httpbin.org/status/404") as response:
        print(response.read())
except urllib.error.HTTPError as error:
    # errore HTTP con codice di stato
    print(f"Errore HTTP: {error.code} {error.reason}")
    # corpo dell'errore restituito dal server
    error_body = error.read().decode("utf-8")
    print(f"Corpo: {error_body}")
except urllib.error.URLError as error:
    # errore di rete (DNS, timeout, connessione rifiutata)
    print(f"Errore di rete: {error.reason}")

Impostazione del timeout

Il parametro timeout di urlopen() definisce il numero di secondi massimi di attesa per la risposta. Al superamento del timeout viene sollevata una socket.timeout, catturabile come urllib.error.URLError.

import urllib.request
import urllib.error

try:
    # timeout di 5 secondi
    with urllib.request.urlopen("https://httpbin.org/delay/10", timeout=5) as response:
        print(response.read())
except urllib.error.URLError as error:
    print(f"Timeout o errore: {error.reason}")

La libreria requests

La libreria requests è lo strumento più diffuso per le richieste HTTP in Python. Offre un'API ad alto livello, gestione automatica di encoding, sessioni, autenticazione e molto altro. Prima di utilizzarla occorre installarla.

pip install requests

Richiesta GET

La sintassi di requests è molto più concisa rispetto a urllib. Il metodo requests.get() restituisce un oggetto Response ricco di attributi.

import requests

# richiesta GET all'endpoint di test
response = requests.get("https://httpbin.org/get")

# codice di stato HTTP
print(response.status_code)
# header della risposta come dizionario
print(response.headers["Content-Type"])
# corpo come testo decodificato
print(response.text)
# corpo deserializzato da JSON
print(response.json())

Parametri query string

I parametri GET si passano come dizionario all'argomento params; la libreria si occupa automaticamente della codifica.

import requests

# parametri aggiunti automaticamente alla query string
params = {
    "search": "python requests",
    "page": 2,
    "limit": 20
}

response = requests.get("https://httpbin.org/get", params=params)
# URL finale con i parametri codificati
print(response.url)
print(response.json())

Richiesta POST con JSON

Il parametro json serializza automaticamente il dizionario e imposta l'header Content-Type: application/json.

import requests

# payload inviato come JSON
payload = {
    "title": "Articolo di test",
    "body": "Contenuto dell'articolo",
    "userId": 1
}

response = requests.post(
    "https://jsonplaceholder.typicode.com/posts",
    json=payload
)

print(response.status_code)  # 201 Created
print(response.json())

Invio di dati form-encoded

Passando un dizionario al parametro data, la libreria lo codifica come application/x-www-form-urlencoded.

import requests

# dati inviati come form-encoded
form_fields = {
    "username": "luigi",
    "password": "secret123"
}

response = requests.post("https://httpbin.org/post", data=form_fields)
print(response.json()["form"])

Upload di file

L'upload di file avviene tramite il parametro files. La libreria imposta automaticamente il Content-Type multipart.

import requests

# apertura del file in modalità binaria
with open("documento.pdf", "rb") as file_handle:
    # il campo "file" corrisponde al nome del campo nel form
    files = {"file": ("documento.pdf", file_handle, "application/pdf")}
    response = requests.post("https://httpbin.org/post", files=files)
    print(response.json()["files"])

Header personalizzati e autenticazione Bearer

Gli header si passano come dizionario al parametro headers. Per l'autenticazione con token JWT o API key si usa tipicamente l'header Authorization.

import requests

# token di accesso ottenuto in precedenza
access_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

custom_headers = {
    "Authorization": f"Bearer {access_token}",
    "Accept": "application/json",
    "X-Client-Version": "1.0.0"
}

response = requests.get(
    "https://httpbin.org/bearer",
    headers=custom_headers
)

print(response.json())

Autenticazione Basic e Digest

La libreria supporta i metodi di autenticazione standard tramite il parametro auth.

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# autenticazione Basic: credenziali in chiaro codificate in Base64
response_basic = requests.get(
    "https://httpbin.org/basic-auth/user/passwd",
    auth=HTTPBasicAuth("user", "passwd")
)
print(response_basic.status_code)

# autenticazione Digest: più sicura, usa challenge-response
response_digest = requests.get(
    "https://httpbin.org/digest-auth/auth/user/passwd",
    auth=HTTPDigestAuth("user", "passwd")
)
print(response_digest.status_code)

Gestione degli errori

Il metodo raise_for_status() solleva un'eccezione HTTPError automaticamente per qualsiasi codice 4xx o 5xx.

import requests
from requests.exceptions import HTTPError, ConnectionError, Timeout, RequestException

try:
    response = requests.get("https://httpbin.org/status/500", timeout=10)
    # solleva HTTPError se il codice di stato indica un errore
    response.raise_for_status()
    print(response.json())
except HTTPError as error:
    print(f"Errore HTTP: {error.response.status_code}")
except ConnectionError:
    # impossibile stabilire la connessione
    print("Errore di connessione")
except Timeout:
    # la richiesta ha superato il timeout
    print("Timeout della richiesta")
except RequestException as error:
    # qualsiasi altro errore di requests
    print(f"Errore generico: {error}")

Sessioni e riutilizzo della connessione

L'oggetto Session mantiene parametri comuni tra più richieste (header, cookie, autenticazione) e riutilizza le connessioni TCP sottostanti grazie al keep-alive, migliorando le prestazioni.

import requests

# la sessione persiste header, cookie e configurazione
with requests.Session() as session:
    # header comuni a tutte le richieste della sessione
    session.headers.update({
        "Authorization": "Bearer token123",
        "Accept": "application/json"
    })

    # prima richiesta: stabilisce la connessione
    response_one = session.get("https://httpbin.org/get")
    print(response_one.json())

    # seconda richiesta: riutilizza la connessione esistente
    response_two = session.get("https://httpbin.org/headers")
    print(response_two.json())

Cookie

I cookie ricevuti nella risposta vengono automaticamente memorizzati nella sessione e inviati nelle richieste successive. È possibile anche impostarli manualmente.

import requests

with requests.Session() as session:
    # simulazione di login che imposta un cookie di sessione
    login_response = session.post("https://httpbin.org/cookies/set/session_id/abc123")
    print(login_response.cookies)

    # la richiesta successiva invia automaticamente il cookie
    protected_response = session.get("https://httpbin.org/cookies")
    print(protected_response.json())

Timeout e retry con urllib3

È possibile configurare il retry automatico montando un HTTPAdapter con una Retry strategy dalla libreria urllib3, che è una dipendenza di requests.

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry() -> requests.Session:
    """Crea una sessione con retry automatico per errori di rete."""
    session = requests.Session()

    retry_strategy = Retry(
        total=3,              # numero massimo di tentativi
        backoff_factor=1,     # attesa esponenziale tra i tentativi
        status_forcelist=[429, 500, 502, 503, 504],  # codici da ritentare
        allowed_methods=["GET", "POST"]
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    # montaggio dell'adapter per HTTP e HTTPS
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    return session

session = create_session_with_retry()
response = session.get("https://httpbin.org/get", timeout=10)
print(response.status_code)

Streaming di grandi risposte

Per scaricare file di grandi dimensioni senza caricarli interamente in memoria si usa il parametro stream=True e si itera sul contenuto a blocchi.

import requests

url = "https://httpbin.org/bytes/1048576"  # 1 MB di dati casuali

with requests.get(url, stream=True) as response:
    response.raise_for_status()

    # scrittura in un file locale a blocchi di 8 KB
    with open("output.bin", "wb") as output_file:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:  # filtra i keep-alive chunk vuoti
                output_file.write(chunk)

print("Download completato")

Richieste asincrone con httpx

La libreria httpx offre un'API molto simile a requests ma con supporto nativo per il codice asincrono basato su asyncio. È la scelta ideale per applicazioni che devono effettuare molte richieste in parallelo.

pip install httpx

Client sincrono

httpx fornisce anche un client sincrono compatibile con l'API di requests, utile per sostituirlo senza refactoring profondo.

import httpx

# client sincrono, funziona come requests
with httpx.Client(timeout=10.0) as client:
    response = client.get("https://httpbin.org/get", params={"lang": "python"})
    print(response.status_code)
    print(response.json())

Client asincrono

Il client asincrono si utilizza all'interno di funzioni async con la parola chiave await.

import httpx
import asyncio

async def fetch_data(url: str) -> dict:
    """Effettua una richiesta GET asincrona e restituisce il JSON."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

async def main() -> None:
    data = await fetch_data("https://httpbin.org/get")
    print(data)

asyncio.run(main())

Richieste parallele con asyncio.gather()

La vera potenza del client asincrono emerge quando si devono effettuare molte richieste contemporaneamente. asyncio.gather() le esegue in parallelo senza bloccare il thread principale.

import httpx
import asyncio
from typing import List

async def fetch_post(client: httpx.AsyncClient, post_id: int) -> dict:
    """Scarica un singolo post dall'API di test."""
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

async def fetch_all_posts(post_ids: List[int]) -> List[dict]:
    """Scarica tutti i post in parallelo con un unico client condiviso."""
    async with httpx.AsyncClient(timeout=15.0) as client:
        # esecuzione di tutte le coroutine in parallelo
        tasks = [fetch_post(client, pid) for pid in post_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    # filtraggio dei risultati validi, escludendo le eccezioni
    return [r for r in results if isinstance(r, dict)]

async def main() -> None:
    posts = await fetch_all_posts(list(range(1, 21)))
    print(f"Scaricati {len(posts)} post")
    for post in posts[:3]:
        print(f"  [{post['id']}] {post['title']}")

asyncio.run(main())

Streaming asincrono

Anche lo streaming di grandi risposte è supportato in modalità asincrona con il metodo aiter_bytes().

import httpx
import asyncio

async def download_file(url: str, destination: str) -> None:
    """Scarica un file in modo asincrono e lo salva su disco."""
    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream("GET", url) as response:
            response.raise_for_status()

            with open(destination, "wb") as output_file:
                # iterazione asincrona sui chunk della risposta
                async for chunk in response.aiter_bytes(chunk_size=8192):
                    output_file.write(chunk)

asyncio.run(download_file("https://httpbin.org/bytes/65536", "output.bin"))
print("Download asincrono completato")

Richieste asincrone con aiohttp

aiohttp è una libreria asincrona matura, nata prima di httpx, molto usata in produzione. Rispetto ad httpx offre anche un server HTTP asincrono integrato.

pip install aiohttp

Richiesta GET di base

import aiohttp
import asyncio

async def get_data(url: str) -> dict:
    """Effettua una GET e restituisce il JSON della risposta."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            # verifica del codice di stato (solleva exception se 4xx/5xx)
            response.raise_for_status()
            return await response.json()

async def main() -> None:
    data = await get_data("https://httpbin.org/get")
    print(data)

asyncio.run(main())

Richiesta POST con JSON

import aiohttp
import asyncio

async def create_resource(payload: dict) -> dict:
    """Invia un POST con corpo JSON e restituisce la risposta."""
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://jsonplaceholder.typicode.com/posts",
            json=payload  # serializzazione e Content-Type automatici
        ) as response:
            response.raise_for_status()
            return await response.json()

async def main() -> None:
    new_post = await create_resource({
        "title": "Test aiohttp",
        "body": "Corpo del post",
        "userId": 42
    })
    print(new_post)

asyncio.run(main())

Richieste parallele con asyncio.gather()

import aiohttp
import asyncio
from typing import List

async def fetch_user(session: aiohttp.ClientSession, user_id: int) -> dict:
    """Scarica i dati di un singolo utente."""
    async with session.get(
        f"https://jsonplaceholder.typicode.com/users/{user_id}"
    ) as response:
        response.raise_for_status()
        return await response.json()

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    """Scarica tutti gli utenti condividendo la stessa ClientSession."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user(session, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    return [r for r in results if isinstance(r, dict)]

async def main() -> None:
    users = await fetch_all_users(list(range(1, 11)))
    for user in users:
        print(f"{user['id']}: {user['name']} ({user['email']})")

asyncio.run(main())

Timeout con aiohttp

aiohttp usa un oggetto ClientTimeout per configurare diversi tipi di timeout in modo granulare.

import aiohttp
import asyncio

async def get_with_timeout(url: str) -> None:
    # timeout totale di 5 secondi, di connessione di 2 secondi
    timeout_config = aiohttp.ClientTimeout(total=5, connect=2)

    try:
        async with aiohttp.ClientSession(timeout=timeout_config) as session:
            async with session.get(url) as response:
                print(await response.text())
    except aiohttp.ServerTimeoutError:
        print("Timeout lato server")
    except asyncio.TimeoutError:
        print("Timeout totale superato")
    except aiohttp.ClientError as error:
        print(f"Errore client: {error}")

asyncio.run(get_with_timeout("https://httpbin.org/delay/10"))

Pattern avanzati

Interceptor e middleware con requests

Gli Event Hooks di requests consentono di registrare callback che vengono eseguiti dopo ogni risposta, utili per logging centralizzato o per la gestione automatica del token refresh.

import requests
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def log_response(response: requests.Response, *args, **kwargs) -> None:
    """Hook eseguito dopo ogni risposta per registrare i dettagli."""
    duration_ms = response.elapsed.total_seconds() * 1000
    logger.info(
        "%s %s -> %d (%.0f ms)",
        response.request.method,
        response.url,
        response.status_code,
        duration_ms
    )

with requests.Session() as session:
    # registrazione dell'hook di risposta
    session.hooks["response"].append(log_response)

    session.get("https://httpbin.org/get")
    session.get("https://httpbin.org/delay/1")

Client con autenticazione OAuth2 e token refresh

Un pattern comune consiste nel wrappare il client in una classe che gestisce automaticamente il rinnovo del token alla scadenza.

import requests
import time

class OAuthClient:
    """Client HTTP con gestione automatica del token OAuth2."""

    def __init__(self, token_url: str, client_id: str, client_secret: str) -> None:
        self._token_url = token_url
        self._client_id = client_id
        self._client_secret = client_secret
        self._access_token: str | None = None
        self._token_expiry: float = 0.0
        self._session = requests.Session()

    def _is_token_expired(self) -> bool:
        """Verifica se il token è scaduto con un margine di 30 secondi."""
        return time.time() >= (self._token_expiry - 30)

    def _refresh_token(self) -> None:
        """Ottiene un nuovo access token tramite client credentials flow."""
        response = self._session.post(self._token_url, data={
            "grant_type": "client_credentials",
            "client_id": self._client_id,
            "client_secret": self._client_secret
        })
        response.raise_for_status()
        token_data = response.json()
        self._access_token = token_data["access_token"]
        # calcolo del timestamp di scadenza
        self._token_expiry = time.time() + token_data.get("expires_in", 3600)

    def get(self, url: str, **kwargs) -> requests.Response:
        """Effettua una GET rinnovando il token se necessario."""
        if self._is_token_expired():
            self._refresh_token()
        kwargs.setdefault("headers", {})
        kwargs["headers"]["Authorization"] = f"Bearer {self._access_token}"
        return self._session.get(url, **kwargs)

Rate limiting con time.sleep() e semafori asincroni

Quando si interagisce con API che impongono limiti di frequenza, è necessario controllare il ritmo delle richieste. In ambiente sincrono si usa time.sleep(), in ambiente asincrono si usa un asyncio.Semaphore.

import asyncio
import httpx
from typing import List

# semaforo che limita le connessioni simultanee a 5
rate_limiter = asyncio.Semaphore(5)

async def fetch_limited(client: httpx.AsyncClient, url: str) -> dict:
    """Effettua una richiesta rispettando il limite di concorrenza."""
    async with rate_limiter:
        response = await client.get(url)
        response.raise_for_status()
        # pausa tra le richieste per rispettare il rate limit dell'API
        await asyncio.sleep(0.2)
        return response.json()

async def fetch_batch(urls: List[str]) -> List[dict]:
    """Scarica un batch di URL con rate limiting integrato."""
    async with httpx.AsyncClient(timeout=15.0) as client:
        tasks = [fetch_limited(client, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if isinstance(r, dict)]

Paginazione automatica

Molte API REST restituiscono i risultati in pagine. Un generatore Python consente di iterare su tutte le pagine in modo trasparente.

import requests
from typing import Generator

def paginate(
    session: requests.Session,
    base_url: str,
    page_size: int = 10
) -> Generator[list, None, None]:
    """Generatore che itera su tutte le pagine di una API paginata."""
    page = 1

    while True:
        response = session.get(base_url, params={"page": page, "limit": page_size})
        response.raise_for_status()
        data = response.json()

        items = data.get("data") or data  # struttura variabile a seconda dell'API
        if not items:
            break  # nessun risultato: fine della paginazione

        yield items
        page += 1

        # alcune API indicano l'ultima pagina con un flag esplicito
        if not data.get("has_more", True):
            break

with requests.Session() as session:
    for page_items in paginate(session, "https://reqres.in/api/users", page_size=6):
        for user in page_items:
            print(f"{user.get('id')}: {user.get('first_name')} {user.get('last_name')}")

Test delle richieste HTTP

Testare il codice che effettua richieste HTTP richiede di sostituire le chiamate reali con risposte simulate. La libreria responses intercetta le chiamate di requests senza modificare il codice da testare.

pip install responses

import requests
import responses
import unittest

def get_user(user_id: int) -> dict:
    """Funzione da testare: recupera un utente dall'API."""
    response = requests.get(f"https://api.example.com/users/{user_id}")
    response.raise_for_status()
    return response.json()

class TestGetUser(unittest.TestCase):

    @responses.activate
    def test_successful_response(self) -> None:
        """Verifica che l'utente venga deserializzato correttamente."""
        # registrazione della risposta mock
        responses.add(
            method=responses.GET,
            url="https://api.example.com/users/1",
            json={"id": 1, "name": "Mario Rossi"},
            status=200
        )

        user = get_user(1)
        self.assertEqual(user["id"], 1)
        self.assertEqual(user["name"], "Mario Rossi")

    @responses.activate
    def test_not_found(self) -> None:
        """Verifica che un 404 sollevi HTTPError."""
        responses.add(
            method=responses.GET,
            url="https://api.example.com/users/999",
            status=404
        )

        with self.assertRaises(requests.HTTPError):
            get_user(999)

if __name__ == "__main__":
    unittest.main()

Confronto tra le librerie

La scelta della libreria dipende dalle esigenze del progetto. urllib è adatta quando non si vogliono dipendenze esterne e le richieste sono semplici. requests è la scelta standard per applicazioni sincrone grazie alla sua API intuitiva e alla vastità dell'ecosistema. httpx è preferibile quando si ha bisogno sia del client sincrono sia di quello asincrono con la stessa API, oppure quando si lavora con HTTP/2. aiohttp è la scelta consolidata per applicazioni interamente asincrone ad alto throughput, soprattutto se si ha già familiarità con il suo ecosistema.

In tutti i casi, adottare l'oggetto sessione o client, configurare timeout adeguati e gestire le eccezioni in modo granulare sono pratiche fondamentali per costruire applicazioni robuste e affidabili.