Skip to content

9. Architecture en couches et accès aux données

Les exemples précédents gardaient leurs données en mémoire. Une vraie application les garde dans une base de données, et s'organise en couches (cf. schéma de l'introduction) :

  • la couche web (contrôleurs, vues) : lire la requête, appeler la couche métier, choisir la vue ;
  • la couche métier : les règles de gestion ;
  • la couche DAO (*Data Access Object*) : les requêtes à la base de données.

Chaque couche ne connaît que la couche du dessous, et le moins possible de ses détails : la couche web ignore le SQL, la couche métier ignore le HTML. Cette séparation rend chaque couche plus simple à comprendre, à tester et à faire évoluer. L'exemple 28 l'applique à la gestion d'une liste de personnes : c'est, en plus petit, la gestion des médecins et des clients de l'étude de cas.

En [Python], l'accès à une base de données repose sur plusieurs étages :

  • la norme DB-API (PEP 249) définit l'interface commune des pilotes de bases de données : ouvrir une connexion, exécuter un ordre SQL, lire les lignes. Pour [MySQL] (et [MariaDB]), nous utilisons [PyMySQL], un pilote écrit entièrement en [Python], qui s'installe sans compilateur ;
  • [SQLAlchemy] est la bibliothèque d'accès aux données la plus utilisée en [Python]. Elle a deux niveaux. Son noyau (*Core*) gère les connexions (dont un pool de connexions, qui garde quelques connexions ouvertes et les prête aux requêtes, car ouvrir une connexion est coûteux) et construit les ordres SQL à partir d'expressions [Python], quel que soit le SGBD. Son ORM (*Object-Relational Mapper*) fait correspondre des classes [Python] (les entités) aux tables de la base, et transforme les lignes lues en objets.

C'est l'équivalent de [JDBC], [JPA] et [Hibernate] en [Java], ou d'[Entity Framework Core] en [C#]. L'extension [Flask-SQLAlchemy] simplifie un peu la configuration de [SQLAlchemy] dans une application [Flask] ; nous utilisons [SQLAlchemy] directement, ce qui montre ce qui se passe, et garde les couches métier et DAO indépendantes de [Flask].

L'exemple nécessite un serveur [MySQL] (ou [MariaDB]) en fonctionnement : l'annexe explique comment en installer un avec [Laragon].

9.1. Exemple 28 : une gestion de personnes en trois couches

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
28-couches/
├── app.py
├── config.py
├── controllers/
│   └── personnes.py
├── dao/
│   └── personnes_dao.py
├── database/
│   └── dbpersonnes.sql
├── entities/
│   └── personne.py
├── metier/
│   ├── donnees.py
│   ├── exception.py
│   └── personnes_metier.py
├── models/
│   └── formulaire_personne.py
├── static/
│   ├── css/
│   │   └── styles.css
│   └── js/
│       └── app.js
└── templates/
    ├── fragments/
    │   ├── champ.html
    │   └── confirmation.html
    ├── layout.html
    └── personnes/
        ├── formulaire.html
        └── liste.html

9.1.1. La base de données

La base [dbpersonnes] est créée par un script SQL. Avec [Laragon], on l'exécute par exemple dans l'outil [HeidiSQL] (menu Fichier / Exécuter un fichier SQL), ou en ligne de commande, depuis le dossier de l'exemple :

1
mysql -u root < database/dbpersonnes.sql

Le fichier [database/dbpersonnes.sql] :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
-- base de l'exemple 28 : une seule table, [personnes]
-- le script est écrit en UTF-8
SET NAMES utf8mb4;
DROP DATABASE IF EXISTS dbpersonnes;
CREATE DATABASE dbpersonnes CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE dbpersonnes;

CREATE TABLE personnes (
  id INT NOT NULL AUTO_INCREMENT,
  -- la version de la ligne, augmentée par SQLAlchemy à chaque modification
  version INT NOT NULL DEFAULT 0,
  prenom VARCHAR(30) NOT NULL,
  nom VARCHAR(30) NOT NULL,
  age INT NOT NULL,
  PRIMARY KEY (id),
  -- deux personnes ne peuvent pas avoir les mêmes prénom et nom
  CONSTRAINT unq_personnes_prenom_nom UNIQUE (prenom, nom)
) ENGINE = InnoDB;

