Assembler les modules et leurs contrats

Quand les modules d’un outil sont développés séparément, leurs contrats peuvent diverger : le service attend un modèle que le stockage ne fournit pas, ou la collecte renvoie des résultats dans un format différent de celui qu’utilise le calcul. Vous allez assembler les composants autour de formats et d’interfaces explicites. Le but est de pouvoir remplacer le stockage ou la source des taux sans réécrire la logique métier.

Vous avez déjà défini les annotations, les modèles et les types interchangeables dans les leçons précédentes. Ici, vous les utilisez comme contrats entre modules : une entrée est validée en modèle, le service applique les règles métier, et une interface de stockage conserve les dépenses. La collecte des taux reste séparée, car une consultation distante est une opération d’entrée-sortie asynchrone.

Définir le modèle partagé

Le module modeles.py est le point de référence pour la forme d’une dépense. Le format d’entrée sert à recevoir des données, tandis que la dataclass représente une dépense validée utilisée dans le service. La conversion vérifie notamment que la description n’est pas vide et que le montant est strictement positif : les autres modules peuvent alors compter sur ces invariants.

from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from typing import TypedDict


class Categorie(str, Enum):
    ALIMENTATION = "alimentation"
    TRANSPORT = "transport"
    AUTRE = "autre"


class DepenseEntree(TypedDict):
    description: str
    montant: Decimal
    categorie: Categorie


@dataclass(frozen=True)
class Depense:
    description: str
    montant: Decimal
    categorie: Categorie


def creer_depense(entree: DepenseEntree) -> Depense:
    description = entree["description"].strip()
    montant = entree["montant"]
    if not description:
        raise ValueError("La description ne peut pas être vide.")
    if montant <= Decimal("0"):
        raise ValueError("Le montant doit être strictement positif.")
    return Depense(description, montant, entree["categorie"])

Relier le service au stockage

Le service dépend d’un protocole de stockage, pas d’une implémentation précise. Toute classe qui fournit les deux méthodes décrites peut satisfaire ce contrat. Dans cet exemple, StockageMemoire fournit une implémentation exécutable pour la démonstration ; en production, vous pourrez fournir un adaptateur vers un stockage réel sans modifier ajouter_depense ni totaliser.

from decimal import Decimal
from typing import Protocol

from depenses.modeles import Depense, DepenseEntree, creer_depense


class StockageDepenses(Protocol):
    def enregistrer(self, depense: Depense) -> None:
        ...

    def lister(self) -> list[Depense]:
        ...


class StockageMemoire:
    def __init__(self) -> None:
        self._depenses: list[Depense] = []

    def enregistrer(self, depense: Depense) -> None:
        self._depenses.append(depense)

    def lister(self) -> list[Depense]:
        return list(self._depenses)


def ajouter_depense(
    entree: DepenseEntree, stockage: StockageDepenses
) -> Depense:
    depense = creer_depense(entree)
    stockage.enregistrer(depense)
    return depense


def totaliser(depenses: list[Depense]) -> Decimal:
    return sum((depense.montant for depense in depenses), Decimal("0"))

Conserver un contrat asynchrone pour les taux

La collecte prend une liste de devises et une fonction de consultation asynchrone. Elle renvoie un dictionnaire qui associe chaque devise à son taux, au lieu de laisser les composants suivants dépendre de l’ordre de réponses. Le type du rappel précise à la fois son entrée et le fait qu’il produit une coroutine. La fonction ci-dessous reste un adaptateur de démonstration ; remplacez-la par le client du service distant dans le point de composition.

import asyncio
from collections.abc import Awaitable, Callable
from decimal import Decimal

ConsultationTaux = Callable[[str], Awaitable[Decimal]]


async def recuperer_taux(
    devises: list[str], consulter: ConsultationTaux
) -> dict[str, Decimal]:
    taux = await asyncio.gather(*(consulter(devise) for devise in devises))
    return dict(zip(devises, taux, strict=True))


async def taux_de_demo(devise: str) -> Decimal:
    valeurs = {"EUR": Decimal("1"), "USD": Decimal("1.08")}
    await asyncio.sleep(0)
    return valeurs[devise]

Composer les dépendances au point d’entrée

Le point d’entrée rassemble les choix concrets : ici, le stockage en mémoire et la consultation de démonstration. Il construit une entrée typée, appelle le service, puis lance la collecte avec asyncio.run. Les modules gardent ainsi des responsabilités distinctes et les choix d’implémentation ne se répandent pas dans le code métier.

import asyncio
from decimal import Decimal

from depenses.collecte import recuperer_taux, taux_de_demo
from depenses.modeles import Categorie, DepenseEntree
from depenses.service import StockageMemoire, ajouter_depense, totaliser


async def executer() -> None:
    stockage = StockageMemoire()
    entree: DepenseEntree = {
        "description": "Train professionnel",
        "montant": Decimal("42.50"),
        "categorie": Categorie.TRANSPORT,
    }
    ajouter_depense(entree, stockage)
    depenses = stockage.lister()
    taux = await recuperer_taux(["EUR", "USD"], taux_de_demo)
    print(f"Dépenses enregistrées : {len(depenses)}")
    print(f"Total : {totaliser(depenses):.2f} EUR")
    print(f"Taux USD : {taux['USD']}")


def main() -> None:
    asyncio.run(executer())


if __name__ == "__main__":
    main()

Dépenses enregistrées : 1
Total : 42.50 EUR
Taux USD : 1.08

Ajoutez un point d’entrée de paquet minimal, puis lancez le programme depuis la racine du projet avec python -m depenses.cli après avoir installé le paquet ou configuré le chemin d’import. Le marqueur py.typed n’est pas ajouté ici : il ne figure pas dans le contrat de fichiers. Les imports du code restent donc fondés sur le paquet depenses et les chemins définis dans le projet.

"""Outil de traitement des dépenses."""

La composition ci-dessus sert de vérification d’assemblage, pas encore de validation complète du parcours. Elle montre que les signatures s’emboîtent : l’entrée acceptée par le service produit un modèle stockable, et la collecte reçoit une fonction qui respecte son contrat asynchrone. Exécutez ensuite mypy sur le paquet et les tests existants ; corrigez toute divergence au niveau de l’interface concernée plutôt qu’en ajoutant des conversions implicites.

À retenir

• Le modèle partagé définit la forme d’une dépense validée ; le service ne manipule pas directement les données d’entrée non vérifiées.
• Un protocole décrit les opérations attendues du stockage et permet de remplacer son implémentation.
• La collecte asynchrone renvoie des résultats associés explicitement aux devises.
• Le point d’entrée assemble les implémentations concrètes ; les modules métier restent découplés de ces choix.
• Vérifiez l’assemblage avec mypy et les tests ; la validation du parcours complet intervient dans la prochaine leçon.

Dans la prochaine leçon, vous validerez le parcours complet à l’aide de tests d’intégration, y compris l’exécution des tests asynchrones.