Créer une API de données sur les demandes de service municipal

Ce que vous saurez faire

Créez une vraie API sur une vraie base de données, de la première requête aux étapes de déploiement. Les analystes du service des transports d’une ville veulent une source unique pour leurs tableaux de bord. Ils veulent lire les demandes de réparation de rues page par page, avec des chiffres quotidiens. L’API ne doit jamais modifier les données ni exécuter du SQL écrit par un client. Elle ne doit jamais sauter ni répéter une ligne pendant qu’un tableau de bord lit les pages.

Étude de casPièce de portfolio

Jeu de données

NYC 311 Service Requests, service des transports (DOT), janvier à juin 2025

Licence

NYC Open Data : aucune restriction d'utilisation

Durée

Environ 5 heures

S’exécute dans

VS Code, Python 3.12+

Compétences utilisées

• SQLite depuis Python
• Accès en lecture seule
• SQL paramétré
• Filtres validés
• Pagination par clé
• Statistiques quotidiennes
• Tests du contrat
• Postman

Télécharger le projet de départ (.zip)Projet terminé (.zip)

Décompressez le projet de départ, ouvrez le dossier dans VS Code et suivez les étapes ci-dessous. Elles commencent par un nouvel environnement virtuel.

Le projet de départ est un petit projet FastAPI qui contient les éléments que vous avez construits dans le cours. Il comprend les paramètres, le format d’erreur, la vérification de la clé d’API, les identifiants de requête et la limite du corps. Vous ajoutez l’accès à la base de données, les points de terminaison, les tests et la documentation. Le projet terminé montre le résultat, pour vous aider en cas de blocage ou pour comparer. Le code du projet est sous licence MIT. Vous pouvez donc publier votre version.

Avant de commencerCe qu’il faut savoir et avoir sous la mainEnviron 300 minutes

Ce que vous devez savoir

Cette étude de cas utilise tout le cours. Voici les leçons sur lesquelles elle s’appuie le plus :

• Un seul format d’erreur et le rejet des champs inconnus, dans Validation et réponses d’erreur.
• Les paramètres et les clients de test avec leurs propres paramètres, dans Configuration et journalisation : des tests avec leurs propres paramètres.
• Le contrat révisé et son test, dans Garder la documentation exacte : exemple détaillé : le contrat révisé.
• L’exécution d’une collection avec la CLI de Postman, dans Tests et exécutions reproductibles : exécuter depuis le terminal : Postman CLI.
• Les instructions SQL SELECT, WHERE, GROUP BY et ORDER BY. La page explique chaque requête. Vous pouvez donc suivre sans avoir fait de cours de SQL.

Cette page enseigne deux choses que le cours n’a pas abordées : le module Python sqlite3 et la pagination.

Ce qu’il vous faut sur votre ordinateur

• Python 3.12 ou plus récent, VS Code et curl, comme dans le cours.
• Environ 60 Mo d’espace disque libre pour les paquets et la base de données (8 Mo à télécharger, 36 Mo décompressée).
• La CLI de Postman pour la partie 8, présentée dans Tests et exécutions reproductibles. L’application Postman est facultative.

Testé sur macOS. Versions : Python 3.14.6, SQLite 3.50.4, FastAPI 0.142.2, pytest 9.1.1, Postman CLI 1.70.0. Les commandes Windows suivent celles du cours, mais elles n’ont pas été testées. Le déploiement de la partie 9 n’a pas été exécuté pour cette étude de cas.

La situation

Teresa Okafor dirige une petite équipe d’analyse qui travaille pour le service des transports d’une ville. Les gens appellent ou écrivent au service 311 de la ville. Ils signalent des nids-de-poule, des feux de circulation défectueux, des lampadaires éteints et des trottoirs fissurés. Chaque appel ou message devient une demande de service, que les équipes du service des transports traitent. Chaque lundi, les gestionnaires examinent les semaines précédentes : d’où viennent les demandes et à quelle vitesse elles sont fermées.

Aujourd’hui, les chiffres viennent d’un fichier exporté d’un tableur. Quelqu’un télécharge les demandes, les nettoie à la main et envoie le fichier par courriel. Chaque tableau de bord nettoie à nouveau le fichier, un peu différemment. Deux tableaux de bord affichent donc deux chiffres différents pour la même semaine. Personne ne sait lequel est bon. Le fichier est aussi volumineux et déjà périmé dès son envoi.

Teresa veut une seule source pour tous les tableaux de bord. Une API Web doit fournir les demandes nettoyées, avec les mêmes règles pour tout le monde. Tomás Reyes, un développeur de l’équipe, la construit. Priya Nair crée les tableaux de bord qui l’utiliseront. Elle pose les questions qu’un client poserait.

Les gestionnaires doivent décider où porter leur attention : quels types de demandes et quels arrondissements prennent trop longtemps à fermer les demandes. L’API appuie cette décision. Elle ne la prend pas.

Les personnes et l’équipe sont fictives. Les demandes sont réelles : elles viennent des données ouvertes de la ville de New York.

Ce que vous allez construire

Vous allez construire city-requests-api, une API en lecture seule. Elle répond aux questions sur les données, mais aucune requête ne peut les modifier. Les tableaux de bord ont quatre besoins. L’API a donc quatre points de terminaison principaux :

1. Une liste de demandes, avec des filtres : arrondissement, type de plainte, état et date de création. La liste est divisée en pages. Un tableau de bord parcourt les pages jusqu’à la fin.
2. Une demande obtenue par son identifiant, pour afficher ses détails.
3. Les types de plaintes, avec le nombre de demandes de chaque type, pour les menus du tableau de bord.
4. Des chiffres quotidiens : combien de demandes ont été ouvertes chaque jour, combien ont été fermées en 14 jours et le délai médian de fermeture.

Point de terminaison	Ce qu’il renvoie
GET /v1/service-requests	Une page de demandes, les plus anciennes d’abord, et un curseur pour la page suivante
GET /v1/service-requests/{id}	Une demande, ou 404 not_found (introuvable)
GET /v1/complaint-types	Les types de plaintes, avec leur nombre de demandes
GET /v1/stats/daily	Une ligne par jour : demandes ouvertes, fermées en 14 jours, proportion et nombre médian de jours avant fermeture
GET /v1/dataset	La source, la date de l’instantané et les règles de nettoyage
GET /health, GET /ready	Le fonctionnement du processus et sa capacité à servir les requêtes (la base de données est lisible)

Chaque point de terminaison sous /v1 exige l’en-tête X-API-Key. Chaque erreur utilise le format du cours. À la fin, vous avez un projet testé avec un contrat révisé, une collection Postman, un README et les étapes de déploiement.

Les données

Les données viennent du jeu de données 311 Service Requests from 2020 to Present de New York, sur NYC Open Data. La ville publie chaque demande et les met à jour quotidiennement. NYC Open Data indique : « Il n’y a aucune restriction sur l’utilisation des données ouvertes. » La ville demande aux utilisateurs d’indiquer la source, la version et les modifications. Cette section et le fichier DATASET.md du projet le font.

Nextia Learning a extrait les demandes du service des transports (organisme DOT) le 7 octobre 2026. Il s’agit des 315 424 demandes créées de janvier 2024 à juin 2025. Cette copie figée est partagée avec le cours de SQL. Cette étude de cas utilise le premier semestre de 2025 et applique neuf règles de nettoyage. Voici les plus importantes :

• Conserver janvier à juin 2025. Cela représente 110 618 demandes.
• Supprimer les envois répétés. Deux lignes identiques dans toutes les colonnes sauf l’identifiant représentent une demande envoyée deux fois. On supprime 146 lignes.
• Exclure les dates impossibles. Pour 557 demandes, la fermeture précède la création. Nous ne pouvons pas les corriger avec confiance. Nous les excluons donc.
• Ne conserver aucune adresse ni aucun texte libre. Le fichier extrait ne contient aucune adresse municipale. Les règles suppriment aussi les réponses en texte libre de la ville.
• Une seule façon de dire « non consigné ». « Unspecified » (non précisé), « UNKNOWN » (inconnu) et les valeurs vides deviennent NULL.
• Des codes courts et fixes, comme staten_island pour l’arrondissement et in_progress (en cours) pour l’état. Ils facilitent l’écriture d’une chaîne de requête. Le cours de SQL conserve les valeurs de la ville, comme STATEN ISLAND. Une API les normalise une seule fois pour éviter ce travail à chaque client.

Il reste 109 915 demandes. Elles se trouvent dans un seul fichier de base de données SQLite, service_requests_2025h1.db. Le script scripts/build_db.py du projet applique les règles. Le fichier DATASET.md présente les neuf règles. L’étude de cas Préparer une table sans fuite pour les demandes de réparation d’une ville du cours de SQL construit le pipeline de nettoyage complet sur le même fichier. Elle explique chaque règle. Vous n’en avez pas besoin ici : vous téléchargez la base de données prête à utiliser.

Les colonnes

Colonne	Signification
id	L’identifiant de la demande attribué par la ville
created_at, closed_at	La date de création et de fermeture, à l’heure locale de New York, par exemple 2025-03-03T07:00:00
status	open (ouverte), assigned (attribuée), in_progress, pending (en attente) ou closed (fermée), au 7 octobre 2026
complaint_type, descriptor	Le sujet de la demande, par exemple Street Condition (état de la rue) et Pothole (nid-de-poule)
borough	bronx, brooklyn, manhattan, queens ou staten_island; NULL pour 598 demandes
location_type, community_board, council_district, incident_zip	Le lieu, selon les termes de la ville
channel	online (en ligne), phone (téléphone), mobile ou other (autre); NULL pour 65 906 demandes
latitude, longitude	Le lieu, souvent une intersection; NULL pour 16 647 demandes
days_to_close	closed_at moins created_at, en jours

Deux faits comptent pour une API sur ces données. D’abord, les dates et heures n’ont aucun fuseau horaire. Elles sont à l’heure locale de New York, comme la ville les publie. Ensuite, l’état est un instantané. Une demande en attente le 7 octobre 2026 peut être fermée aujourd’hui.

Préparer le projet

Les étapes utilisent VS Code et son terminal. Le terminal s’ouvre dans le dossier du projet, où toutes les commandes s’exécutent.

1. Téléchargez le projet de départ depuis l’encadré en haut de cette page. Décompressez-le dans votre dossier de projets, par exemple ~/projects. Vous obtenez le dossier city-requests-api.

2. Dans VS Code, choisissez File > Open Folder (fichier, ouvrir un dossier), puis ouvrez city-requests-api. Ouvrez ensuite un terminal : Terminal > New Terminal (terminal, nouveau terminal).

3. Créez et activez un environnement virtuel :

   macOS

   python3 -m venv .venv
   source .venv/bin/activate

   Windows

   python -m venv .venv
   .venv\Scripts\Activate.ps1

   Linux

   python3 -m venv .venv
   source .venv/bin/activate

   L’invite commence par (.venv). Si VS Code propose d’utiliser le nouvel environnement pour ce dossier, choisissez Yes (oui).

4. Installez les versions exactes du cours :

   python -m pip install -r requirements-lock.txt

5. Téléchargez la base de données, vérifiez-la et décompressez-la :

   python scripts/get_db.py

   Sortie

   Downloading https://github.com/Nextia-Intelligence/nextia-learning-labs/raw/main/C16/case-studies/data/service_requests_2025h1.db.gz
   Checked service_requests_2025h1.db.gz: SHA-256 26492959add9... is correct
   Wrote data/service_requests_2025h1.db: 109,915 service requests

6. Exécutez les tests :

   python -m pytest

   Sortie (fin)

   tests/test_health.py ..                                                  [100%]
   
   ============================== 2 passed in 0.01s ===============================

Ce que cela nous apprend. Les paquets fonctionnent et la base de données est le fichier que nous avons testé. Le script a comparé sa somme de contrôle SHA-256 avec celle que nous avons publiée. Cette somme est une empreinte des octets exacts du fichier. Un fichier tronqué par une mauvaise connexion ou modifié a une somme de contrôle différente. Le script s’arrête alors. Les deux tests montrent que l’application de départ fonctionne. /health répond, et un chemin inconnu reçoit un 404 dans le format d’erreur du cours. Les tests n’ont pas besoin du téléchargement. Ils créent leur propre petite base de données, comme vous le verrez dans la partie 7.

Démarrez maintenant l’API et appelez-la depuis un deuxième terminal (dans VS Code, le + du panneau du terminal) :

macOS

API_KEY=local-dev-key fastapi dev

Windows

$env:API_KEY = "local-dev-key"
fastapi dev

Linux

API_KEY=local-dev-key fastapi dev

Second terminal (on Windows: curl.exe)

curl -i http://127.0.0.1:8000/health