INSERT INTO personnes (prenom, nom, age) VALUES
  ('Marie', 'Pélissier', 42),
  ('Jacques', 'Bromard', 35),
  ('Philippe', 'Jandot', 58);
  • ligne 3 : le script est écrit en [UTF-8] : on l'indique au serveur, pour que les accents des données ([Pélissier]) arrivent intacts ;
  • ligne 5 : le jeu de caractères [utf8mb4] accepte tous les caractères (accents, emojis...). La collation [utf8mb4_unicode_ci] compare les chaînes sans tenir compte des majuscules ni des accents : pour [MySQL], [pelissier] et [Pélissier] sont égaux ;
  • ligne 11 : un numéro de version, augmenté à chaque modification de la ligne (cf. verrou optimiste, plus bas) ;
  • ligne 17 : une contrainte d'unicité nommée : la base refuse deux personnes de mêmes prénom et nom. Son nom servira à reconnaître l'erreur.

9.1.2. La configuration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
# config.py : la configuration de l'application (cf. exemple 25)

# la clé secrète : elle signe la session (messages flash) et les jetons anti-CSRF
SECRET_KEY = "une-cle-secrete-pour-l-exemple-28"

# les messages d'erreur de WTForms sont traduits par WTForms lui-même (cf. exemple 19)
WTF_I18N_ENABLED = False

# la connexion à la base : une URL de SQLAlchemy.
#   mysql+pymysql : le SGBD MySQL, par le pilote PyMySQL
#   root:         : l'utilisateur root, sans mot de passe (ceux de Laragon) ;
#                   avec un mot de passe : root:monMotDePasse@...
#   localhost:3306/dbpersonnes : le serveur, son port, la base
#   charset=utf8mb4 : les échanges avec la base se font en UTF-8
DATABASE_URL = "mysql+pymysql://root:@localhost:3306/dbpersonnes?charset=utf8mb4"

# True : les ordres SQL envoyés à la base sont affichés dans la console
SQL_ECHO = False
  • ligne 15 : [SQLAlchemy] désigne une base par une URL. [mysql+pymysql] : le dialecte SQL ([MySQL]) et le pilote ([PyMySQL]) ; puis l'utilisateur et son mot de passe (ici [root], sans mot de passe, la configuration de [Laragon]), le serveur, son port et la base ; enfin, des paramètres : [charset=utf8mb4] fait dialoguer le pilote et le serveur en [UTF-8]. Si votre serveur a un mot de passe, modifiez cette ligne, ou définissez la variable d'environnement [FLASK_DATABASE_URL] (exemple 25) ;
  • ligne 18 : à [True], [SQLAlchemy] affiche dans la console chaque ordre SQL envoyé à la base. C'est très instructif, et nous l'utiliserons plus bas.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# app.py : l'application de l'exemple 28 ; l'assemblage des trois couches
from pathlib import Path

from flask import Blueprint, Flask
from flask_wtf.csrf import CSRFProtect
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from controllers.personnes import personnes
from dao.personnes_dao import PersonnesDao
from metier.personnes_metier import PersonnesMetier

BOOTSTRAP = Path(__file__).resolve().parent.parent / "commun" / "bootstrap"


def create_app():
    app = Flask(__name__)
    app.jinja_env.trim_blocks = True
    app.jinja_env.lstrip_blocks = True
    app.config.from_pyfile("config.py")
    app.config.from_prefixed_env()

    # l'accès à la base : le "moteur" de SQLAlchemy gère un POOL de connexions
    # (des connexions ouvertes une fois, puis réutilisées d'une requête à l'autre).
    # pool_pre_ping : une connexion est testée avant d'être réutilisée
    moteur = create_engine(app.config["DATABASE_URL"], echo=app.config["SQL_ECHO"],
                           pool_pre_ping=True)
    # la fabrique des sessions ; expire_on_commit=False : les objets lus restent
    # utilisables après la fin de la transaction (par la vue, notamment)
    sessions = sessionmaker(moteur, expire_on_commit=False)

    # l'assemblage des couches : DAO -> métier ; la couche web ne connaît que le métier
    app.extensions["metier"] = PersonnesMetier(sessions, PersonnesDao())

    app.register_blueprint(Blueprint("bootstrap", __name__, static_folder=BOOTSTRAP,
                                     static_url_path="/bootstrap"))
    app.register_blueprint(personnes)
    CSRFProtect(app)
    return app


