Analyse détaillée du code source de Flask

Connaissances préliminaires

Dans la définition de fonction suivante, les annotations de type indiquent que le paramètre peut être soit un objet ContextVar contenant un dictionnaire, soit None, avec une valeur par défaut de None. Cette syntaxe utilise des types d'union pour flexibilité.

def __init__(self, variable_contexte: ContextVar[dict[str, t.Any]] | None = None) -> None:
    pass

  • regle : chaîne représentant la règle URL, par exemple /bonjour.
  • point_terminal : nom du point terminal de la route, par défaut None.
  • fonction_vue : fonction gérant cette route, par défaut vide.

Werkzeug et compréhension de WSGI (via run_simple)

Flask repose sur Werkzeug pour implémenter WSGI. Lorsqu'une requête arrive, la méthode app.__call__ est invoquée. L'paplication est d'abord instanciée via app.run().

Examen de app.run()

La méthode run configure le serveur et invoque run_simple de Werkzeug. Voici une version modifiée du code :

def demarrer(
    self,
    hote: str | None = None,
    port: int | None = None,
    debug: bool | None = None,
    charger_env: bool = True,
    **options: t.Any,
) -> None:
    nom_serveur = self.config.get("NOM_SERVEUR")
    hote_sn = port_sn = None
    if nom_serveur:
        hote_sn, _, port_sn = nom_serveur.partition(":")
    if not hote:
        hote = hote_sn or "127.0.0.1"
    if port or port == 0:
        port = int(port)
    elif port_sn:
        port = int(port_sn)
    else:
        port = 5000
    from werkzeug.serving import run_simple
    try:
        run_simple(t.cast(str, hote), port, self, **options)
    finally:
        self._premiere_requete = False

En interne, run_simple lance un serveur WSGI qui appelle l'application Flask pour chaque requête. Un exemple simplifié :

from werkzeug.serving import run_simple

def mon_appli(environ, demarrer_reponse):
    demarrer_reponse('200 OK', [('Content-Type', 'text/plain')])
    return [b'Bonjour, monde!']

if __name__ == '__main__':
    run_simple('localhost', 5000, mon_appli, use_reloader=True, use_debugger=True)

La méthode app.__call__

Cette méthode génère une réponse WSGI. Elle délègue à wsgi_app, qui gère le contexte et le routage. Code modifié :

def __call__(self, environ: dict, demarrer_reponse: t.Callable) -> t.Any:
    return self.appli_wsgi(environ, demarrer_reponse)

def appli_wsgi(self, environ: dict, demarrer_reponse: t.Callable) -> t.Any:
    ctx = self.contexte_requete(environ)
    erreur: BaseException | None = None
    try:
        try:
            ctx.pousser()
            reponse = self.dispatch_complet()
        except Exception as e:
            erreur = e
            reponse = self.gerer_exception(e)
        except:
            erreur = sys.exc_info()[1]
            raise
        return reponse(environ, demarrer_reponse)
    finally:
        if "werkzeug.debug.preserve_context" in environ:
            environ["werkzeug.debug.preserve_context"](_cv_appli.get())
            environ["werkzeug.debug.preserve_context"](_cv_requete.get())
        if erreur is not None and self.ignorer_erreur(erreur):
            erreur = None
        ctx.pop(erreur)

Flask encapsule la logique WSGI via cette méthode, en utilisant Werkzeug pour la couche bas-niveau.

Initialisation de l'instance Flask

Lors de Flask(__name__), le constructeur initialise des chemins et confgiurations par défaut pour les fichiers statiques et templates. Exemple de code modifié pour l'initialisation :

class Flask:
    def __init__(
        self,
        nom_import: str,
        chemin_url_statique: str | None = None,
        dossier_statique: str | os.PathLike[str] | None = "statique",
        hote_statique: str | None = None,
        correspondance_hote: bool = False,
        correspondance_sous_domaine: bool = False,
        dossier_templates: str | os.PathLike[str] | None = "templates",
        chemin_instance: str | None = None,
        config_relative_instance: bool = False,
        chemin_racine: str | None = None,
    ):
        self.cli = cli.GroupeCommandes()
        self.cli.name = self.name
        if self.dossier_statique_existe:
            assert bool(hote_statique) == correspondance_hote, "Combinaison invalide"
            reference_self = weakref.ref(self)
            self.ajouter_regle_url(
                f"{self.chemin_url_statique}/<nom_fichier>",
                point_terminal="statique",
                hote=hote_statique,
                fonction_vue=lambda **kw: reference_self().envoyer_fichier_statique(**kw),
            )
</nom_fichier>

Cette étape configure les bases de l'application Flask.

Code source des routes et vues

Liaison routes-fonctions via add_url_rule