Sortie

HTTP/1.1 200 OK
date: Thu, 08 Oct 2026 00:25:12 GMT
server: uvicorn
content-length: 15
content-type: application/json
x-request-id: 05fc5a9e6b43

{"status":"ok"}

Ce que cela nous apprend. C’est la première vérification réussie : l’API fonctionne. Chaque réponse contient un identifiant de requête fourni par l’intergiciel du cours. Laissez le serveur fonctionner pendant votre travail. fastapi dev le redémarre chaque fois que vous enregistrez un fichier Python. La clé local-dev-key est réservée à votre ordinateur. Chaque commande curl ci-dessous destinée à /v1 l’envoie avec -H "X-API-Key: local-dev-key".

Partie 1 : examiner la base de données depuis Python

L’API lit une base de données SQLite. SQLite conserve toute une base de données dans un seul fichier et n’a pas besoin de serveur de base de données. Python inclut le module sqlite3 dans sa bibliothèque standard. Il n’y a donc rien à installer. Trois idées suffisent pour ce projet :

• sqlite3.connect(path) ouvre le fichier et renvoie une connexion.
• con.execute(sql) exécute une instruction SQL et renvoie un curseur sur les lignes du résultat.
• .fetchone() récupère une ligne du curseur, et .fetchall() les récupère toutes, sous forme de tuples.

Démarrez Python dans le dossier du projet avec python. L’invite est >>>. Tapez les lignes après chaque >>> :

>>> import sqlite3
>>> con = sqlite3.connect("data/service_requests_2025h1.db")
>>> con.execute("SELECT COUNT(*) FROM service_requests").fetchone()
(109915,)
>>> rows = con.execute("SELECT id, created_at, complaint_type, borough, status FROM service_requests ORDER BY created_at, id LIMIT 3").fetchall()
>>> for row in rows:
...     print(row)
...
(63580493, '2025-01-01T00:02:00', 'Traffic Signal Condition', 'queens', 'closed')
(63582137, '2025-01-01T00:05:00', 'Traffic Signal Condition', 'queens', 'closed')
(63578936, '2025-01-01T00:10:00', 'Traffic Signal Condition', 'queens', 'closed')
>>> con.execute("SELECT created_at, COUNT(*) FROM service_requests GROUP BY created_at ORDER BY COUNT(*) DESC LIMIT 3").fetchall()
[('2025-03-03T07:00:00', 34), ('2025-01-15T07:00:00', 32), ('2025-02-24T07:00:00', 30)]
>>> con.execute("SELECT value FROM meta WHERE name = 'pulled_on'").fetchone()
('2026-10-07',)
>>> con.close()

Tapez exit() pour quitter Python.

Ce que cela nous apprend. Chaque ligne est un tuple, dans l’ordre des colonnes du SELECT. Les identifiants ne suivent pas l’ordre chronologique : la demande 63578936 a été créée après la demande 63582137. L’API ne peut donc pas trier par identifiant quand les tableaux de bord veulent l’ordre chronologique. La deuxième requête révèle un piège : 34 demandes ont exactement la même date et heure, le 3 mars 2025 à 7 h. Elles concernent des nids-de-poule de Staten Island saisis en un seul lot. Dans toute la table, 11 489 demandes partagent leur date et heure de création avec au moins une autre demande. Retenez ce fait pour la pagination dans la partie 5. La table meta stocke des faits sur les données, comme la date de l’instantané. L’API peut ainsi indiquer leur âge à ses clients.

Partie 2 : ouvrir la base de données en lecture seule

Le problème

Un simple appel à sqlite3.connect(path) permet de lire et d’écrire. L’API a seulement besoin de lire. Supposons qu’un bogue, ou une requête qui trompe le code, exécute DELETE ou UPDATE. Une connexion qui permet l’écriture l’exécute, et tous les tableaux de bord montrent les dégâts. Un bon code rend cela peu probable. Une connexion en lecture seule le rend impossible. C’est une garantie beaucoup plus forte.

Il y a un deuxième problème, moins important. SQLite suppose que d’autres programmes peuvent modifier le fichier à tout moment. Avant chaque lecture, il verrouille donc le fichier et vérifie s’il a changé. Notre fichier ne change jamais pendant que l’API fonctionne. Ce travail est donc inutile.

L’idée : dire à SQLite ce que nous garantissons

SQLite accepte la base de données sous forme d’URI, une adresse de fichier avec des options, comme file:/path/to/file.db?mode=ro&immutable=1 :

• mode=ro (lecture seule) : SQLite refuse toute écriture sur cette connexion.
• immutable=1 : nous garantissons que le fichier ne change pas pendant qu’il est ouvert. SQLite ne le verrouille ni ne le vérifie donc.

Voici le coût de immutable=1 : si quelqu’un remplace le fichier pendant que l’API fonctionne, elle peut lire des données incorrectes. Pour publier de nouvelles données, vous remplacez donc le fichier et redémarrez l’API. La partie 9 fait exactement cela.

Construire la solution

Créez le fichier city_api/db.py :

city_api/db.py

import sqlite3
from pathlib import Path

SCHEMA_VERSION = 1  # scripts/build_db.py writes it with PRAGMA user_version


class DatabaseUnavailable(Exception):
    """The database cannot be read now. The request may work later."""


def connect(path: Path) -> sqlite3.Connection:
    """Open the database read-only.

    mode=ro: SQLite refuses every write, even a wrong one in our code.
    immutable=1: we promise that the file never changes while the API runs,
    so SQLite does not lock it or check it for changes made by others.
    """
    if not path.is_file():
        raise DatabaseUnavailable(f"no database file at {path}")
    uri = f"{path.resolve().as_uri()}?mode=ro&immutable=1"
    try:
        con = sqlite3.connect(uri, uri=True, check_same_thread=False)
        version = con.execute("PRAGMA user_version").fetchone()[0]
    except sqlite3.Error as exc:
        raise DatabaseUnavailable(f"cannot read {path}: {exc}") from None
    if version != SCHEMA_VERSION:
        con.close()
        raise DatabaseUnavailable(
            f"{path} has schema version {version}, expected {SCHEMA_VERSION}"
        )
    con.row_factory = sqlite3.Row
    return con

Ligne	Pourquoi
path.resolve().as_uri()	Produit une adresse file: correcte sur chaque système, même avec des espaces dans le chemin ou C:\ sur Windows.
uri=True	Indique à sqlite3 de lire les options après ?.
check_same_thread=False	FastAPI peut ouvrir la connexion dans un fil d’exécution et l’utiliser dans un autre. Chaque requête a sa propre connexion, donc c’est sûr.
PRAGMA user_version	Un nombre que le script de construction stocke dans le fichier. Si quelqu’un déploie un fichier avec une autre structure de tables, l’API indique « pas prête » plutôt que d’échouer à chaque requête.
con.row_factory = sqlite3.Row	Permet de lire les lignes par nom de colonne, row["created_at"], et de les convertir en dictionnaire avec dict(row).

Vérifiez la garantie dans Python, dans le dossier du projet :

>>> from pathlib import Path
>>> from city_api import db
>>> con = db.connect(Path("data/service_requests_2025h1.db"))
>>> con.execute("SELECT COUNT(*) FROM service_requests").fetchone()[0]
109915
>>> con.execute("DELETE FROM service_requests")
Traceback (most recent call last):
  ...
sqlite3.OperationalError: attempt to write a readonly database
>>> db.connect(Path("data/missing.db"))
Traceback (most recent call last):
  ...
city_api.db.DatabaseUnavailable: no database file at data/missing.db

Ce que cela nous apprend. La lecture fonctionne et le DELETE échoue avant de modifier une seule ligne : SQLite lui-même le refuse. Un fichier absent déclenche notre propre exception, DatabaseUnavailable, que l’API peut convertir en réponse claire.

Capacité à servir : l’API peut-elle lire ses données?

Comme dans Publier en toute sécurité, /health indique « le processus fonctionne », et /ready indique « il peut faire son travail ». Ici, ce travail exige la base de données. Si le fichier manque, /ready doit répondre 503. Tous les points de terminaison de données doivent faire de même, sans trace d’appels.

Dans city_api/errors.py, importez l’exception en haut :

city_api/errors.py (add to the imports)

from city_api.db import DatabaseUnavailable

Ajoutez ensuite son gestionnaire dans add_error_handlers, avant le gestionnaire de Exception :

city_api/errors.py (add in add_error_handlers)

    @app.exception_handler(DatabaseUnavailable)
    async def database_unavailable(request: Request, exc: DatabaseUnavailable):
        logger.warning("database unavailable: %s id=%s", exc, request.state.request_id)
        return error_response(
            request,
            503,
            "database_unavailable",
            "The data is not available. Try again later.",
        )

La ligne du journal contient les détails, comme le chemin. Le client reçoit seulement un code stable et un court message. Le chemin d’un fichier sur le serveur ne concerne pas le client.

Ajoutez le modèle à city_api/models.py, après Health :

city_api/models.py (add)

class Readiness(BaseModel):
    status: Literal["ready"]

Dans city_api/main.py, ajoutez l’importation de db et modifiez l’importation des modèles :

city_api/main.py (imports)

from city_api import db
from city_api.config import Settings, load_settings
from city_api.errors import add_error_handlers, error_response
from city_api.models import ErrorResponse, Health, Readiness

Dans create_app, après l’avertissement sur API_KEY, ajoutez un avertissement pour une base de données absente. C’est un avertissement, pas une erreur : l’API démarre quand même. Ainsi, /health et /ready peuvent signaler le problème.

city_api/main.py (in create_app, after the API_KEY warning)

    if not settings.database_path.is_file():
        logger.warning("no database at %s: /ready answers 503", settings.database_path)

Ajoutez ensuite /ready, après /health :

city_api/main.py (in create_app, after /health)

    @app.get(
        "/ready",
        tags=["health"],
        summary="Check that the service can read the data",
        responses={503: {"model": ErrorResponse, "description": "Not ready."}},
    )
    def ready() -> Readiness:
        con = db.connect(settings.database_path)
        try:
            con.execute("SELECT id FROM service_requests LIMIT 1").fetchone()
        finally:
            con.close()
        return Readiness(status="ready")

Le chemin de la base de données vient du paramètre DATABASE_PATH dans city_api/config.py. Sa valeur par défaut est data/service_requests_2025h1.db. Vérifiez les deux cas. D’abord, avec la base de données :

Second terminal (on Windows: curl.exe)

curl -s http://127.0.0.1:8000/ready

Sortie

{"status":"ready"}

Arrêtez ensuite le serveur et redémarrez-le avec un chemin qui n’existe pas :

macOS

API_KEY=local-dev-key DATABASE_PATH=data/nothing.db fastapi dev

Windows

$env:DATABASE_PATH = "data/nothing.db"
fastapi dev

Linux

API_KEY=local-dev-key DATABASE_PATH=data/nothing.db fastapi dev

Appelez à nouveau /ready depuis le deuxième terminal :

Output of curl -s http://127.0.0.1:8000/ready

{"error":{"code":"database_unavailable","message":"The data is not available. Try again later.","request_id":"623f2048eda4","fields":[]}}