if __name__ == "__main__":
    create_app().run(port=5000, debug=True)
  • lignes 26-27 : le moteur (*engine*) de [SQLAlchemy] est l'objet qui gère les connexions à une base, et leur pool. Il est créé une fois, au démarrage ; la première connexion n'est ouverte qu'à la première requête. [pool_pre_ping=True] fait tester une connexion du pool avant de la réutiliser : un serveur [MySQL] ferme les connexions restées trop longtemps inactives, et l'application recevrait sinon une erreur ;
  • ligne 30 : une session de l'ORM est l'objet par lequel on lit et on écrit des entités ; elle emprunte une connexion au pool, et suit les entités qu'elle a lues pour détecter leurs modifications. [sessionmaker] crée une fabrique de sessions, dont chaque appel produira une session. Par défaut, à la fin d'une transaction, la session « expire » les entités (elle oublie leurs valeurs, pour les relire à la prochaine utilisation). [expire_on_commit=False] le lui interdit : les entités rendues par la couche métier restent utilisables après la transaction, par la vue qui les affiche ;
  • ligne 33 : l'assemblage des couches : la couche métier reçoit la fabrique de sessions et la couche DAO. La couche web ne connaîtra que la couche métier ;
  • ligne 38 : la protection anti-CSRF de toute l'application (exemple 21).

9.1.3. L'entité Personne

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# entities/personne.py : une ligne de la table [personnes]
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


# la classe mère de toutes les entités : SQLAlchemy y enregistre la
# correspondance classe <-> table de chacune de ses classes filles
class Base(DeclarativeBase):
    pass


class Personne(Base):
    # le nom de la table
    __tablename__ = "personnes"

    # la clé primaire, générée par la base (AUTO_INCREMENT) ; SQLAlchemy relit
    # sa valeur après chaque INSERT. Mapped[int] : le type Python de la colonne
    id: Mapped[int] = mapped_column(primary_key=True)
    # le verrou optimiste : cf. __mapper_args__ plus bas
    version: Mapped[int] = mapped_column()
    prenom: Mapped[str] = mapped_column(String(30))
    nom: Mapped[str] = mapped_column(String(30))
    age: Mapped[int]

    # version_id_col : SQLAlchemy augmente la version à chaque UPDATE, et ajoute
    # la condition "AND version = <valeur lue>"
    __mapper_args__ = {"version_id_col": version}

    def __init__(self, prenom: str, nom: str, age: int):
        self.prenom = prenom
        self.nom = nom
        self.age = age
  • lignes 8-9 : toutes les entités d'une application héritent d'une même classe de base, dérivée de [DeclarativeBase]. [SQLAlchemy] y enregistre la description de chacune (sa table, ses colonnes) : c'est le modèle déclaratif ;
  • ligne 14 : la table associée à la classe ;
  • ligne 18 : une colonne est déclarée par un attribut annoté [Mapped[type]] : le type [Python] de la colonne. [mapped_column] précise ses caractéristiques : ici, la clé primaire. Une clé primaire entière est, par défaut, générée par la base ([AUTO_INCREMENT]) : [SQLAlchemy] relit sa valeur après chaque [INSERT] ;
  • lignes 21-23 : [String(30)] précise la longueur de la colonne (utile si [SQLAlchemy] créait lui-même la table). Une colonne sans particularité peut se contenter de l'annotation ([age]). Le type [Mapped[str]] rend la colonne obligatoire ([NOT NULL]) ; [Mapped[str | None]] la rendrait facultative ;
  • ligne 27 : [version_id_col] désigne la colonne de version : [SQLAlchemy] l'augmente à chaque [UPDATE], et ajoute à celui-ci la condition [AND version = <la valeur lue>]. Si une autre requête a modifié la ligne entre-temps, sa version a changé, l'[UPDATE] ne modifie aucune ligne, et [SQLAlchemy] lève une exception [StaleDataError]. C'est le verrou optimiste : on ne bloque rien, mais on détecte les modifications concurrentes ;
  • lignes 29-32 : un constructeur, pour créer une personne en une ligne.

9.1.4. La couche DAO

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# dao/personnes_dao.py : la couche DAO, des requêtes, aucune règle de gestion.
# Chaque méthode reçoit la SESSION de SQLAlchemy, ouverte par la couche
# métier : toutes les requêtes d'une même opération partagent ainsi la même
# connexion, et la même transaction
from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from entities.personne import Personne


