Cours de Python / Collecter une API JSON et mettre à jour une base SQLite

Python

Date de création 2026-11-12 Dernière modification 2026-11-12

Pour collecter une API sans multiplier les mêmes enregistrements à chaque exécution, validez toute la réponse puis enregistrez les données à l’aide d’un identifiant externe stable. Ce projet associe une requête JSON à une transaction SQLite : un identifiant existant reçoit le nouveau nom et le nouveau montant, au lieu de créer une seconde dépense.

Choisir une identité et une règle de mise à jour

Une API est une interface de programmation qui permet à un logiciel d’obtenir des données d’un autre service. Ici, un simple fichier JSON servi sur votre machine représente sa réponse. Chaque objet doit fournir id, name et amount. Un nom et un montant identiques ne suffisent pas à prouver que deux dépenses sont la même opération : il faut un contrat selon lequel l’identifiant externe conserve ce sens.

Le programme effectue un upsert : insertion si l’identifiant est nouveau, mise à jour en cas de conflit sur cet identifiant. Si un même ID apparaît plusieurs fois dans une réponse valide, sa dernière occurrence gagne. Si votre application doit refuser ces répétitions, ajoutez plutôt un contrôle des doublons avant toute écriture. La documentation SQLite sur UPSERT précise le rôle de la contrainte d’unicité dans cette opération.

La réponse doit être une liste de 10 000 éléments au maximum, obtenue dans la limite d’un million d’octets. Chaque ligne doit être un objet ; son identifiant et son nom doivent être des chaînes non vides après contrôle des espaces, et son montant un entier de 0 à 1 000 000 000 000 inclus. Les booléens et les chaînes numériques sont refusés. Le contrôle utilise strip() pour détecter une chaîne vide, mais les valeurs enregistrées gardent leurs espaces : "e1" et " e1 " restent donc deux clés différentes.

Toutes les lignes sont contrôlées avant l’ouverture de la base. Une seule ligne invalide empêche de commencer l’enregistrement de la réponse. Les écritures locales sont ensuite regroupées dans une transaction ; cela ne rend pas la requête réseau et la base locale atomiques ensemble.

Préparer une réponse locale et deux terminaux

Utilisez Python 3.12 ou une version ultérieure : l’argument autocommit=False utilisé ci-dessous a été ajouté dans Python 3.12. Le module sqlite3 doit être disponible dans votre installation. Aucun paquet HTTP externe n’est requis. Créez un nouveau dossier d’exercice afin de ne pas réutiliser une base d’un autre projet.

Enregistrez ce contenu en UTF-8 dans expenses.json :

[{"id":"e1","name":"coffee","amount":3500},{"id":"e2","name":"lunch","amount":8000}]

Dans un premier terminal ouvert dans ce dossier et avec l’environnement du projet sélectionné, démarrez le serveur :

python -m http.server 8000 --bind 127.0.0.1

Laissez ce terminal ouvert. L’adresse 127.0.0.1 limite l’écoute à votre machine. Le serveur peut servir les fichiers du dossier choisi : gardez-y uniquement les données de l’exercice. Si le port 8000 est déjà occupé, choisissez un autre port disponible et utilisez ce même numéro dans la commande cliente.

Collecter, valider et enregistrer la réponse complète

Dans le même dossier, enregistrez collect.py avec ce contenu :

import json
import sqlite3
import sys
from urllib.request import urlopen

with urlopen(sys.argv[1], timeout=5) as response:
    if response.status != 200:
        raise ValueError("unexpected status")
    raw = response.read(1_000_001)
    if len(raw) > 1_000_000:
        raise ValueError("response too large")
    data = json.loads(raw.decode("utf-8"))
if not isinstance(data, list) or len(data) > 10000:
    raise ValueError("invalid batch")
MAX_AMOUNT = 1_000_000_000_000
records = []
for row in data:
    if not isinstance(row, dict):
        raise ValueError("invalid row")
    key, name, amount = row.get("id"), row.get("name"), row.get("amount")
    if not isinstance(key, str) or not key.strip() or not isinstance(name, str) or not name.strip():
        raise ValueError("invalid identity")
    if type(amount) is not int or not 0 <= amount <= MAX_AMOUNT:
        raise ValueError("invalid amount")
    records.append((key, name, amount))