First terminal (the server's log)

WARNING city_api: no database at data/nothing.db: /ready answers 503
WARNING city_api: database unavailable: no database file at data/nothing.db id=623f2048eda4

Ce que cela nous apprend. Le service sait s’il peut faire son travail. Il l’indique dans le format standard, avec l’état 503. Une plateforme d’hébergement ou un outil de surveillance peut maintenant distinguer « le processus est arrêté » de « les données manquent ». Le journal contient les détails et le même identifiant de requête. Une personne du soutien peut donc trouver la cause. Arrêtez le serveur et redémarrez-le sans DATABASE_PATH. Sur Windows, supprimez d’abord le paramètre : Remove-Item Env:DATABASE_PATH.

Partie 3 : SQL paramétré

Le problème : du SQL construit à partir de texte

Les tableaux de bord filtrent par type de plainte. La façon évidente de filtrer consiste à insérer le texte du client dans le SQL avec une chaîne f. Essayez dans Python, avec une connexion en lecture seule :

>>> import sqlite3
>>> con = sqlite3.connect("file:data/service_requests_2025h1.db?mode=ro", uri=True)
>>> wanted = "Bench"
>>> sql = f"SELECT COUNT(*) FROM service_requests WHERE complaint_type = '{wanted}'"
>>> con.execute(sql).fetchone()
(48,)

Cela fonctionne. Supposons maintenant qu’un client envoie ce « type de plainte » :

>>> wanted = "x' OR '1'='1"
>>> sql = f"SELECT COUNT(*) FROM service_requests WHERE complaint_type = '{wanted}'"
>>> print(sql)
SELECT COUNT(*) FROM service_requests WHERE complaint_type = 'x' OR '1'='1'
>>> con.execute(sql).fetchone()
(109915,)

Ce que cela nous apprend. Le texte du client a changé la question. Le guillemet dans le texte a fermé notre chaîne, et OR '1'='1' a rendu la condition vraie pour chaque ligne. C’est une injection SQL : des données externes deviennent une partie du code SQL. Ici, elle a seulement compté toutes les lignes. Avec d’autres instructions SQL, elle peut lire des tables que le client ne doit pas voir. Avec une connexion permettant l’écriture, elle peut modifier et supprimer des données. C’est l’une des failles de sécurité les plus courantes dans les logiciels réels.

L’injection n’est pas seulement une attaque. Une valeur légitime fait échouer le même code :

>>> wanted = "Driver's complaint"
>>> con.execute(f"SELECT COUNT(*) FROM service_requests WHERE complaint_type = '{wanted}'").fetchone()
Traceback (most recent call last):
  ...
sqlite3.OperationalError: near "s": syntax error

L’idée : les valeurs sont transmises séparément

Donnez à SQLite le SQL avec un ? à la place de chaque valeur, et les valeurs dans un tuple séparé. SQLite lit d’abord le SQL. Il utilise ensuite les valeurs, toujours comme des valeurs, jamais comme du code :

>>> con.execute("SELECT COUNT(*) FROM service_requests WHERE complaint_type = ?", ("x' OR '1'='1",)).fetchone()
(0,)
>>> con.execute("SELECT COUNT(*) FROM service_requests WHERE complaint_type = ?", ("Bench",)).fetchone()
(48,)

Ce que cela nous apprend. Avec le ?, l’attaque devient simplement un type de plainte étrange qu’aucune demande ne possède : zéro ligne. Une vraie valeur fonctionne toujours. C’est une requête paramétrée, et le ? est un paramètre. Remarquez la virgule dans ("Bench",) : elle crée un tuple contenant une seule valeur.

Un paramètre peut remplacer seulement une valeur. Il ne peut pas remplacer un nom de colonne, un nom de table ou un mot comme ASC (ordre croissant). Voici donc la règle de ce projet :

• Les valeurs d’une requête arrivent à SQLite uniquement comme paramètres.
• Les noms de colonnes et les mots SQL proviennent uniquement de notre code, jamais d’une requête.

Construction : une demande à partir de son identifiant

Tout le SQL du projet va dans un seul fichier. Ainsi, une personne qui révise le code peut vérifier la règle au même endroit. Créez city_api/queries.py :

city_api/queries.py

"""Every SQL statement of the API.

Values from the request reach SQLite only as parameters (the ? marks), never
as part of the SQL text. Column names and SQL words come only from this file.
"""

import sqlite3
import statistics
from datetime import date, timedelta

COLUMNS = (
    "id, created_at, closed_at, status, complaint_type, descriptor, location_type, "
    "borough, community_board, council_district, incident_zip, channel, "
    "latitude, longitude, days_to_close"
)


def get_request(con: sqlite3.Connection, request_id: int) -> sqlite3.Row | None:
    return con.execute(
        f"SELECT {COLUMNS} FROM service_requests WHERE id = ?", (request_id,)
    ).fetchone()

La chaîne f est sûre ici : COLUMNS est notre propre constante. L’identifiant est toujours transmis comme paramètre. (statistics, date et timedelta serviront à la partie 6.)

La route doit déclencher volontairement une erreur avec un statut et un code : 404 not_found pour un identifiant inconnu. Les parties 4 à 6 en nécessitent d’autres. Ajoutez une classe commune à toutes ces erreurs dans city_api/errors.py, après CODES :

city_api/errors.py (add after CODES)

class ApiError(Exception):
    """An error that a route raises on purpose, with its status and code."""

    def __init__(self, status: int, code: str, message: str, fields=()):
        super().__init__(message)
        self.status, self.code, self.message, self.fields = status, code, message, fields

Son gestionnaire va dans add_error_handlers, après le gestionnaire de HTTPException :

city_api/errors.py (add in add_error_handlers)

    @app.exception_handler(ApiError)
    async def api_error(request: Request, exc: ApiError):
        return error_response(request, exc.status, exc.code, exc.message, exc.fields)

Le modèle de réponse décrit une demande. Ajoutez-le à city_api/models.py, après les importations. Les deux types Literal énumèrent les seules valeurs possibles pour un arrondissement et un statut. EXAMPLE_REQUEST est une vraie demande, présentée dans la documentation et vérifiée par un test :

city_api/models.py (add, after the imports)

Borough = Literal["bronx", "brooklyn", "manhattan", "queens", "staten_island"]
Status = Literal["open", "assigned", "in_progress", "pending", "closed"]

EXAMPLE_REQUEST = {
    "id": 64246800,
    "created_at": "2025-03-03T17:55:20",
    "closed_at": "2025-03-04T17:14:24",
    "status": "closed",
    "complaint_type": "Street Sign - Missing",
    "descriptor": "Speed Limit",
    "location_type": "Street",
    "borough": "manhattan",
    "community_board": "01 MANHATTAN",
    "council_district": 1,
    "incident_zip": "10013",
    "channel": "phone",
    "latitude": 40.721709619645075,
    "longitude": -74.01136041681767,
    "days_to_close": 0.97,
}


class ServiceRequest(BaseModel):
    """One request to the city about a street, a sign, a light or a sidewalk."""

    model_config = ConfigDict(json_schema_extra={"examples": [EXAMPLE_REQUEST]})

    id: int = Field(description="The city's ID of the request (unique_key).")
    created_at: str = Field(description="When it was made, New York local time.")
    closed_at: str | None = Field(description="When it was closed, as published.")
    status: Status | None = Field(
        description="Status on the snapshot date. To find closed requests, use closed_at."
    )
    complaint_type: str
    descriptor: str | None = Field(description="Detail of the complaint, such as Pothole.")
    location_type: str | None
    borough: Borough | None = Field(description="null: the city did not record it.")
    community_board: str | None
    council_district: int | None
    incident_zip: str | None
    channel: Literal["online", "phone", "mobile", "other"] | None = Field(
        description="How the request was made. null: unknown."
    )
    latitude: float | None
    longitude: float | None
    days_to_close: float | None = Field(
        description="Days from created_at to closed_at, when closed_at is set."
    )

Passons aux routes. Créez city_api/routes.py :

city_api/routes.py

import sqlite3
from collections.abc import Iterator
from typing import Annotated

from fastapi import APIRouter, Depends, Path, Request

from city_api import db, queries
from city_api.errors import ApiError
from city_api.models import ErrorResponse, ServiceRequest
from city_api.security import require_api_key

ERRORS = {
    401: {"model": ErrorResponse, "description": "The API key is missing or wrong."},
    422: {"model": ErrorResponse, "description": "A parameter is not valid."},
    503: {"model": ErrorResponse, "description": "The data is not available."},
}

router = APIRouter(prefix="/v1", dependencies=[Depends(require_api_key)], responses=ERRORS)


def get_db(request: Request) -> Iterator[sqlite3.Connection]:
    """Give the route a read-only connection, and close it after the response."""
    con = db.connect(request.app.state.settings.database_path)
    try:
        yield con
    finally:
        con.close()


Database = Annotated[sqlite3.Connection, Depends(get_db)]


@router.get(
    "/service-requests/{request_id}",
    tags=["service requests"],
    summary="Get one service request",
    responses={404: {"model": ErrorResponse, "description": "No request has this ID."}},
)
def get_service_request(
    request_id: Annotated[int, Path(ge=1, description="The request's id.")], con: Database
) -> ServiceRequest:
    row = queries.get_request(con, request_id)
    if row is None:
        raise ApiError(404, "not_found", f"No service request has the ID {request_id}.")
    return ServiceRequest(**dict(row))

Partie	Pourquoi
dependencies=[Depends(require_api_key)] sur le routeur	Chaque route du routeur exige la clé, même les routes que vous ajouterez plus tard. Vous ne pouvez pas l’oublier sur une route.
responses=ERRORS	Documente une seule fois les trois erreurs que chaque route de données peut renvoyer.
get_db avec yield	Une dépendance, comme dans Séparer les responsabilités. Le code avant yield s’exécute avant la route. Le code après s’exécute après la réponse, même après une erreur. Chaque connexion est donc fermée.
Database = Annotated[...]	Un nom court pour « une connexion provenant de get_db », afin que chaque route écrive con: Database.
Path(ge=1)	Un identifiant est un nombre entier à partir de 1. abc ou 0 donne 422, avant toute exécution de SQL.

Ouvrir une connexion pour chaque requête prend bien moins d’une milliseconde avec immutable=1, et garde les requêtes séparées. Enfin, reliez le routeur à l’application dans city_api/main.py. Ajoutez l’importation :

city_api/main.py (add to the imports)

from city_api.routes import router

Puis incluez le routeur dans create_app, juste après add_error_handlers(app) :

city_api/main.py (in create_app)

    app.include_router(router)

Appelez-le depuis le deuxième terminal, avec la clé :

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" http://127.0.0.1:8000/v1/service-requests/64246800
curl -s -H "X-API-Key: local-dev-key" http://127.0.0.1:8000/v1/service-requests/1

Sortie

{"id":64246800,"created_at":"2025-03-03T17:55:20","closed_at":"2025-03-04T17:14:24","status":"closed","complaint_type":"Street Sign - Missing","descriptor":"Speed Limit","location_type":"Street","borough":"manhattan","community_board":"01 MANHATTAN","council_district":1,"incident_zip":"10013","channel":"phone","latitude":40.721709619645075,"longitude":-74.01136041681767,"days_to_close":0.97}
{"error":{"code":"not_found","message":"No service request has the ID 1.","request_id":"5d784c7f4050","fields":[]}}

Ce que cela nous apprend. La demande 64 246 800 concerne un panneau de limite de vitesse manquant dans le sud de Manhattan. Elle a été signalée par téléphone le 3 mars 2025 et fermée 0,97 jour plus tard. Un identifiant inconnu donne 404 avec le code not_found (introuvable), dans le même format que toutes les autres erreurs. La vue détaillée du tableau de bord peut afficher « introuvable » pour ce code, sans lire le message.

Partie 4 : des filtres sans ambiguïté

Le problème : une faute de frappe qui renvoie toute la ville

Le tableau de bord de Priya demande ?borough=Bronx. La base de données stocke bronx. Une API simple transmet le texte au SQL, ne trouve aucune ligne et renvoie une liste vide. Le tableau de bord affiche « aucune demande dans le Bronx ». C’est faux, et personne ne le remarque. Ou le tableau de bord écrit ?boro=bronx. Une API simple ignore le paramètre inconnu et renvoie des demandes de tous les arrondissements sous le titre « Bronx ».

Les deux erreurs ressemblent à des réponses valides. C’est pire qu’une erreur, parce qu’une erreur se remarque. L’API doit donc vérifier chaque paramètre et rejeter ce qu’elle ne connaît pas. Le cours a fait la même chose pour les champs inconnus d’un corps, dans Validation et réponses d’erreur : rejeter les champs inconnus.

L’idée : un modèle pour la chaîne de requête

FastAPI peut lire les paramètres de requête dans un modèle Pydantic, comme il lit un corps JSON. Le modèle donne à chaque paramètre un type, des limites et une description, et extra="forbid" rejette les noms inconnus. Les modèles nécessitent des dates et des heures. Modifiez donc les premières lignes de city_api/models.py :

city_api/models.py (imports)

from datetime import date
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, NaiveDatetime

Ajoutez ensuite les modèles de filtres après EXAMPLE_REQUEST :

city_api/models.py (add, after EXAMPLE_REQUEST)

class RequestFilters(BaseModel):
    """The filters that the list and the statistics share."""

    model_config = ConfigDict(extra="forbid")

    borough: Borough | None = Field(
        default=None, description="Only requests in this borough."
    )
    complaint_type: str | None = Field(
        default=None,
        min_length=1,
        max_length=60,
        description="Only this complaint type, exactly as in /v1/complaint-types.",
    )


class ListQuery(RequestFilters):
    status: Status | None = Field(default=None, description="Only this status.")
    created_from: NaiveDatetime | None = Field(
        default=None,
        description="Created at or after this time (New York local time, no time zone). "
        "A date such as 2025-03-01 means its first second.",
    )
    created_to: NaiveDatetime | None = Field(
        default=None,
        description="Created before this time (not included).",
    )
    limit: int = Field(default=50, ge=1, le=200, description="Requests per page.")

Choix	Pourquoi
Borough et Status sont des listes fixes	Bronx ou unspecified donne 422 avec fields: ["borough"], pas une liste vide.
NaiveDatetime	Une heure sans fuseau horaire, comme dans les données. 2025-03-03T07:00:00Z (UTC) donne 422. Comparer une heure UTC aux heures de New York créerait un décalage de quatre ou cinq heures, sans avertissement.
created_from inclus, created_to exclu	Deux intervalles qui se touchent, comme mars et avril, ne comptent jamais une demande deux fois.
limit de 1 à 200	Le client choisit la taille de la page, mais ne peut pas demander toute la table à la fois. La partie 5 explique pourquoi.
extra="forbid"	?boro=bronx donne 422 avec fields: ["boro"].

Ajoutez le modèle de réponse d’une page après ServiceRequest. Il contient un next_cursor, qui reste null (nul) jusqu’à la partie 5 :

city_api/models.py (add, after ServiceRequest)

class ServiceRequestPage(BaseModel):
    """One page of requests, oldest first."""

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "items": [EXAMPLE_REQUEST],
                    "next_cursor": "eyJjIjoiMjAyNS0wMy0wM1QxNzo1NToyMCIsImkiOjY0MjQ2ODAwLCJmIjoiZDVmMjc4NTIifQ",
                }
            ]
        }
    )

    items: list[ServiceRequest]
    next_cursor: str | None = Field(
        description="Send it as cursor to get the next page. null: this is the last page."
    )