class PersonnesDao:

    # SELECT ... FROM personnes ORDER BY nom, prenom
    def lister(self, session: Session) -> list[Personne]:
        return list(session.scalars(select(Personne).order_by(Personne.nom, Personne.prenom)))

    # SELECT ... WHERE id = ? ; None si la personne n'existe pas
    def lire(self, session: Session, id: int) -> Personne | None:
        return session.get(Personne, id)

    # une personne de mêmes prénom et nom, autre que [sauf_id] :
    # SELECT ... WHERE prenom = ? AND nom = ? AND id <> ? LIMIT 1
    def homonyme(self, session: Session, prenom: str, nom: str, sauf_id: int = 0) -> Personne | None:
        requete = select(Personne).where(Personne.prenom == prenom, Personne.nom == nom,
                                         Personne.id != sauf_id).limit(1)
        return session.scalars(requete).first()

    # INSERT (nouvelle personne) : la personne est ajoutée à la session, et
    # flush envoie TOUT DE SUITE l'ordre SQL (et non au COMMIT)
    def ajouter(self, session: Session, personne: Personne) -> Personne:
        session.add(personne)
        session.flush()
        return personne

    # DELETE FROM personnes WHERE id = ?
    def supprimer(self, session: Session, id: int):
        session.execute(delete(Personne).where(Personne.id == id))

La couche DAO ne contient que des requêtes, sans aucune règle de gestion. Chaque méthode reçoit la session, ouverte par la couche métier : toutes les requêtes d'une même opération métier partagent ainsi la même connexion, et la même transaction.

  • ligne 15 : [select(Personne)] construit une requête [SELECT] sur la table des personnes ; [order_by] la trie. Les colonnes s'écrivent comme des attributs de la classe ([Personne.nom]) : l'éditeur peut vérifier leur nom, et aucun SQL n'est écrit à la main. [session.scalars(...)] exécute la requête et rend les entités construites à partir des lignes lues ;
  • ligne 19 : la lecture par clé primaire, ou [None] ;
  • lignes 23-26 : [where] ajoute des conditions, écrites avec les opérateurs de [Python] ([==], [!=]) : [SQLAlchemy] les traduit en SQL, avec des paramètres (les valeurs ne sont jamais recopiées dans le texte de l'ordre SQL, ce qui rend impossible l'injection SQL). [first()] rend la première entité, ou [None] ;
  • lignes 30-33 : [session.add] confie une nouvelle entité à la session ; [flush] envoie tout de suite l'[INSERT] à la base (sans [flush], il partirait au [COMMIT]). La base génère alors l'id, que [SQLAlchemy] range dans l'entité ;
  • ligne 37 : un [DELETE], construit de la même façon.

9.1.5. La couche métier

1
2
3
4
5
6
7
8
9
# metier/donnees.py : les données d'une personne, telles que la couche web les transmet
from dataclasses import dataclass


@dataclass(frozen=True)
class DonneesPersonne:
    prenom: str
    nom: str
    age: int

Les données d'une personne, telles que la couche web les transmet à la couche métier : un objet simple, sans lien avec [Flask], [WTForms] ou [SQLAlchemy].

1
2
3
4
# metier/exception.py : une règle de gestion refuse l'opération ; le message
# est destiné à l'utilisateur
class MetierException(Exception):
    pass

Une classe d'exception propre à la couche métier : une règle de gestion refuse l'opération, et le message de l'exception est destiné à l'utilisateur.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
# metier/personnes_metier.py : la couche MÉTIER, les règles de gestion, et elles seules.
# Chaque méthode ouvre sa TRANSACTION : validée (COMMIT) à la fin du bloc
# with, annulée (ROLLBACK) si une exception en sort
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.exc import StaleDataError

from dao.personnes_dao import PersonnesDao
from entities.personne import Personne
from metier.donnees import DonneesPersonne
from metier.exception import MetierException

HOMONYME = "Une personne porte déjà ces prénom et nom."
MODIFIEE = "Cette fiche a été modifiée entre-temps : rechargez-la."
INCONNUE = "Cette personne n'existe pas (ou plus)."


