#!/usr/bin/env python3 """API de démonstration AppliFrais (fiches de frais GSB) pour le cours Python de reussirmonbtssio.com. Il suffit de Python (3.13 ou plus récent, comme dans le cours) : aucun paquet à installer. Les données sont gardées en mémoire et repartent de zéro à chaque lancement. Lancer l'API dans un terminal, puis tes scripts dans un autre : python api_gsb_demo.py # écoute sur http://localhost:3000/v1 python api_gsb_demo.py 8000 # sur un autre port Arrêt : Ctrl+C. Jeton à envoyer dans chaque requête (sauf /v1/sante) : Authorization: Bearer demo-jeton-gsb Pour simplifier, ce jeton lit toutes les fiches, comme un comptable, et crée les siennes au nom du visiteur 7, Lucas Martin. Routes : GET /v1/sante état de l'API ; répond 503 à deux appels sur trois, pour l'exercice des nouvelles tentatives GET /v1/fiches?etat=CL&page=1&limite=20 liste paginée (états CR, CL, VA, RB ; 100 par page au plus) GET /v1/fiches/{id} une fiche, ou 404 POST /v1/fiches {"mois": "AAAA-MM"} crée la fiche du visiteur 7 pour ce mois : 201, 422 si le mois est invalide, 409 si elle existe déjà Les erreurs suivent la RFC 9457 (Problem Details), comme dans le cours API REST. """ import json import math import re import socket import sys import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse JETON = "demo-jeton-gsb" VISITEUR_CONNECTE = 7 ETATS = {"CR", "CL", "VA", "RB"} PROBLEMES = "https://api.gsb.example/problemes/" # 17 visiteurs, une fiche par visiteur et par mois, d'août à novembre 2026 (le 8 novembre). # Août, septembre et octobre sont clôturés (CL) ; le comptable a déjà validé (VA) # quatre fiches d'août ; novembre est en cours de saisie (CR). VALIDEES = {2: 615.40, 5: 893.05, 7: 742.18, 11: 528.90} FICHES = [] for mois in ["2026-08", "2026-09", "2026-10", "2026-11"]: for visiteur in range(1, 18): identifiant = len(FICHES) + 1 etat = "CR" if mois == "2026-11" else ("VA" if identifiant in VALIDEES else "CL") FICHES.append({"id": identifiant, "visiteurId": visiteur, "mois": mois, "etat": etat, "montantValide": VALIDEES.get(identifiant)}) def probleme(statut, titre, nom=None, **extensions): """Corps d'erreur au format de la RFC 9457.""" corps = {"type": PROBLEMES + nom if nom else "about:blank", "title": titre, "status": statut} corps.update(extensions) return corps class Handler(BaseHTTPRequestHandler): appels_sante = 0 def log_message(self, format, *args): pass # pas de ligne de journal par requête def envoyer(self, statut, corps, entetes=None): donnees = json.dumps(corps, ensure_ascii=False).encode("utf-8") self.send_response(statut) type_corps = "application/problem+json" if statut >= 400 else "application/json" self.send_header("Content-Type", f"{type_corps}; charset=utf-8") for nom, valeur in (entetes or {}).items(): self.send_header(nom, valeur) self.send_header("Content-Length", str(len(donnees))) self.end_headers() self.wfile.write(donnees) def autorise(self): if self.headers.get("Authorization") == f"Bearer {JETON}": return True self.envoyer(401, probleme(401, "Authentification requise", detail="Jeton absent ou invalide"), {"WWW-Authenticate": "Bearer"}) return False def do_GET(self): url = urlparse(self.path) if url.path == "/v1/sante": Handler.appels_sante += 1 if Handler.appels_sante % 3 != 0: return self.envoyer(503, probleme(503, "Service indisponible"), {"Retry-After": "1"}) return self.envoyer(200, {"statut": "ok", "version": "1.4.0"}) if url.path != "/v1/fiches" and not url.path.startswith("/v1/fiches/"): return self.envoyer(404, probleme(404, "Ressource introuvable")) if not self.autorise(): return une_fiche = re.fullmatch(r"/v1/fiches/(\d+)", url.path) if une_fiche: fiche = next((f for f in FICHES if f["id"] == int(une_fiche[1])), None) if fiche is None: return self.envoyer(404, probleme(404, "Fiche introuvable")) return self.envoyer(200, fiche) if url.path != "/v1/fiches": return self.envoyer(404, probleme(404, "Ressource introuvable")) parametres = parse_qs(url.query) etat = parametres.get("etat", [None])[0] try: page = int(parametres.get("page", ["1"])[0]) limite = min(int(parametres.get("limite", ["20"])[0]), 100) except ValueError: page = limite = 0 if page < 1 or limite < 1 or (etat is not None and etat not in ETATS): return self.envoyer(400, probleme(400, "Paramètre invalide", detail="etat : CR, CL, VA ou RB ; page et limite : entiers positifs")) liste = [f for f in FICHES if etat is None or f["etat"] == etat] debut = (page - 1) * limite self.envoyer(200, {"donnees": liste[debut:debut + limite], "pagination": {"page": page, "limite": limite, "total": len(liste), "pages": math.ceil(len(liste) / limite)}}) def do_POST(self): if urlparse(self.path).path != "/v1/fiches": return self.envoyer(404, probleme(404, "Ressource introuvable")) if not self.autorise(): return longueur = int(self.headers.get("Content-Length", 0)) try: corps = json.loads(self.rfile.read(longueur) or b"{}") except ValueError: return self.envoyer(400, probleme(400, "JSON illisible")) mois = corps.get("mois", "") if isinstance(corps, dict) else "" if not isinstance(mois, str) or not re.fullmatch(r"\d{4}-(0[1-9]|1[0-2])", mois): return self.envoyer(422, probleme(422, "Données invalides", "validation", detail="Un ou plusieurs champs sont invalides.", erreurs=[{"champ": "mois", "message": "Format attendu : AAAA-MM"}])) if any(f["visiteurId"] == VISITEUR_CONNECTE and f["mois"] == mois for f in FICHES): return self.envoyer(409, probleme(409, "Fiche déjà existante", "fiche-existante", detail=f"Une seule fiche par visiteur et par mois : {mois} existe déjà.")) nouvelle = {"id": len(FICHES) + 1, "visiteurId": VISITEUR_CONNECTE, "mois": mois, "etat": "CR", "montantValide": None} FICHES.append(nouvelle) self.envoyer(201, nouvelle, {"Location": f"/v1/fiches/{nouvelle['id']}"}) class ServeurIPv6(ThreadingHTTPServer): address_family = socket.AF_INET6 def demarrer(port): """Écoute sur 127.0.0.1 et, si possible, sur ::1 : « localhost » peut désigner l'une ou l'autre.""" principal = ThreadingHTTPServer(("127.0.0.1", port), Handler) try: ipv6 = ServeurIPv6(("::1", port), Handler) except OSError: pass # pas d'IPv6 sur cette machine : IPv4 suffit else: threading.Thread(target=ipv6.serve_forever, daemon=True).start() print(f"API de démonstration GSB sur http://localhost:{port}/v1 (Ctrl+C pour arrêter)", flush=True) try: principal.serve_forever() except KeyboardInterrupt: print("\nAPI arrêtée.") if __name__ == "__main__": demarrer(int(sys.argv[1]) if len(sys.argv) > 1 else 3000)