Le SQL : des conditions provenant d’une table fixe

Un filtre que le client n’a pas envoyé ne doit pas apparaître dans le SQL. La clause WHERE (où) est donc construite à partir des filtres définis. Les éléments du SQL proviennent d’une table fixe dans notre code, et les valeurs sont transmises comme paramètres. Ajoutez ceci à city_api/queries.py, après COLUMNS :

city_api/queries.py (add, after COLUMNS)

# The filters that a request may use, and the condition each one adds.
# A filter that is not in this table cannot reach the SQL.
CONDITIONS = {
    "borough": "borough = ?",
    "complaint_type": "complaint_type = ?",
    "status": "status = ?",
    "created_from": "created_at >= ?",
    "created_to": "created_at < ?",
}


def where(filters: dict) -> tuple[str, list]:
    """Build the WHERE clause and its parameters from the filters that are set."""
    parts, params = [], []
    for name, value in filters.items():
        if value is not None:
            parts.append(CONDITIONS[name])
            params.append(value)
    return (" AND ".join(parts) or "1 = 1"), params

Voici un petit exemple. Pour {"borough": "bronx", "status": None, "created_from": "2025-03-01T00:00:00"}, where renvoie "borough = ? AND created_at >= ?" et ["bronx", "2025-03-01T00:00:00"]. Sans filtre, la fonction renvoie "1 = 1", une condition toujours vraie. Les heures peuvent être comparées comme du texte, car le format 2025-03-01T07:00:00 se trie dans l’ordre chronologique.

Pour l’instant, la liste renvoie seulement la première page. Ajoutez cette fonction après get_request. La partie 5 la remplacera :

city_api/queries.py (add; replaced in Part 5)

def list_requests(con: sqlite3.Connection, filters: dict, limit: int) -> list[sqlite3.Row]:
    condition, params = where(filters)
    sql = (
        f"SELECT {COLUMNS} FROM service_requests WHERE {condition} "
        "ORDER BY created_at, id LIMIT ?"
    )
    return con.execute(sql, [*params, limit]).fetchall()

La route

Dans city_api/routes.py, ajoutez from datetime import datetime après from collections.abc import Iterator. Modifiez ensuite deux lignes d’importation :

city_api/routes.py (change two imports)

from fastapi import APIRouter, Depends, Path, Query, Request

from city_api.models import ErrorResponse, ListQuery, ServiceRequest, ServiceRequestPage

Ajoutez deux fonctions utilitaires après Database. Ajoutez la route de liste au-dessus de get_service_request, pour que la documentation la présente en premier :

city_api/routes.py (add, after Database)

def iso(moment: datetime | None) -> str | None:
    """2025-03-01 00:00 -> '2025-03-01T00:00:00', the format of the database."""
    return moment.isoformat() if moment else None


def check_range(start, end) -> None:
    if start is not None and end is not None and start >= end:
        raise ApiError(
            422,
            "invalid_range",
            "created_from must be before created_to.",
            ["created_from", "created_to"],
        )


@router.get(
    "/service-requests",
    tags=["service requests"],
    summary="List service requests, one page at a time",
)
def list_service_requests(
    query: Annotated[ListQuery, Query()], con: Database
) -> ServiceRequestPage:
    check_range(query.created_from, query.created_to)
    filters = {
        "borough": query.borough,
        "complaint_type": query.complaint_type,
        "status": query.status,
        "created_from": iso(query.created_from),
        "created_to": iso(query.created_to),
    }
    rows = queries.list_requests(con, filters, query.limit)
    return ServiceRequestPage(items=[dict(row) for row in rows], next_cursor=None)

Annotated[ListQuery, Query()] indique à FastAPI de remplir le modèle à partir de la chaîne de requête. check_range rejette un intervalle vide ou inversé avec son propre code, invalid_range (intervalle invalide). Chaque date est valide individuellement, et seule la paire est incorrecte. Un code distinct aide donc le client.

Essayez un filtre et trois erreurs :

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=staten_island&created_from=2025-03-03T07:00:00&limit=2"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=Queens"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?boro=queens"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?created_from=2025-03-05&created_to=2025-03-01"

Sortie

{"items":[{"id":64250466,"created_at":"2025-03-03T07:00:00","closed_at":"2025-03-04T11:00:00","status":"closed","complaint_type":"Street Condition","descriptor":"Pothole","location_type":null,"borough":"staten_island","community_board":"03 STATEN ISLAND","council_district":51,"incident_zip":"10312","channel":null,"latitude":40.527655151030274,"longitude":-74.16543676128964,"days_to_close":1.17},{"id":64250471,"created_at":"2025-03-03T07:00:00","closed_at":"2025-03-04T10:30:00","status":"closed","complaint_type":"Street Condition","descriptor":"Pothole","location_type":null,"borough":"staten_island","community_board":"03 STATEN ISLAND","council_district":null,"incident_zip":"10312","channel":null,"latitude":null,"longitude":null,"days_to_close":1.15}],"next_cursor":null}
{"error":{"code":"invalid_request","message":"The request is not valid.","request_id":"cced4a993526","fields":["borough"]}}
{"error":{"code":"invalid_request","message":"The request is not valid.","request_id":"e289414ce95c","fields":["boro"]}}
{"error":{"code":"invalid_range","message":"created_from must be before created_to.","request_id":"b4c504655829","fields":["created_from","created_to"]}}

Ce que cela nous apprend. Le filtre fonctionne : deux nids-de-poule du lot de 7 h, dans l’ordre chronologique, puis par identifiant. Le deuxième n’a ni coordonnées ni district municipal. Les données réelles ont des lacunes. L’API les montre avec null au lieu de cacher la ligne. Chaque erreur donne maintenant 422 et nomme le paramètre. Priya trouve donc son bogue en quelques secondes. Mais regardez next_cursor : sa valeur est null, et il reste d’autres demandes de Staten Island après ces deux-là. Le client n’a pas encore de moyen de les obtenir.

Partie 5 : des pages sans omissions ni répétitions

Le problème : trop de données pour une seule réponse

Un tableau de bord qui affiche toutes les demandes du semestre a besoin de 109 915 lignes. En JSON, 200 demandes occupent environ 76 Ko. Toutes les demandes occupent donc environ 42 Mo dans une seule réponse. Le serveur la construit en mémoire, le réseau la transporte et le navigateur l’analyse, pendant que l’utilisateur attend. Un deuxième tableau de bord demande la même chose, puis un troisième. L’API envoie donc les données par pages : un nombre limité de lignes par réponse, avec un moyen de demander la page suivante.

La première idée, et pourquoi elle échoue : OFFSET

SQL peut sauter des lignes : LIMIT 50 OFFSET 100 signifie « sauter 100 lignes, puis en donner 50 ». La page 3 utilise OFFSET 100. C’est simple, et beaucoup d’API l’utilisent. Cette méthode pose deux problèmes.

Problème 1 : les données changent entre deux pages. Un tableau de bord lit les demandes ouvertes, deux par page. Entre la page 1 et la page 2, la ville ferme la demande 102. Le projet de départ contient un petit script qui montre ce qui se passe. Il utilise six demandes dans une base de données en mémoire :

python scripts/offset_problem.py

Sortie

With OFFSET:
  page 1: [101, 102]
  request 102 is closed
  page 2: [104, 105]
With a keyset (after the last row of the page):
  page 1: [101, 102]
  request 102 is closed
  page 2: [103, 104]

Ce que cela nous apprend. Avec OFFSET 2, la page 2 saute la demande 103. La demande 102 a quitté la liste des demandes ouvertes. Chaque ligne suivante a donc remonté d’une place. La demande 103 occupe maintenant une place déjà couverte par la page 1. Personne ne reçoit d’erreur : le tableau de bord ne voit tout simplement jamais la demande 103. Si une demande avait plutôt été ajoutée au début, une ligne apparaîtrait deux fois. Notre fichier de base de données ne change pas pendant que l’API fonctionne. Mais l’équipe prévoit une actualisation quotidienne. Un tableau de bord qui parcourt les pages pendant ce changement rencontre ce problème.

Problème 2 : les pages éloignées deviennent plus lentes. Pour sauter 100 000 lignes, SQLite doit quand même les parcourir. Le deuxième script mesure une page de 50 lignes à différentes profondeurs, dans la vraie base de données :

python scripts/time_pages.py

Output (the times change a little on each run)

page after       0 rows: OFFSET  0.05 ms   keyset  0.06 ms
page after  10,000 rows: OFFSET  0.17 ms   keyset  0.06 ms
page after  50,000 rows: OFFSET  0.64 ms   keyset  0.06 ms
page after 100,000 rows: OFFSET  1.47 ms   keyset  0.06 ms

Ce que cela nous apprend. Avec OFFSET, le temps augmente avec la profondeur de la page : 25 fois plus lent à 100 000 lignes. Avec la pagination par clés, il reste constant. Pour ce fichier, 1,5 ms reste rapide. Le premier problème est donc la vraie raison de changer ici. Mais la ville reçoit des millions de demandes par année. Dans une table 30 fois plus grande, la dernière page prendrait environ 45 ms. Le coût augmente avec chaque nouvelle ligne.

L’idée : mémoriser où vous vous êtes arrêté

Pensez à un signet. Vous n’écrivez pas « J’ai lu 100 pages ». Vous placez le signet après la dernière page que vous avez lue. Si quelqu’un retire une page avant lui, le signet reste au bon endroit.

La pagination par clés utilise ce signet pour les lignes. L’API trie selon (created_at, id) et mémorise la dernière ligne de la page. La page suivante demande les lignes après cette ligne :

WHERE (created_at, id) > ('2025-03-03T07:00:00', 64250471)
ORDER BY created_at, id
LIMIT 51

Pourquoi les deux colonnes? Rappelez-vous les 34 demandes à exactement 7 h dans la partie 1. Supposons qu’une page se termine au milieu de ce groupe. Avec seulement created_at > '2025-03-03T07:00:00', la page suivante saute le reste du groupe. Avec created_at >=, elle répète le groupe. L’id départage les lignes, car deux demandes ne peuvent pas avoir le même identifiant. La paire (created_at, id) est donc unique et donne une position exacte dans l’ordre. SQLite compare les paires comme des mots dans un dictionnaire : d’abord created_at, puis id seulement quand les heures sont égales.

Pourquoi LIMIT 51 pour une page de 50? Si 51 lignes reviennent, il existe au moins une autre page. L’API en renvoie 50 et crée un curseur. Si 50 lignes ou moins reviennent, c’est la dernière page, et next_cursor vaut null. Le client ne reçoit donc jamais de dernière page vide.

Le schéma possède un index sur (created_at, id). Il en possède aussi un sur chaque colonne de filtre suivie de (created_at, id). Grâce à eux, SQLite va directement à la position après le signet. C’est pourquoi le temps de la pagination par clés reste constant.

Le curseur : opaque et lié à ses filtres

Le client doit renvoyer le signet. L’API le fournit sous forme d’une seule valeur textuelle, le curseur. Il est opaque : le client le renvoie tel quel, sans le lire ni le construire. L’API peut ainsi modifier son contenu plus tard sans perturber les clients.