class PersonnesMetier:

    # sessions : la "fabrique" de sessions SQLAlchemy (cf. app.py)
    def __init__(self, sessions: sessionmaker, dao: PersonnesDao):
        self._sessions = sessions
        self._dao = dao

    def lister(self) -> list[Personne]:
        # sessions.begin() : une session ET sa transaction ; le bloc with les ferme
        with self._sessions.begin() as session:
            return self._dao.lister(session)

    def lire(self, id: int) -> Personne:
        with self._sessions.begin() as session:
            return self._lire(session, id)

    def ajouter(self, donnees: DonneesPersonne) -> Personne:
        try:
            with self._sessions.begin() as session:
                # règle 1 : pas deux personnes de mêmes prénom et nom
                if self._dao.homonyme(session, donnees.prenom, donnees.nom):
                    raise MetierException(HOMONYME)
                return self._dao.ajouter(session, Personne(donnees.prenom, donnees.nom, donnees.age))
        except IntegrityError as e:
            raise self._traduire(e)

    def modifier(self, id: int, version: int | None, donnees: DonneesPersonne) -> Personne:
        try:
            with self._sessions.begin() as session:
                personne = self._lire(session, id)
                # règle 2 : on ne modifie pas une fiche modifiée entre-temps par un autre
                if personne.version != version:
                    raise MetierException(MODIFIEE)
                if self._dao.homonyme(session, donnees.prenom, donnees.nom, id):
                    raise MetierException(HOMONYME)
                # l'entité lue est "suivie" par la session : il suffit de la
                # modifier ; flush envoie l'UPDATE tout de suite
                personne.prenom = donnees.prenom
                personne.nom = donnees.nom
                personne.age = donnees.age
                session.flush()
                return personne
        except StaleDataError:
            # l'UPDATE "WHERE id = ... AND version = ..." n'a modifié aucune ligne
            raise MetierException(MODIFIEE)
        except IntegrityError as e:
            raise self._traduire(e)

    def supprimer(self, id: int) -> Personne:
        with self._sessions.begin() as session:
            personne = self._lire(session, id)
            self._dao.supprimer(session, id)
            return personne

    def _lire(self, session, id: int) -> Personne:
        personne = self._dao.lire(session, id)
        if personne is None:
            raise MetierException(INCONNUE)
        return personne

    # deux ajouts simultanés : la contrainte d'unicité de la base a refusé le
    # second ; e.orig : l'erreur d'origine, celle du pilote PyMySQL
    @staticmethod
    def _traduire(e: IntegrityError) -> Exception:
        if "unq_personnes_prenom_nom" in str(e.orig):
            return MetierException(HOMONYME)
        return e
  • ligne 27 : chaque méthode ouvre sa session et sa transaction : [sessions.begin()] rend une session dont la transaction est commencée. À la fin du bloc [with], la transaction est validée ([COMMIT]) et la session fermée ; si une exception sort du bloc, la transaction est annulée ([ROLLBACK]) : aucune des modifications faites dans le bloc n'est conservée. La couche métier décide ainsi de l'étendue des transactions : une opération métier, une transaction ;
  • lignes 34-42 : la règle n° 1 : pas deux personnes de mêmes prénom et nom. La couche métier vérifie la règle avant d'écrire, et lève une [MetierException] si elle n'est pas respectée. Mais cette vérification ne suffit pas : deux requêtes simultanées pourraient vérifier toutes les deux qu'il n'y a pas d'homonyme, puis insérer toutes les deux. La contrainte d'unicité de la base refuse alors le second [INSERT], et [SQLAlchemy] lève une [IntegrityError], que la méthode [_traduire] transforme en la même [MetierException]. La base est le dernier rempart ;
  • lignes 44-64 : la règle n° 2 : on ne modifie pas une fiche qui a été modifiée entre-temps par quelqu'un d'autre. Le formulaire de modification transporte la version de la fiche lue (dans un champ caché) ; si elle diffère de la version actuelle, la fiche a changé depuis son affichage. Là encore, la base (par le verrou optimiste) détecte le cas où la modification concurrente a lieu entre la lecture et l'[UPDATE] : c'est l'exception [StaleDataError]. Remarquons que la modification se fait simplement en changeant les attributs de l'entité lue : la session la suit, et [flush] envoie un [UPDATE] des seules colonnes modifiées ;
  • lignes 66-70 : la suppression rend la personne supprimée, pour le message de confirmation ;
  • lignes 72-76 : une personne inexistante est une erreur métier ;
  • lignes 80-84 : une méthode statique ne reçoit ni objet ni classe : c'est une simple fonction, rangée dans la classe. [e.orig] est l'exception d'origine, celle du pilote [PyMySQL], dont le message contient le nom de la contrainte violée. Une autre erreur d'intégrité est rendue telle quelle : ce n'est pas une erreur métier, mais une erreur de programmation.

9.1.6. La couche web

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# models/formulaire_personne.py : les champs du formulaire et leurs contraintes (cf. exemple 19)
from flask_wtf import FlaskForm
from wtforms import HiddenField, IntegerField, StringField
from wtforms.validators import DataRequired, InputRequired, Length, NumberRange, Optional