con = sqlite3.connect("collected.db", autocommit=False)
try:
    with con:
        con.execute("CREATE TABLE IF NOT EXISTS expenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, amount INTEGER NOT NULL CHECK(amount BETWEEN 0 AND 1000000000000))")
        con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
    count = con.execute("SELECT COUNT(*) FROM expenses").fetchone()[0]
    total = sum(row[0] for row in con.execute("SELECT amount FROM expenses"))
    print("stored:", count, "total:", total)
finally:
    con.close()

Dans un deuxième terminal, ouvert dans ce même dossier et avec le même environnement Python, lancez :

python collect.py http://127.0.0.1:8000/expenses.json

La première exécution doit afficher stored: 2 total: 11500 et terminer avec le code 0. Elle crée collected.db dans le dossier courant du client. Relancez la même commande : le résultat doit encore être deux lignes et 11 500, sans création de deux copies supplémentaires. Le compteur et le total portent sur toute la table conservée, pas seulement sur les éléments de la dernière réponse.

Les requêtes utilisent des paramètres ? pour les valeurs. Le bloc with con valide les écritures si elles réussissent et les annule si une exception survient ; le finally ferme ensuite explicitement la connexion. Le cours sur les transactions SQLite et la référence de autocommit détaillent cette frontière locale.

Le total est calculé avec sum en Python sur les entiers lus dans la table. Le SUM entier de SQLite peut déborder lorsque la somme dépasse sa capacité, même si chaque valeur individuelle y tient ; total() utilise un résultat flottant et ne résout pas ce besoin d’exactitude monétaire. La référence des agrégats SQLite explique ces différences. Les limites de l’exercice restent importantes pour la validation de chaque montant.

Vérifier les mises à jour, les échecs et la fin de collecte

Exécutez deux fois la réponse initiale et comparez le nombre de lignes, le total et le code de sortie.
Dans expenses.json, changez seulement le montant de e2 de 8000 à 9000. Relancez le client : il doit afficher stored: 2 total: 12500.
Remplacez ensuite ce montant par la chaîne "9000". La validation doit échouer avec ValueError: invalid amount. Le client termine en échec ; le contenu validé précédemment dans la base doit rester à deux lignes et 12 500.
Rétablissez le nombre 8000 sans guillemets, puis relancez : le total doit revenir à 11 500. Arrêtez le serveur dans son terminal avec Ctrl+C lorsque les essais sont terminés.

Contrôlez le code de sortie immédiatement après le client avec $LASTEXITCODE sous PowerShell ou echo $? sous un shell macOS/Ubuntu. Une erreur HTTP, une connexion refusée, un JSON mal formé ou une exception SQLite n’est pas un succès vide. Lisez le traceback et n’exécutez pas les traitements qui supposent une collecte réussie. Le cours sur le client HTTP et la référence de urlopen décrivent les réponses et erreurs réseau ; le délai de cinq secondes utilisé ici concerne les opérations bloquantes concernées et ne constitue pas un délai global garanti pour tout le travail.

Une liste vide valide ne supprime pas les anciennes dépenses. De même, retirer un ID de la réponse ne l’efface pas de la base : ce projet fait des insertions et des mises à jour, sans synchronisation des suppressions. CREATE TABLE IF NOT EXISTS ne modifie pas les contraintes d’une table déjà présente ; une vraie évolution de schéma nécessite une migration, et les exercices doivent partir d’un dossier neuf.

Pour une API externe, définissez aussi l’authentification, la pagination, les limites de requêtes et le sens des changements ou suppressions. Une page réussie ne prouve pas que toutes les pages ont été collectées. Évitez les reprises sans limite et la transformation de chaque erreur en liste vide : elles masqueraient des données manquantes. Décidez explicitement quand une collecte est complète avant de la transmettre à un traitement suivant.

Python