Notre curseur est un petit objet JSON encodé en base64. Base64 écrit les données uniquement avec des lettres, des chiffres, - et _. Le curseur peut donc être utilisé dans une URL sans problème. Il contient trois valeurs : le dernier created_at, le dernier id et une courte empreinte des filtres. L’empreinte détecte un bogue courant. Supposons qu’un tableau de bord prenne un curseur dans une liste de demandes du Bronx et l’envoie avec borough=queens. Le signet appartient à une autre liste, donc le résultat n’aurait aucun sens. L’API le refuse avec invalid_cursor (curseur invalide).

Opaque ne signifie pas secret. N’importe qui peut décoder base64, et ce n’est pas un problème. Au pire, un curseur modifié peut faire commencer la liste à un autre endroit. L’API vérifie chaque valeur qu’il contient. Ces valeurs arrivent dans SQL uniquement comme paramètres.

Créez city_api/pagination.py :

city_api/pagination.py

import base64
import binascii
import hashlib
import json
from datetime import datetime

from city_api.errors import ApiError


def fingerprint(filters: dict) -> str:
    """A short code for a set of filters, so that a cursor works only with them."""
    text = json.dumps(filters, sort_keys=True, default=str)
    return hashlib.sha256(text.encode()).hexdigest()[:8]


def encode_cursor(created_at: str, request_id: int, filters: dict) -> str:
    """The position after the last request of a page, as an opaque text."""
    data = {"c": created_at, "i": request_id, "f": fingerprint(filters)}
    raw = json.dumps(data, separators=(",", ":")).encode()
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")


def decode_cursor(cursor: str, filters: dict) -> tuple[str, int]:
    """Return (created_at, id) from a cursor, or raise ApiError invalid_cursor."""
    try:
        raw = base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4))
        data = json.loads(raw)
        created_at, request_id = data["c"], data["i"]
        datetime.fromisoformat(created_at)
        if type(request_id) is not int:
            raise ValueError("id is not an integer")
    except (binascii.Error, UnicodeDecodeError, ValueError, KeyError, TypeError):
        raise ApiError(
            422, "invalid_cursor", "The cursor is not valid.", ["cursor"]
        ) from None
    if data.get("f") != fingerprint(filters):
        raise ApiError(
            422,
            "invalid_cursor",
            "The cursor belongs to other filters. Start again without a cursor.",
            ["cursor"],
        )
    return created_at, request_id

decode_cursor vérifie tout ce qui peut être incorrect. Le texte peut ne pas être du base64 ou du JSON, ou ne pas contenir c ou i. L’heure peut être invalide, ou l’identifiant ne pas être un nombre entier. Chaque problème donne le même code clair 422, jamais un 500. rstrip("=") retire le remplissage que base64 ajoute à la fin, et decode_cursor le remet.

Utilisez maintenant le curseur. Dans city_api/models.py, ajoutez le paramètre à la fin de ListQuery :

city_api/models.py (add to ListQuery)

    cursor: str | None = Field(
        default=None,
        max_length=200,
        description="next_cursor from the previous page. Use it with the same filters.",
    )

Dans city_api/queries.py, remplacez list_requests par cette version :

city_api/queries.py (replace list_requests)

def list_requests(
    con: sqlite3.Connection,
    filters: dict,
    after: tuple[str, int] | None,
    limit: int,
) -> list[sqlite3.Row]:
    """One page, oldest first. Asks for limit + 1 rows to know if more exist."""
    condition, params = where(filters)
    if after is not None:
        condition += " AND (created_at, id) > (?, ?)"
        params += list(after)
    sql = (
        f"SELECT {COLUMNS} FROM service_requests WHERE {condition} "
        "ORDER BY created_at, id LIMIT ?"
    )
    return con.execute(sql, [*params, limit + 1]).fetchall()

Dans city_api/routes.py, importez les deux fonctions de curseur après l’importation des modèles :

city_api/routes.py (add to the imports)

from city_api.pagination import decode_cursor, encode_cursor

Remplacez ensuite la route de liste par la version finale. Elle ajoute une description pour la documentation et les six dernières lignes de la fonction :

city_api/routes.py (replace the list route)

@router.get(
    "/service-requests",
    tags=["service requests"],
    summary="List service requests, one page at a time",
    description=(
        "Requests that match every filter, oldest first (by created_at, then id). "
        "Each page has at most `limit` requests. If `next_cursor` is not null, send it "
        "as `cursor`, with the same filters, to get the next page."
    ),
)
def list_service_requests(
    query: Annotated[ListQuery, Query()], con: Database
) -> ServiceRequestPage:
    check_range(query.created_from, query.created_to)
    filters = {
        "borough": query.borough,
        "complaint_type": query.complaint_type,
        "status": query.status,
        "created_from": iso(query.created_from),
        "created_to": iso(query.created_to),
    }
    after = decode_cursor(query.cursor, filters) if query.cursor else None
    rows = queries.list_requests(con, filters, after, query.limit)
    page = rows[: query.limit]
    next_cursor = None
    if len(rows) > query.limit:
        last = page[-1]
        next_cursor = encode_cursor(last["created_at"], last["id"], filters)
    return ServiceRequestPage(items=[dict(row) for row in page], next_cursor=next_cursor)

Le limit ne fait pas partie de l’empreinte. Un client peut donc lire une première page de 5 lignes, puis le reste par pages de 200.

Pagination par clés : le curseur est un signet après la dernière ligne, et la page suivante commence là.

Description textuelle du diagramme

Le tableau de bord demande la première page des demandes du Bronx, 50 à la fois. L’API demande à SQLite 51 lignes, triées par created_at, puis par id. 51 lignes reviennent, donc une page suivante existe. L’API envoie les 50 premières lignes et un curseur contenant le created_at et l’id de la ligne 50. Le tableau de bord envoie de nouveau la même requête, avec le curseur. L’API le décode et vérifie que les filtres sont identiques. Elle demande ensuite à SQLite les lignes après cette paire, encore une fois 51 au maximum. SQLite trouve immédiatement la position dans l’index. À la dernière page, moins de 51 lignes reviennent, et next_cursor vaut null.

Suivez le curseur à la main. Demandez de nouveau les deux premières demandes de Staten Island. Copiez le next_cursor de la réponse dans la deuxième requête. Envoyez ensuite le même curseur avec un autre arrondissement :

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=staten_island&created_from=2025-03-03T07:00:00&limit=2"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=staten_island&created_from=2025-03-03T07:00:00&limit=2&cursor=eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUwNDcxLCJmIjoiNTIxODUyYWEifQ"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=queens&limit=2&cursor=eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUwNDcxLCJmIjoiNTIxODUyYWEifQ"

Output (each request shortened to its id and created_at)

{"items":[{"id":64250466,"created_at":"2025-03-03T07:00:00",...},{"id":64250471,"created_at":"2025-03-03T07:00:00",...}],"next_cursor":"eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUwNDcxLCJmIjoiNTIxODUyYWEifQ"}
{"items":[{"id":64250476,"created_at":"2025-03-03T07:00:00",...},{"id":64251586,"created_at":"2025-03-03T07:00:00",...}],"next_cursor":"eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUxNTg2LCJmIjoiNTIxODUyYWEifQ"}
{"error":{"code":"invalid_cursor","message":"The cursor belongs to other filters. Start again without a cursor.","request_id":"5526ce765f49","fields":["cursor"]}}

Ce que cela nous apprend. La page 1 se termine par la demande 64250471 à 7 h, à l’intérieur du lot de 34. La page 2 continue avec 64250476, l’identifiant suivant à la même heure. Aucune demande du lot n’est sautée ni répétée. Une fois décodé, le curseur est {"c":"2025-03-03T07:00:00","i":64250471,"f":"521852aa"} : exactement le signet. Le même curseur avec d’autres filtres est refusé. Un message indique au client quoi faire.

Erreur courante

Le tableau de bord affiche toujours les mêmes demandes ou s'arrête après une page

Symptôme : le tableau de bord de Priya affiche les 50 premières demandes à répétition, jusqu’à ce qu’il s’arrête. Ou il n’affiche que 50 demandes, même si des milliers correspondent au filtre.

Cause : le client n’utilise pas next_cursor correctement. Dans le premier cas, il lit next_cursor, mais envoie la requête suivante sans lui. L’API recommence donc au début chaque fois. Dans le deuxième cas, il s’arrête après la première page plutôt que lorsque next_cursor vaut null.

Solution : la boucle du client suit une seule règle. Elle envoie le curseur de la dernière réponse, avec les mêmes filtres, jusqu’à ce que next_cursor vaille null. En Python, avec httpx :

cursor = None
while True:
    params = {"borough": "bronx", "limit": 200}
    if cursor:
        params["cursor"] = cursor
    page = httpx.get(url, params=params, headers=headers).json()
    handle(page["items"])
    cursor = page["next_cursor"]
    if cursor is None:
        break

Le test test_paging_gives_every_request_once de la partie 7 exécute cette boucle avec l’API.

Partie 6 : types de plaintes et chiffres quotidiens

Les types de plaintes, pour les menus

Les tableaux de bord ont besoin d’un menu des types de plaintes, avec les plus courants en premier. Un seul GROUP BY suffit. Ajoutez-le à city_api/queries.py :

city_api/queries.py (add)

def complaint_types(con: sqlite3.Connection) -> list[sqlite3.Row]:
    return con.execute(
        "SELECT complaint_type, COUNT(*) AS count FROM service_requests "
        "GROUP BY complaint_type ORDER BY count DESC, complaint_type"
    ).fetchall()

complaint_type comme deuxième clé de tri donne un ordre fixe quand deux types ont le même nombre de demandes. Le menu ne change donc jamais entre deux appels.

Les chiffres quotidiens : quoi compter

La revue du lundi pose trois questions pour chaque jour. Combien de demandes sont arrivées? Combien les équipes ont-elles traitées rapidement? Combien de temps ont-elles généralement pris? Chaque question exige une définition exacte. Sinon, deux tableaux de bord comptent encore différemment.

• opened (ouvertes) : les demandes créées ce jour-là.
• closed_within_14_days (fermées en 14 jours ou moins) : parmi elles, celles avec un closed_at et un days_to_close de 14 ou moins. Quatorze jours est la cible de l’équipe. L’API utilise closed_at, pas status, car la ville ne remplit pas le statut de façon uniforme. 203 demandes ont une heure de fermeture, mais un statut comme assigned (attribuée). L’étude de cas du cours SQL et préparation des données pour l’IA utilise la même définition. Les deux cours comptent donc les mêmes demandes.
• share_closed_within_14_days (proportion fermée en 14 jours ou moins) : le deuxième nombre divisé par le premier, de 0 à 1.
• median_days_to_close (nombre médian de jours avant fermeture) : la valeur du milieu des days_to_close des demandes avec un closed_at.

Pourquoi la médiane plutôt que la moyenne? Regardez les onze demandes du 3 mars dans les données de test de la partie 7. Dix ont été fermées en 4 jours ou moins, et une réparation de bordure a pris 60,98 jours. La moyenne est de 6,79 jours, soit plus que le temps pris par dix des onze demandes. La médiane est de 1,15 jour : la moitié ont été fermées plus vite, l’autre moitié plus lentement. Une demande lente déplace beaucoup la moyenne, mais peu la médiane. Pour « combien de temps prend généralement une demande », la médiane donne une réponse fidèle. La proportion fermée en 14 jours ou moins montre les demandes lentes.

SQLite n’a pas de fonction MEDIAN. SQL compte donc les demandes, et Python calcule chaque médiane à partir des valeurs, avec statistics.median. Ajoutez cette fonction à city_api/queries.py :

city_api/queries.py (add)

def daily_stats(con: sqlite3.Connection, filters: dict, start: date, end: date) -> list[dict]:
    """Opened, closed within 14 days and the median days to close, for each day."""
    condition, params = where(
        {**filters, "created_from": start.isoformat(), "created_to": end.isoformat()}
    )
    counts = con.execute(
        "SELECT substr(created_at, 1, 10) AS day, COUNT(*) AS opened, "
        "SUM(CASE WHEN closed_at IS NOT NULL AND days_to_close <= 14 THEN 1 ELSE 0 END) AS within_14 "
        f"FROM service_requests WHERE {condition} GROUP BY day",
        params,
    ).fetchall()
    # SQLite has no MEDIAN function, so Python computes it from the values.
    durations: dict[str, list[float]] = {}
    for day, days in con.execute(
        "SELECT substr(created_at, 1, 10), days_to_close FROM service_requests "
        f"WHERE {condition} AND closed_at IS NOT NULL",
        params,
    ):
        durations.setdefault(day, []).append(days)

    by_day = {row["day"]: row for row in counts}
    result = []
    for offset in range((end - start).days):
        day = (start + timedelta(days=offset)).isoformat()
        row = by_day.get(day)
        opened = row["opened"] if row else 0
        within = row["within_14"] if row else 0
        values = durations.get(day)
        result.append(
            {
                "date": day,
                "opened": opened,
                "closed_within_14_days": within,
                "share_closed_within_14_days": round(within / opened, 3) if opened else None,
                "median_days_to_close": round(statistics.median(values), 2) if values else None,
            }
        )
    return result