def nettoyer(valeur):
    return valeur.strip() if isinstance(valeur, str) else valeur


# une version : un entier, ou None si le champ est vide (formulaire d'ajout)
def entier_ou_none(valeur):
    return int(valeur) if valeur not in (None, "") else None


class PersonneForm(FlaskForm):

    class Meta:
        locales = ["fr"]

    prenom = StringField("Prénom", filters=[nettoyer], validators=[
        DataRequired("Le prénom est obligatoire."), Length(max=30, message="30 caractères au plus.")])
    nom = StringField("Nom", filters=[nettoyer], validators=[
        DataRequired("Le nom est obligatoire."), Length(max=30, message="30 caractères au plus.")])
    age = IntegerField("Âge", validators=[
        InputRequired("L'âge est obligatoire."),
        NumberRange(min=0, max=150, message="L'âge doit être compris entre 0 et 150.")])
    # la version de la fiche, transportée par un champ caché (modification seulement)
    version = HiddenField(filters=[entier_ou_none])

Le formulaire [WTForms] (exemple 19). Le champ caché [version] a un filtre qui le convertit en entier, ou en [None] s'il est vide (formulaire d'ajout).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
# controllers/personnes.py : la couche WEB ; elle ne connaît que la couche métier
from flask import Blueprint, current_app, flash, redirect, render_template, url_for

from metier.donnees import DonneesPersonne
from metier.exception import MetierException
from metier.personnes_metier import PersonnesMetier
from models.formulaire_personne import PersonneForm

personnes = Blueprint("personnes", __name__)


def metier() -> PersonnesMetier:
    return current_app.extensions["metier"]


@personnes.get("/")
def accueil():
    return redirect(url_for("personnes.liste"))


@personnes.get("/personnes")
def liste():
    return render_template("personnes/liste.html", personnes=metier().lister())


@personnes.get("/personnes/ajout")
def ajout():
    return formulaire("Ajouter une personne", url_for("personnes.ajouter"), PersonneForm())


@personnes.post("/personnes/ajout")
def ajouter():
    form = PersonneForm()
    if not form.validate():
        return formulaire("Ajouter une personne", url_for("personnes.ajouter"), form)
    try:
        p = metier().ajouter(DonneesPersonne(form.prenom.data, form.nom.data, form.age.data))
    except MetierException as e:
        # une règle métier refuse : on réaffiche le formulaire, le message en bas de page
        return formulaire("Ajouter une personne", url_for("personnes.ajouter"), form, str(e))
    flash(f"{p.prenom} {p.nom} a été ajouté(e).", "succes")
    return redirect(url_for("personnes.liste"))


@personnes.get("/personnes/<int:id>/modification")
def modification(id):
    try:
        p = metier().lire(id)
    except MetierException as e:
        flash(str(e), "erreur")
        return redirect(url_for("personnes.liste"))
    # obj=p : les champs du formulaire sont remplis avec les attributs de même nom de p
    return formulaire("Modifier une personne", url_for("personnes.modifier", id=id),
                      PersonneForm(obj=p))


@personnes.post("/personnes/<int:id>/modification")
def modifier(id):
    action = url_for("personnes.modifier", id=id)
    form = PersonneForm()
    if not form.validate():
        return formulaire("Modifier une personne", action, form)
    try:
        donnees = DonneesPersonne(form.prenom.data, form.nom.data, form.age.data)
        p = metier().modifier(id, form.version.data, donnees)
    except MetierException as e:
        return formulaire("Modifier une personne", action, form, str(e))
    flash(f"{p.prenom} {p.nom} a été modifié(e).", "succes")
    return redirect(url_for("personnes.liste"))


@personnes.post("/personnes/<int:id>/suppression")
def supprimer(id):
    try:
        p = metier().supprimer(id)
        flash(f"{p.prenom} {p.nom} a été supprimé(e).", "succes")
    except MetierException as e:
        # pas de formulaire à réafficher : l'erreur voyage elle aussi en message flash
        flash(str(e), "erreur")
    return redirect(url_for("personnes.liste"))