Flask utilise plusieurs méthodes pour lier les routes aux fonctions vue, mais toutes convergent vers add_url_rule. Exemple de code modifié :

@app.route('/route', methods=['GET'])
def ma_route():
    return 'route'

@app.get('/get')
def obtenir():
    return 'get'

def fonction_ajoutee():
    return 'ajout'

app.ajouter_regle_url('/ajout', fonction_vue=fonction_ajoutee)

La fonction add_url_rule est implémentée comme suit (version modifiée) :

def ajouter_regle_url(
    self,
    regle: str,
    point_terminal: str | None = None,
    fonction_vue: ft.RouteCallable | None = None,
    fournir_options_auto: bool | None = None,
    **options: t.Any,
) -> None:
    if point_terminal is None:
        point_terminal = _point_terminal_depuis_fonction_vue(fonction_vue)
    options["point_terminal"] = point_terminal
    methodes = options.pop("methodes", None)
    if methodes is None:
        methodes = getattr(fonction_vue, "methodes", None) or ("GET",)
    if isinstance(methodes, str):
        raise TypeError("Les méthodes doivent être une liste de chaînes.")
    methodes = {item.upper() for item in methodes}
    methodes_requises = set(getattr(fonction_vue, "methodes_requises", ()))
    if fournir_options_auto is None:
        fournir_options_auto = getattr(fonction_vue, "fournir_options_auto", None)
    if fournir_options_auto is None:
        if "OPTIONS" not in methodes:
            fournir_options_auto = True
            methodes_requises.add("OPTIONS")
        else:
            fournir_options_auto = False
    methodes |= methodes_requises
    regle_obj = self.classe_regle_url(regle, methodes=methodes, **options)
    regle_obj.fournir_options_auto = fournir_options_auto
    self.carte_url.ajouter(regle_obj)
    if fonction_vue is not None:
        ancienne_fonction = self.fonctions_vue.get(point_terminal)
        if ancienne_fonction is not None and ancienne_fonction != fonction_vue:
            raise AssertionError("Écrasement de fonction vue existante.")
        self.fonctions_vue[point_terminal] = fonction_vue

Le point terminal (endpoint) sert de clé pour mapper les fonctions vue. Il est automatiquement dérivé du nom de la fonction si non spécifié.

Correspondance chemin-Fonction vue

Lorsqu'une requête arrive, Flask extrait la règle URL du contexte de requête pour trouver la fonction vue correspondante. Code modifié pour dispatch_request :

def dispatch_requete(self) -> ft.RetourValeurReponse:
    req = contexte_requete.request
    if req.exception_routage is not None:
        self.lever_exception_routage(req)
    regle: Regle = req.regle_url
    if getattr(regle, "fournir_options_auto", False) and req.methode == "OPTIONS":
        return self.reponse_options_par_defaut()
    args_vue: dict[str, t.Any] = req.args_vue
    return self.assurer_synchrone(self.fonctions_vue[regle.point_terminal])(**args_vue)

La classe Regle est créée par Werkzeug et stockée dans Carte. La correspondance est effectuée via carte.match pendant le push du contexte.

Résumé des relations entre point_terminal, Regle et Carte

Durant le démarrage, add_url_rule initialise une Regle, la compile, et l'ajoute à la Carte via un StateMachineMatcher. À la requête, la Carte identifie la Regle à partir du chemin, puis le point_terminal est utilisé pour récupérer la fonction vue.

Contextes

Flask utilise des contextes pour gérer les données par requête, permettant l'accès global aux objets comme request.

Contexte de requête

Le contexte de requête est poussé et dépilé via push et pop. Code modifié pour la gestion du contexte :

def pousser(self) -> None:
    ctx_appli = _cv_appli.get(None)
    if ctx_appli is None or ctx_appli.appli is not self.appli:
        ctx_appli = self.appli.contexte_appli()
        ctx_appli.pousser()
    else:
        ctx_appli = None
    self._jetons_cv.append((_cv_requete.set(self), ctx_appli))
    if self.session is None:
        interface_session = self.appli.interface_session
        self.session = interface_session.ouvrir_session(self.appli, self.requete)
        if self.session is None:
            self.session = interface_session.creer_session_nulle(self.appli)
    if self.adaptateur_url is not None:
        self.correspondre_requete()

L'objet request global est un LocalProxy qui accède dynamiquement au contexte courant. Exemple :

from flask import request, contexte_requete

@app.get('/')
def index():
    print(request is contexte_requete)  # False
    print(request.headers is contexte_requete.request.headers)  # True
    return 'Bonjour'

LocalProxy utilise des atrtibuts dynamiques pour rediriger vers l'objet sous-jacent, assurant la sécurité des threads.

Étiquettes: Flask Werkzeug Python WSGI routage

Publié le 2 juillet à 01h10