Partie	Pourquoi
substr(created_at, 1, 10)	Les 10 premiers caractères de 2025-03-03T07:00:00 donnent le jour, 2025-03-03.
SUM(CASE WHEN ... THEN 1 ELSE 0 END)	Compte les lignes où la condition est vraie.
La boucle sur chaque jour	Un jour sans demandes apparaît dans le résultat, avec opened à 0. Un graphique où il manque un jour trace une ligne droite par-dessus, et le lecteur ne voit pas le trou.
None pour la proportion et la médiane	0 divisé par 0 n’a pas de réponse. null signifie « aucune donnée », mais 0 signifierait « aucune demande fermée », ce qui est faux.

Le point de terminaison des statistiques exige une limite sur la période. Demander cinq années de chiffres quotidiens pour chaque arrondissement représente beaucoup de travail pour un seul appel. Personne ne lit un tel graphique. Les tableaux de bord affichent au plus un trimestre. La limite est donc de 92 jours, soit le trimestre le plus long. Une période plus longue reçoit son propre code, range_too_large (période trop longue). Le client sait ainsi quoi changer. Ajoutez le modèle de requête à city_api/models.py, après ListQuery :

city_api/models.py (add, after ListQuery)

class StatsQuery(RequestFilters):
    created_from: date = Field(description="The first day, for example 2025-03-01.")
    created_to: date = Field(
        description="The day after the last day (not included). At most 92 days after created_from."
    )

Ajoutez les modèles de réponse après ServiceRequestPage :

city_api/models.py (add, after ServiceRequestPage)

class ComplaintTypeCount(BaseModel):
    complaint_type: str
    count: int = Field(description="Requests of this type in the whole database.")


class ComplaintTypeList(BaseModel):
    items: list[ComplaintTypeCount] = Field(description="The most common type first.")


class DailyStat(BaseModel):
    date: date
    opened: int = Field(description="Requests created on this day.")
    closed_within_14_days: int = Field(
        description="Of those, the ones with a closed_at at most 14 days after created_at."
    )
    share_closed_within_14_days: float | None = Field(
        description="closed_within_14_days / opened, from 0 to 1. null when opened is 0."
    )
    median_days_to_close: float | None = Field(
        description="The middle value of days_to_close of the requests with a closed_at. null when none."
    )


class DailyStats(BaseModel):
    """One row for every day of the range, also days without requests."""

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "items": [
                        {
                            "date": "2025-03-03",
                            "opened": 752,
                            "closed_within_14_days": 680,
                            "share_closed_within_14_days": 0.904,
                            "median_days_to_close": 1.52,
                        }
                    ]
                }
            ]
        }
    )

    items: list[DailyStat]


class DatasetInfo(BaseModel):
    """Where the data comes from, and how it was changed."""

    source: str
    pulled_on: str = Field(description="The snapshot date: statuses are as of this day.")
    created_from: str
    created_to: str = Field(description="Not included.")
    rows: int
    rules: str

DatasetInfo sert à un autre petit point de terminaison, /v1/dataset. Grâce à lui, chaque tableau de bord peut afficher « données au 7 octobre 2026 », une information essentielle pour les gestionnaires. Sa requête lit la table meta. Ajoutez-la à city_api/queries.py :

city_api/queries.py (add)

def dataset_info(con: sqlite3.Connection) -> dict[str, str]:
    return dict(con.execute("SELECT name, value FROM meta").fetchall())

Passons aux trois routes. Dans city_api/routes.py, la ligne from city_api.models import ErrorResponse, ListQuery, ServiceRequest, ServiceRequestPage importe quatre modèles. Remplacez cette seule ligne par cette liste complète de huit modèles :

city_api/routes.py (replace the import of the models)

from city_api.models import (
    ComplaintTypeList,
    DailyStats,
    DatasetInfo,
    ErrorResponse,
    ListQuery,
    ServiceRequest,
    ServiceRequestPage,
    StatsQuery,
)

Les autres importations restent telles quelles. Ajoutez ensuite la constante MAX_STATS_DAYS après la dernière importation, from city_api.security import require_api_key. Laissez une ligne vide avant elle :

city_api/routes.py (add after the imports)

MAX_STATS_DAYS = 92

Ajoutez ensuite les routes à la fin du fichier :

city_api/routes.py (add at the end)

@router.get(
    "/complaint-types",
    tags=["service requests"],
    summary="List the complaint types, with the number of requests",
)
def list_complaint_types(con: Database) -> ComplaintTypeList:
    return ComplaintTypeList(items=[dict(row) for row in queries.complaint_types(con)])


@router.get(
    "/stats/daily",
    tags=["statistics"],
    summary="Count the requests opened and closed for each day",
    description=(
        "For each day from `created_from` to the day before `created_to`: the requests "
        "opened, how many of them were closed within 14 days, and the median days to "
        f"close. The range is at most {MAX_STATS_DAYS} days. Days without requests are "
        "included, with `opened` 0."
    ),
)
def daily_stats(query: Annotated[StatsQuery, Query()], con: Database) -> DailyStats:
    check_range(query.created_from, query.created_to)
    if (query.created_to - query.created_from).days > MAX_STATS_DAYS:
        raise ApiError(
            422,
            "range_too_large",
            f"The range is longer than {MAX_STATS_DAYS} days. Ask for a shorter one.",
            ["created_from", "created_to"],
        )
    filters = {"borough": query.borough, "complaint_type": query.complaint_type}
    items = queries.daily_stats(con, filters, query.created_from, query.created_to)
    return DailyStats(items=items)


@router.get(
    "/dataset",
    tags=["statistics"],
    summary="Where the data comes from, and its snapshot date",
)
def dataset(con: Database) -> DatasetInfo:
    return DatasetInfo(**queries.dataset_info(con))

Les statistiques partagent RequestFilters avec la liste. Ainsi, ?borough=Bronx et les paramètres inconnus échouent ici de la même façon. Essayez le menu et les chiffres :

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/complaint-types"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/stats/daily?created_from=2025-03-01&created_to=2025-03-08&complaint_type=Sidewalk%20Condition"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/stats/daily?created_from=2025-01-01&created_to=2025-06-01"

Output (the complaint types shortened to the first four)

{"items":[{"complaint_type":"Street Condition","count":39381},{"complaint_type":"Traffic Signal Condition","count":26576},{"complaint_type":"Street Light Condition","count":14342},{"complaint_type":"Sidewalk Condition","count":12061},...]}
{"items":[{"date":"2025-03-01","opened":37,"closed_within_14_days":30,"share_closed_within_14_days":0.811,"median_days_to_close":9.81},{"date":"2025-03-02","opened":15,"closed_within_14_days":10,"share_closed_within_14_days":0.667,"median_days_to_close":8.62},{"date":"2025-03-03","opened":77,"closed_within_14_days":62,"share_closed_within_14_days":0.805,"median_days_to_close":3.94},{"date":"2025-03-04","opened":98,"closed_within_14_days":78,"share_closed_within_14_days":0.796,"median_days_to_close":7.71},{"date":"2025-03-05","opened":100,"closed_within_14_days":54,"share_closed_within_14_days":0.54,"median_days_to_close":12.87},{"date":"2025-03-06","opened":70,"closed_within_14_days":51,"share_closed_within_14_days":0.729,"median_days_to_close":9.43},{"date":"2025-03-07","opened":65,"closed_within_14_days":55,"share_closed_within_14_days":0.846,"median_days_to_close":10.3}]}
{"error":{"code":"range_too_large","message":"The range is longer than 92 days. Ask for a shorter one.","request_id":"895aff6d300a","fields":["created_from","created_to"]}}

Ce que cela nous apprend. Il existe 31 types de plaintes, et quatre représentent 84 % de toutes les demandes. Les problèmes de chaussée, comme les nids-de-poule, forment le plus grand groupe. La semaine de demandes concernant les trottoirs montre pourquoi les gestionnaires veulent ce point de terminaison. Les trottoirs prennent du temps : la médiane quotidienne va de 4 à 13 jours. Seulement 54 % des demandes du 5 mars ont été fermées en 14 jours ou moins. Avec les mêmes requêtes pour tout le mois de mars, les quatre grands types donnent un portrait clair. De 95 % à 96 % des demandes concernant la chaussée et les feux de circulation ont été fermées en 14 jours ou moins. Seulement 78 % des demandes concernant les trottoirs l’ont été, et 73 % de celles concernant l’éclairage public. C’est le genre de différence que la revue du lundi vise à examiner. La période de cinq mois est refusée avec un code clair. Une période de 92 jours prend environ 55 ms sur un ordinateur portable.

Partie 7 : tests

Une petite base de données pour les tests

Les tests ne doivent pas utiliser le téléchargement de 36 Mo. Il est lent à récupérer, et un test ne doit pas dépendre du réseau. Vous ne pouvez pas non plus vérifier les nombres à la main sur 109 915 lignes. Donc, tests/fixture_rows.csv contient 30 lignes réelles de la base de données, choisies pour couvrir les cas difficiles :

• six demandes du lot de 7 h le 3 mars, toutes à la même seconde;
• des demandes du 1er au 5 mars, de plusieurs types, statuts et arrondissements;
• deux demandes sans arrondissement, et une demande en attente le 1er mars;
• une demande attribuée avec une heure de fermeture, et une réparation de bordure qui a pris 60,98 jours;
• une deuxième demande en attente le 12 mars, une semaine après les autres, pour laisser des jours sans demandes entre les deux.

tests/conftest.py construit la base de données à partir de ces lignes, avec le même schema.sql que la vraie base. Il le fait dans un dossier temporaire, une fois par exécution des tests. Chaque client de test reçoit ses paramètres, comme dans Configuration et journalisation : des tests avec leurs propres paramètres. Le fichier conftest du projet de départ inscrit le nombre 1 comme version du schéma. Maintenant que db.py existe, utilisez plutôt sa constante. Dans tests/conftest.py, ajoutez from city_api.db import SCHEMA_VERSION aux importations, et modifiez la ligne contenant PRAGMA user_version :

tests/conftest.py (change)

    con.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")

Le test le plus important : parcourir toutes les pages

La promesse de la partie 5 est « aucune demande omise, aucune répétée ». Un test peut vérifier exactement cela. Il suit next_cursor jusqu’à la fin, recueille tous les identifiants et les compare à ceux d’une seule grande page. Il le fait avec des pages de 1, 2, 4, 7, 29, 30 et 31 lignes. Certaines de ces tailles divisent 30, d’autres non, et certaines sont plus grandes. Voici le début de tests/test_pagination.py :

tests/test_pagination.py (start)

import base64
import json

import pytest


def walk(client, limit, **filters) -> tuple[list[int], int]:
    """Follow next_cursor to the end. Return every id and the number of pages."""
    ids, pages, cursor = [], 0, None
    while True:
        params = {"limit": limit, **filters}
        if cursor:
            params["cursor"] = cursor
        response = client.get("/v1/service-requests", params=params)
        assert response.status_code == 200, response.text
        body = response.json()
        assert len(body["items"]) <= limit
        ids += [item["id"] for item in body["items"]]
        pages += 1
        cursor = body["next_cursor"]
        if cursor is None:
            return ids, pages


def one_page(client, **filters) -> list[int]:
    response = client.get("/v1/service-requests", params={"limit": 200, **filters})
    return [item["id"] for item in response.json()["items"]]