# la page du formulaire : titre, adresse d'envoi, formulaire, erreur globale
def formulaire(titre: str, action: str, form: PersonneForm, erreur: str | None = None):
    return render_template("personnes/formulaire.html", titre=titre, action=action,
                           form=form, erreur=erreur)
  • lignes 12-13 : la couche métier, seule couche connue du contrôleur ;
  • lignes 32-18 : l'ajout, en trois temps : la validation du formulaire (les erreurs de saisie), l'appel de la couche métier (les règles de gestion), puis le schéma PRG. Une [MetierException] réaffiche le formulaire, avec son message en bas de page. [flash] reçoit une catégorie en second paramètre (["succes"]), que le gabarit utilise pour choisir la présentation du message ;
  • lignes 46-54 : le formulaire de modification, pré-rempli : le paramètre [obj] remplit les champs du formulaire avec les attributs de même nom de l'objet, y compris la version. Une personne inexistante renvoie à la liste, avec un message d'erreur flash ;
  • lignes 58-42 : la modification, sur le même modèle que l'ajout ;
  • lignes 73-51 : la suppression. Il n'y a pas de formulaire à réafficher : l'erreur éventuelle voyage elle aussi dans un message flash, de catégorie ["erreur"].

La couche web ne contient ni SQL ni règle de gestion : elle traduit la requête en appel de la couche métier, et le résultat en page.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
<!doctype html>
<html lang="fr">
<head>
  <meta charset="utf-8">
  <title>Personnes</title>
  <link rel="stylesheet" href="{{ url_for('bootstrap.static', filename='css/bootstrap.min.css') }}">
  <link rel="stylesheet" href="{{ url_for('static', filename='css/styles.css') }}">