@pytest.mark.parametrize("limit", [1, 2, 4, 7, 29, 30, 31])
def test_paging_gives_every_request_once(client, limit):
    ids, pages = walk(client, limit)
    assert len(ids) == len(set(ids)) == 30  # no duplicates, no gaps
    assert ids == one_page(client)  # in the same order
    assert pages == -(-30 // limit)  # 30 / limit, rounded up

len(ids) == len(set(ids)) == 30 vérifie les deux promesses à la fois. Un ensemble conserve chaque identifiant une seule fois. Des longueurs égales signifient donc aucun doublon, et 30 signifie aucune omission. -(-30 // limit) correspond à 30 divisé par la limite, arrondi vers le haut. Pour limit=4, 7 pages de 4 lignes et 1 page de 2 donnent 8 pages. Le test prouve donc aussi qu’aucune dernière page vide n’est envoyée. Le reste du fichier contient cinq autres tests. Ils divisent le lot de 7 h, utilisent des filtres et changent la limite. Ils envoient aussi des curseurs invalides et des curseurs provenant d’autres filtres.

Les règles sous forme de tests

Trois autres tests protègent les règles des parties 2 à 4. Ils se trouvent dans tests/test_service_requests.py :

tests/test_service_requests.py (three of its tests)

def test_sql_in_a_filter_is_only_text(client):
    attack = "x' OR '1'='1"
    response = client.get("/v1/service-requests", params={"complaint_type": attack})
    assert response.status_code == 200
    assert response.json()["items"] == []


def test_the_connection_cannot_write(fixture_db):
    con = db.connect(fixture_db)
    with pytest.raises(sqlite3.OperationalError, match="readonly"):
        con.execute("DELETE FROM service_requests")
    con.close()


@pytest.mark.parametrize(
    ("params", "field"),
    [
        ({"borough": "Queens"}, "borough"),
        ({"borough": "unspecified"}, "borough"),
        ({"status": "done"}, "status"),
        ({"limit": "0"}, "limit"),
        ({"limit": "201"}, "limit"),
        ({"limit": "ten"}, "limit"),
        ({"created_from": "March 3"}, "created_from"),
        ({"created_from": "2025-03-03T07:00:00Z"}, "created_from"),
        ({"complaint_type": "x" * 61}, "complaint_type"),
        ({"boro": "queens"}, "boro"),
    ],
)
def test_invalid_parameters_name_the_field(client, params, field):
    response = client.get("/v1/service-requests", params=params)
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "invalid_request"
    assert response.json()["error"]["fields"] == [field]

Des nombres vérifiés à la main

Un point de terminaison de statistiques peut être faux tout en semblant correct. Le test de /v1/stats/daily compare donc les résultats à des nombres vérifiables sur papier, à partir des 30 lignes. Prenez le 4 mars : quatre demandes, toutes fermées en moins de 14 jours, après 6,02, 1,02, 7,71 et 9,76 jours. Après le tri, les deux valeurs centrales sont 6,02 et 7,71, et leur moyenne est 6,865. La médiane est donc de 6,87 jours, et la proportion est de 4 / 4 = 1,0 :

tests/test_stats.py (one row of test_daily_numbers)

        {"date": "2025-03-04", "opened": 4, "closed_within_14_days": 4,
         "share_closed_within_14_days": 1.0, "median_days_to_close": 6.87},

Exécutez-les

Copiez quatre fichiers du dossier tests du projet terminé dans le vôtre : test_health.py, test_pagination.py, test_service_requests.py et test_stats.py. Le nouveau test_health.py ajoute deux tests pour /ready : un sans fichier de base de données, et un avec une ancienne version du schéma. Lisez chaque fichier, puis exécutez tout :

python -m pytest

Sortie (fin)

tests/test_health.py .....                                               [  7%]
tests/test_pagination.py ...................                             [ 37%]
tests/test_service_requests.py .................................         [ 89%]
tests/test_stats.py .......                                              [100%]

============================== 64 passed in 0.42s ==============================

Ce que cela nous apprend. 64 tests s’exécutent en moins d’une seconde, sans serveur ni téléchargement. Ils couvrent chaque point de terminaison, chaque code d’erreur sauf 500, la promesse de pagination avec sept tailles de page et les deux règles de sécurité. Quand vous modifierez le SQL, une erreur fera échouer un test plutôt qu’un tableau de bord.

Partie 8 : documentation, contrat et Postman

Le contrat révisé

Les tableaux de bord dépendent des noms, des types et des codes d’erreur de l’API. Comme dans Garder la documentation exacte, le projet conserve une copie révisée du document OpenAPI. Un test compare l’API à cette copie. Le projet de départ contient city_api/export_contract.py. Écrivez le contrat :

python -m city_api.export_contract

Sortie

WARNING city_api: API_KEY is not set: /v1 accepts requests without a key
WARNING city_api: API_KEY is not set: /v1 accepts requests without a key
Wrote contract/openapi.json

Les avertissements sont attendus : le script crée l’application sans les paramètres de votre terminal, seulement pour lire son document OpenAPI. Lisez le fichier avant de lui faire confiance. Vérifiez, par exemple, que /v1/service-requests présente chaque paramètre de requête avec sa description. Vérifiez aussi que borough contient un enum avec les cinq codes, et que limit a minimum: 1 et maximum: 200. Ouvrez http://127.0.0.1:8000/docs pour voir le même document dans Swagger UI.

Créez ensuite tests/test_contract.py :

tests/test_contract.py

import json
from pathlib import Path

CONTRACT = Path(__file__).parent.parent / "contract" / "openapi.json"


def test_openapi_matches_the_reviewed_contract(client):
    current = json.dumps(client.get("/openapi.json").json(), indent=2) + "\n"
    reviewed = CONTRACT.read_text(encoding="utf-8")
    assert current == reviewed, (
        "The API contract changed. If the change is intended, run "
        "python -m city_api.export_contract and review the difference."
    )


def test_documented_request_is_the_real_one(client):
    schemas = client.get("/openapi.json").json()["components"]["schemas"]
    example = schemas["ServiceRequest"]["examples"][0]
    response = client.get(f"/v1/service-requests/{example['id']}")
    assert response.json() == example


def test_documented_examples_have_the_real_shape(client):
    # The examples show numbers of the full database; the tests have 30 rows.
    # So the test compares the fields, not the values.
    schemas = client.get("/openapi.json").json()["components"]["schemas"]
    page = client.get("/v1/service-requests", params={"limit": 1}).json()
    assert page.keys() == schemas["ServiceRequestPage"]["examples"][0].keys()
    day = client.get(
        "/v1/stats/daily", params={"created_from": "2025-03-03", "created_to": "2025-03-04"}
    ).json()["items"][0]
    assert day.keys() == schemas["DailyStats"]["examples"][0]["items"][0].keys()

La demande donnée en exemple, 64246800, fait partie des 30 lignes de test. Un test peut donc la vérifier exactement. L’exemple quotidien montre les vrais nombres du 3 mars, présents seulement dans la base complète. Ce test vérifie donc les champs, pas les valeurs. Exécutez de nouveau python -m pytest : 67 tests réussis.

Un README pour l’équipe des tableaux de bord

Un client ne devrait pas avoir besoin du code. Le fichier README.md du projet terminé explique à quoi sert l’API et comment la configurer. Il décrit chaque point de terminaison en une ligne, présente les codes d’erreur et montre comment exécuter les vérifications. Sa dernière section indique la source des données, la date de l’instantané et la licence. Copiez-le, et adaptez-le aux différences de votre projet.

La collection Postman

Le projet terminé contient postman/city-requests-api.postman_collection.json, avec 14 requêtes dans trois dossiers. Il contient aussi les environnements Postman Local (local) et Deployed (déployé), sans clés, comme dans le cours. La plupart des requêtes vérifient un statut et un corps, comme dans le module 5. Deux d’entre elles font quelque chose de nouveau : elles testent la pagination entre deux requêtes. La première requête enregistre des valeurs de sa réponse dans des variables de collection, et la deuxième les utilise :

Tests of 'List the first page' (the end)

pm.collectionVariables.set("nextCursor", body.next_cursor);
pm.collectionVariables.set("firstPageIds", JSON.stringify(body.items.map((r) => r.id)));
pm.collectionVariables.set("lastCreatedAt", body.items[body.items.length - 1].created_at);

Tests of 'Get the next page' (its URL ends with &cursor={{nextCursor}})

const items = pm.response.json().items;
const firstPage = JSON.parse(pm.collectionVariables.get("firstPageIds"));
pm.test("no request of page 1 again", () => items.forEach((r) => pm.expect(firstPage).to.not.include(r.id)));
pm.test("continues after page 1", () => pm.expect(items[0].created_at >= pm.collectionVariables.get("lastCreatedAt")).to.be.true);

Copiez le dossier postman du projet terminé. Avec l’API en marche et API_KEY=local-dev-key, exécutez la collection depuis le deuxième terminal :

postman collection run postman/city-requests-api.postman_collection.json -e postman/local.postman_environment.json --env-var "apiKey=local-dev-key"

Output (start and end)

No authorization data found. Please use the `postman login` command.
Refer:https://learning.postman.com/docs/postman-cli/postman-cli-options/#postman-login

Running your collection...

City Service Requests API

Folder service requests
Sub-folder List the first page

  GET http://127.0.0.1:8000/v1/service-requests?borough=brooklyn&complaint_type=Street%20Condition&created_from=2025-03-01&created_to=2025-04-01&limit=3 [200 OK, 1.38 KB, 17 ms]
  Pass  response is JSON
  Pass  status is 200
  Pass  has 3 requests
  Pass  every request matches the filters
  Pass  has a next_cursor
  Pass  answers in under 500 ms
...
|                requests |                 14 |                 0 |
|              assertions |                 57 |                 0 |

Les premières lignes indiquent seulement que vous n’êtes pas connecté à Postman, ce qui n’est pas nécessaire pour un fichier local. Ce que cela nous apprend. 14 requêtes et 57 vérifications réussissent avec la vraie base de données, via un vrai serveur. pytest a vérifié le code avec 30 lignes. La collection vérifie le service en marche avec toutes les données, et sert de vérification sommaire pour un déploiement. Pour l’utiliser dans l’application Postman, importez les trois fichiers de postman/ (Import, importer).

Partie 9 : déployer l’API (étapes non exécutées par nous)

Non exécuté pour cette étude de cas

Nous n’avons pas déployé cette API. Les étapes ci-dessous suivent le déploiement du cours vers FastAPI Cloud dans Publier en toute sécurité et Vérifier et nettoyer. Une étape est nouvelle et non testée : téléverser la base de données de 36 Mo avec le code. FastAPI Cloud est en version bêta, donc les commandes et les limites peuvent changer. Si vous déployez, considérez le service comme temporaire et supprimez-le après avoir recueilli vos preuves.

Ce qui diffère de l’API de billets

L’API de billets ne contenait que du code. Cette API a aussi besoin de son fichier de données sur le serveur. Git et la commande de déploiement excluent les fichiers indiqués dans .gitignore. .gitignore contient data/*.db, car un fichier de 36 Mo n’a pas sa place dans un dépôt Git. FastAPI Cloud lit aussi un fichier .fastapicloudignore, qui peut réinclure un fichier ignoré avec !. Le projet de départ ne le contient pas. Créez .fastapicloudignore dans le dossier du projet (celui contenant pyproject.toml), avec ce contenu. Le projet terminé contient le même fichier :

.fastapicloudignore

# FastAPI Cloud reads this file together with .gitignore.
# Git ignores the database because it is large, but the server needs it:
!data/service_requests_2025h1.db
# The server does not need these:
tests/
postman/
scripts/

Le téléversement contient alors le code et la base de données, mais pas les tests ni les outils. Comme le fichier est ouvert avec immutable=1, de nouvelles données exigent un nouveau déploiement. La plateforme démarre une nouvelle copie de l’application avec le nouveau fichier. C’est la façon sûre de publier de nouvelles données. Dans notre test local, l’API en marche utilisait moins de 80 Mo de mémoire, bien en dessous des 500 Mio d’une instance gratuite.

Le serveur construit aussi votre projet sous forme de paquet Python. Ce téléversement contient trois dossiers à la racine, city_api/, data/ et contract/. L’outil de construction ne peut pas deviner lequel est le paquet. Le déploiement réel de l’étude de cas Servir un modèle de prix des maisons a échoué pour cette raison précise. Donc, pyproject.toml nomme le paquet :

pyproject.toml

[tool.setuptools]
# The project has more than one top-level folder: city_api/ (the code),
# data/ and contract/. Name the package, so the build does not guess.
packages = ["city_api"]

Vérifiez-le sur votre ordinateur avant de déployer : python -m pip wheel . --no-deps -w build doit terminer avec Successfully built city-requests-api (construction de city-requests-api réussie). Sans ces trois lignes, la commande échoue avec Multiple top-level packages discovered in a flat-layout (plusieurs paquets découverts à la racine). Supprimez ensuite les deux dossiers créés par la vérification, build et city_requests_api.egg-info.

Exécutez l’application comme le serveur le fera

D’abord, la simulation locale du déploiement de Premiers pas avec l’hébergement : mode production, documentation désactivée et collection.

macOS

API_KEY=local-dev-key SHOW_DOCS=false fastapi run

Windows

$env:API_KEY = "local-dev-key"
$env:SHOW_DOCS = "false"
fastapi run

Linux

API_KEY=local-dev-key SHOW_DOCS=false fastapi run

Exécutez la collection depuis un deuxième terminal, comme dans la partie 8. Les 14 requêtes réussissent. Arrêtez le serveur.

Déployez, vérifiez et nettoyez

1. Connectez-vous. Créez ensuite l’application et liez-y le dossier, avant le premier déploiement :

   fastapi login
   fastapi cloud apps create --name city-requests-api --link

2. Créez une nouvelle clé aléatoire et enregistrez-la comme secret. Désactivez la documentation :

   macOS

   KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
   printf %s "$KEY" | fastapi cloud env set API_KEY --secret --value-stdin --no-redeploy
   fastapi cloud env set SHOW_DOCS false --no-redeploy
   echo "$KEY"

   Windows

   $key = python -c "import secrets; print(secrets.token_urlsafe(32))"
   $key | fastapi cloud env set API_KEY --secret --value-stdin --no-redeploy
   fastapi cloud env set SHOW_DOCS false --no-redeploy
   $key

   Linux

   KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
   printf %s "$KEY" | fastapi cloud env set API_KEY --secret --value-stdin --no-redeploy
   fastapi cloud env set SHOW_DOCS false --no-redeploy
   echo "$KEY"

   Copiez la clé affichée par la dernière commande, puis fermez le terminal.

3. Ouvrez un nouveau terminal et activez l’environnement. Déployez ensuite, et notez l’adresse affichée par la commande :

   fastapi deploy

4. Vérifiez les trois réponses de base. Remplacez l’adresse par la vôtre. Sur Windows, tapez curl.exe :

   curl -i https://city-requests-api.fastapicloud.dev/health
   curl -i https://city-requests-api.fastapicloud.dev/ready
   curl -i https://city-requests-api.fastapicloud.dev/v1/complaint-types

   Vous attendez 200, 200 et 401. Si /ready renvoie 503 database_unavailable (base de données indisponible), la base de données n’a pas atteint le serveur. Vérifiez .fastapicloudignore et lisez les journaux (logs) (étape 6).

5. Définissez votre adresse comme baseUrl dans postman/deployed.postman_environment.json, sans / à la fin. Exécutez ensuite la vérification sommaire, en saisissant la clé :

   macOS

   read -s CITY_API_KEY
   postman collection run postman/city-requests-api.postman_collection.json -e postman/deployed.postman_environment.json --env-var "apiKey=$CITY_API_KEY"

   Windows

   $CITY_API_KEY = Read-Host "API key"
   postman collection run postman/city-requests-api.postman_collection.json -e postman/deployed.postman_environment.json --env-var "apiKey=$CITY_API_KEY"

   Linux

   read -s CITY_API_KEY
   postman collection run postman/city-requests-api.postman_collection.json -e postman/deployed.postman_environment.json --env-var "apiKey=$CITY_API_KEY"

   Le premier appel après une pause peut être lent, car la plateforme a arrêté l’application (réduction à zéro instance). Si seule la vérification « répond en moins de 500 ms » échoue, exécutez de nouveau la collection.

6. Lisez les journaux et trouvez une requête par son identifiant : fastapi cloud logs --no-follow --since 10m.

7. Nettoyez. Supprimez l’application dans le tableau de bord FastAPI Cloud et exécutez fastapi cloud unlink. Supprimez la clé de Postman et vérifiez que /health ne répond plus.

Si un tableau de bord dans un navigateur doit appeler l’API directement, définissez ALLOWED_ORIGINS avec son adresse. Suivez Accès depuis le navigateur et CORS. Mais chaque utilisateur de cette page peut y voir une clé, et CORS n’est pas une authentification. Pour une vraie équipe, le serveur du tableau de bord appelle l’API et conserve la clé.

Utilisation responsable et limites

Une API rend les données faciles à utiliser, mais aussi faciles à mal interpréter. Le README et /v1/dataset doivent transmettre ces limites à chaque tableau de bord :

• Une demande n’est pas un problème. Plusieurs personnes signalent le même nid-de-poule, et la ville ferme beaucoup de ces demandes comme doublons, souvent simultanément. 4 890 demandes ont la même heure de création et de fermeture. Comptez les demandes, et appelez-les des demandes.
• Qui signale. Le 311 montre où les gens signalent des problèmes, pas seulement où les rues sont en mauvais état. Un quartier où moins de gens utilisent le 311 peut sembler en meilleur état qu’il ne l’est. Ne classez pas les quartiers selon le nombre de demandes comme si ce nombre mesurait la qualité des routes.
• Fermée ne veut pas dire réparée. Une demande peut être fermée parce qu’un inspecteur n’a trouvé aucun problème, ou parce qu’un autre organisme est responsable. Le nettoyage a supprimé le texte qui précise la raison. « Fermée en moins de 14 jours » mesure le traitement des demandes, pas les réparations.
• Un instantané. Les statuts datent du 7 octobre 2026, et les données se terminent le 30 juin 2025. Un tableau de bord doit afficher la date de l’instantané, fournie par /v1/dataset.
• Des heures sans fuseau horaire. Toutes les heures sont locales à New York. Un client qui les traite comme UTC se trompe de quatre ou cinq heures.
• Emplacement. Les coordonnées indiquent une rue ou une intersection, pas une personne. Les données ne contiennent ni nom ni numéro de téléphone des personnes ayant fait un signalement. Ne les joignez tout de même pas à d’autres données pour découvrir qui a signalé quoi.
• Ce que cette API n’a pas. Elle a une seule clé pour toute l’équipe et aucune limite de débit. Un client peut donc envoyer autant de requêtes qu’il le souhaite. Des clés par client et des limites de débit sont les prochaines étapes pour un vrai service. La connexion des utilisateurs est enseignée dans un cours ultérieur.

Faites-en un projet pour votre portfolio

Ce projet montre des compétences recherchées par les employeurs. Il a une base de données derrière une API, du SQL sûr et une pagination qui reste correcte quand les données changent. Il a aussi des entrées validées, des tests avec un jeu de données de test, un contrat révisé et un plan de déploiement.

Quoi mettre dans le dépôt :

city-requests-api/
  README.md               what it is, the endpoints, the error codes, how to run and check it
  DATASET.md              the data's source, version, licence, cleaning rules and limits
  LICENSE                 MIT, for the code
  pyproject.toml          the pinned versions (requirements-lock.txt for a local set-up)
  city_api/               the API: db.py, queries.py, pagination.py, routes.py, models.py
  scripts/                get_db.py, build_db.py, and the two OFFSET scripts
  tests/                  the 67 tests, with fixture_rows.csv
  contract/openapi.json   the reviewed contract
  postman/                the collection, and its two environments without keys
  data/                   empty in Git: get_db.py downloads the database

Pour le publier :

1. Créez un nouveau dépôt Git à partir du dossier de votre projet et poussez-le vers GitHub. .gitignore exclut l’environnement virtuel et la base de données. Tout le monde peut récupérer la base de données avec python scripts/get_db.py.
2. Conservez LICENSE (MIT, pour le code) et DATASET.md (source, licence et modifications des données).
3. Assurez-vous que python -m pytest réussit à partir d’un nouveau clone, et indiquez-le dans le README.
4. Écrivez le README pour quelqu’un qui ne connaît pas le projet. Un modèle de README :

README.md (outline)

# city-requests-api

## What it is
A read-only API over 109,915 New York 311 street-repair requests (Jan to Jun 2025),
for dashboards: filtered lists in pages, single requests, complaint types, daily numbers.

## Design choices
- Read-only SQLite (mode=ro, immutable=1); new data means a new deployment.
- Parameterised SQL only; every query in city_api/queries.py.
- Keyset pagination on (created_at, id) with an opaque cursor tied to its filters.
- Strict query parameters: fixed lists, no unknown names, no time zones.

## Run it / Check it
(set-up, pytest, the Postman CLI)

## Data and limits
(source, snapshot date, cleaning rules, what the numbers do not mean)

Comment le décrire en entrevue, en deux phrases : « J’ai créé un service FastAPI en lecture seule pour les tableaux de bord d’analystes, sur 110 000 vraies demandes de réparation de rues de New York dans SQLite, avec du SQL paramétré, des filtres validés et une pagination par clés derrière un curseur opaque. Un test parcourt les données avec sept tailles de page et prouve qu’il n’omet ni ne répète jamais une ligne, et un contrat révisé et une collection Postman maintiennent les promesses faites aux tableaux de bord. »

Allez plus loin : chacune de ces options est une bonne prochaine étape, avec un changement principal.

• Les plus récentes d’abord. Ajoutez order=desc. La condition de pagination par clés devient <, et le curseur doit retenir l’ordre.
• Une actualisation quotidienne. Construisez le fichier de demain avec build_db.py, vérifiez-le avec les tests et déployez-le. Notez le comportement d’un tableau de bord qui parcourt les pages pendant le changement.
• Statistiques par arrondissement. GET /v1/stats/by-borough pour une période, avec les mêmes définitions que les nombres quotidiens.
• Mise en cache. Les données ne changent pas entre les déploiements. Les réponses peuvent donc avoir un en-tête ETag ou Cache-Control. Mesurez les économies obtenues.

Vérification des connaissances

Vérifiez votre compréhension

Un collègue dit : « Notre fichier de base de données ne change jamais pendant que l'API fonctionne. Les pages avec OFFSET conviennent donc. » Quand est-ce vrai, et qu'est-ce qui change cela?

Afficher la réponse

C’est vrai pour cette configuration précise. Avec un seul fichier fixe, OFFSET retourne les mêmes lignes pour la même page, sans omission ni répétition. Seul le coût des pages éloignées demeure. Pour 110 000 lignes, il est d’environ 1,5 ms. Ce n’est plus vrai dès que les données peuvent changer entre deux requêtes d’un même client. Par exemple, une actualisation quotidienne pendant la pagination, ou une base de données où d’autres programmes écrivent. Une ligne qui quitte ou rejoint la liste avant la page actuelle déplace alors toutes les lignes suivantes. OFFSET omet ou répète donc une ligne. La pagination par clés sur (created_at, id) ne dépend pas des positions et reste donc correcte. Son coût est aussi constant, ce qui en fait le choix par défaut le plus sûr.

Vérifiez votre compréhension

Pourquoi la pagination par clés utilise-t-elle (created_at, id) plutôt que created_at seul? Utilisez un nombre de ces données dans votre réponse.

Afficher la réponse

created_at n’est pas unique : 34 demandes ont exactement la valeur 2025-03-03T07:00:00. De plus, 11 489 demandes partagent leur heure avec une autre demande. Si une page finit au milieu d’un tel groupe, created_at > last omet le reste du groupe. À l’inverse, created_at >= last répète la partie déjà envoyée. L’identifiant id unique rend la paire unique. Ainsi, « après la dernière ligne » désigne une position précise dans l’ordre. Le test test_a_page_can_end_inside_a_group_of_equal_times vérifie ce cas.

Vérifiez votre compréhension

Le point de terminaison de liste construit sa clause WHERE avec une chaîne f. Pourquoi n'est-ce pas une injection SQL? Quel changement au code en ferait une injection?

Afficher la réponse

La chaîne f joint uniquement des textes provenant de notre code : les conditions dans CONDITIONS, et COLUMNS. Les valeurs de la requête sont transmises à SQLite comme paramètres, représentés par les signes ?. SQLite ne les interprète donc jamais comme du SQL. Il y aurait injection si une partie du texte SQL provenait de la requête. Par exemple, une valeur insérée dans le texte de la condition plutôt que dans les paramètres. Autre exemple : un nom de colonne choisi par le client, comme ?sort=created_at, inséré dans ORDER BY. Il faut vérifier toute option de tri à partir d’une liste fixe dans le code. Seule la constante choisie entre dans le SQL.

Récapitulatif

• Une connexion en lecture seule (mode=ro) rend toute écriture impossible, et pas seulement improbable. immutable=1 supprime le verrouillage, et de nouvelles données exigent un redémarrage.
• Les valeurs d’une requête atteignent SQL uniquement comme paramètres ?. Les noms de colonnes et les mots SQL proviennent uniquement du code.
• Un modèle Pydantic pour la chaîne de requête utilise des listes fixes et extra="forbid". Il transforme les réponses erronées silencieuses en erreurs 422 claires.
• Les pages avec OFFSET omettent ou répètent des lignes quand les données changent, et ralentissent quand on avance. Une pagination par clés sur une paire unique, comme (created_at, id), avec un curseur opaque, évite ces deux problèmes.
• Les statistiques exigent des définitions précises : ce qui est compté, la médiane des durées, et une limite sur la plage. Il faut aussi utiliser null plutôt que 0 en l’absence de données.
• Un petit jeu de données de test composé de lignes réelles permet de vérifier les nombres à la main. Les tests peuvent aussi parcourir toutes les pages en une seconde.

Vous avez créé une API de données à laquelle une équipe peut faire confiance. Le travail final vous demande de modifier l’API de billets du cours comme une équipe le ferait, du contrat au déploiement.

Leçon suivante : Travail final