</head>
<body>
  <div class="container py-3" style="max-width: 40rem">
    {# les messages de succès (messages flash de catégorie "succes") : en haut #}
    {% for message in get_flashed_messages(category_filter=["succes"]) %}
    <div class="alert alert-success" role="status">{{ message }}</div>
    {% endfor %}
    <main>
      {% block contenu %}{% endblock %}
    </main>
    {# l'erreur : de la requête en cours, ou d'avant la redirection (message flash) #}
    {% for message in get_flashed_messages(category_filter=["erreur"]) + ([erreur] if erreur else []) %}
    <div class="zone-erreur" role="alert">{{ message }}</div>
    {% endfor %}
  </div>
  {% include "fragments/confirmation.html" %}
  <script src="{{ url_for('static', filename='js/app.js') }}"></script>
</body>
</html>
  • lignes 12-14 : [category_filter] ne rend que les messages flash des catégories indiquées : les succès, en haut de page ;
  • lignes 19-21 : les erreurs, en bas de page : celles venues d'avant une redirection (messages flash), suivies de celle de la requête en cours (variable [erreur]). L'opérateur [+] concatène deux listes.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
{% extends "layout.html" %}

{% block contenu %}
  <h1 class="h3">Personnes</h1>
  <table class="table align-middle">
    <thead><tr><th>Prénom</th><th>Nom</th><th>Âge</th><th></th></tr></thead>
    <tbody>
      {% for p in personnes %}
      <tr>
        <td>{{ p.prenom }}</td>
        <td>{{ p.nom }}</td>
        <td>{{ p.age }}</td>
        <td class="text-end">
          <a class="btn btn-sm btn-outline-primary" href="{{ url_for('personnes.modification', id=p.id) }}">Modifier</a>
          <form method="post" class="d-inline" action="{{ url_for('personnes.supprimer', id=p.id) }}"
                data-confirmation="Supprimer {{ p.prenom }} {{ p.nom }} ?">
            <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
            <button class="btn btn-sm btn-outline-danger" type="submit">Supprimer</button>
          </form>
        </td>
      </tr>
      {% endfor %}
    </tbody>
  </table>
  <a class="btn btn-primary" href="{{ url_for('personnes.ajout') }}">Ajouter une personne</a>
{% endblock %}

La liste reprend les liens et les boutons de l'exemple 21, avec sa fenêtre de confirmation (les fichiers [fragments/confirmation.html] et [static/js/app.js] sont ceux de l'exemple 21).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
{% extends "layout.html" %}
{% from "fragments/champ.html" import champ %}

{% block contenu %}
  <h1 class="h3">{{ titre }}</h1>
  <form method="post" action="{{ action }}" novalidate>
    {{ form.csrf_token }}
    {# la version de la fiche lue : elle reviendra avec le formulaire #}
    {{ form.version }}
    {{ champ(form.prenom) }}
    {{ champ(form.nom) }}
    {{ champ(form.age) }}
    <button class="btn btn-primary" type="submit">Enregistrer</button>
    <a class="btn btn-link" href="{{ url_for('personnes.liste') }}">Annuler</a>
  </form>
{% endblock %}
  • ligne 9 : le champ caché qui transporte la version de la fiche lue ;
  • lignes 10-12 : la macro de l'exemple 19.

9.1.7. Exécution

La liste des personnes :

Image

Un formulaire d'ajout mal rempli : les erreurs de saisie sont détectées par la couche web.

Image

Un homonyme : la saisie est correcte, mais la couche métier refuse l'ajout. La comparaison étant faite par la base, elle ne tient compte ni des majuscules ni des accents : [marie pelissier] serait lui aussi un homonyme de [Marie Pélissier].

Image

Un ajout réussi :

Image

Pour observer la règle n° 2, ouvrons la même fiche dans deux onglets. Dans le premier, modifions l'âge et enregistrons. Dans le second, qui affiche encore l'ancienne version, modifions aussi la fiche :

Image

La suppression, avec sa confirmation :

Image

Image

9.1.8. Les ordres SQL

Relançons l'application avec la variable d'environnement [FLASK_SQL_ECHO=true] (sous [Windows] : [$env:FLASK_SQL_ECHO="true"]), et suivons dans la console les ordres SQL d'une modification. L'affichage du formulaire de modification de la personne n° 1 :

1
2
3
4
5
6
INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO sqlalchemy.engine.Engine SELECT personnes.id, personnes.version, personnes.prenom, personnes.nom, personnes.age
FROM personnes
WHERE personnes.id = %(pk_1)s
INFO sqlalchemy.engine.Engine [generated in 0.00017s] {'pk_1': 1}
INFO sqlalchemy.engine.Engine COMMIT

Une transaction, un [SELECT] par clé primaire, dont la valeur est passée en paramètre ([%(pk_1)s], la syntaxe des paramètres de [PyMySQL]). L'enregistrement de la fiche, avec un nouvel âge :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO sqlalchemy.engine.Engine SELECT personnes.id, personnes.version, personnes.prenom, personnes.nom, personnes.age
FROM personnes
WHERE personnes.id = %(pk_1)s
INFO sqlalchemy.engine.Engine [cached since 0.05972s ago] {'pk_1': 1}
INFO sqlalchemy.engine.Engine SELECT personnes.id, personnes.version, personnes.prenom, personnes.nom, personnes.age
FROM personnes
WHERE personnes.prenom = %(prenom_1)s AND personnes.nom = %(nom_1)s AND personnes.id != %(id_1)s
 LIMIT %(param_1)s
INFO sqlalchemy.engine.Engine [generated in 0.00016s] {'prenom_1': 'Marie', 'nom_1': 'Pélissier', 'id_1': 1, 'param_1': 1}
INFO sqlalchemy.engine.Engine UPDATE personnes SET version=%(version)s, age=%(age)s WHERE personnes.id = %(personnes_id)s AND personnes.version = %(personnes_version)s
INFO sqlalchemy.engine.Engine [generated in 0.00015s] {'version': 1, 'age': 43, 'personnes_id': 1, 'personnes_version': 0}
INFO sqlalchemy.engine.Engine COMMIT

On y reconnaît, dans une seule transaction : la lecture de la fiche, la recherche d'un homonyme, puis l'[UPDATE]. Celui-ci ne modifie que la colonne changée ([age]), augmente la version (de 0 à 1), et porte la condition du verrou optimiste ([AND personnes.version = 0]). La mention [cached since] indique que [SQLAlchemy] a réutilisé la traduction en SQL d'une requête déjà construite. Envoyons maintenant le même formulaire une seconde fois, toujours avec la version 0 :

1
2
3
4
5
6
INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO sqlalchemy.engine.Engine SELECT personnes.id, personnes.version, personnes.prenom, personnes.nom, personnes.age
FROM personnes
WHERE personnes.id = %(pk_1)s
INFO sqlalchemy.engine.Engine [cached since 0.09593s ago] {'pk_1': 1}
INFO sqlalchemy.engine.Engine ROLLBACK

La version lue (1) n'est pas celle du formulaire (0) : la couche métier a levé une [MetierException], qui a annulé la transaction ([ROLLBACK]), et la page affiche « Cette fiche a été modifiée entre-temps : rechargez-la. ». De même, l'ajout d'un homonyme s'arrête après la recherche de l'homonyme, sans [INSERT].

https://stahe.github.io
Ce tutoriel, rédigé par l’IA d’Anthropic Claude à la demande de Serge Tahé, est mis à la disposition du public selon les termes de la licence Creative Commons Attribution – Pas d’utilisation commerciale – ShareAlike 3.0 Unported.