06. Application & ports

Les ports : qui déclare le besoin ?#

Un port est une interface qui exprime un besoin sans dire comment il sera satisfait. La question intéressante est : dans quelle couche le déclarer ?

La réponse tient en une question :

Qui a besoin de ce concept ?

• Le concept « je dois pouvoir retrouver une Task par son identité » est exprimé en termes du domaine : il parle de Task et de TaskId, pas de lignes ni de tables. Ce projet place donc le port dans domain/task/repository.py.
• Le concept « je mets certaines lectures en cache » n'appartient à personne dans le métier. Aucun invariant de Task ou de User n'en dépend. C'est une décision de cas d'usage. Le port vit dans application/shared/cache.py.

C'est le point où beaucoup de cours se trompent en assénant « les ports sont dans le domaine ». Ce serait faux ici : mettre CachePort dans domain/ signifierait que le modèle métier des tâches a une opinion sur le cache. Il n'en a aucune.

Le port de persistance, lui, est bien dans le domaine — et c'est là toute l'inversion de dépendance : le domaine possède l'interface, l'infrastructure lui obéit.

# domain/task/repository.py
class TaskRepository(ABC):
    @abstractmethod
    async def save(self, task: Task) -> None: ...
    @abstractmethod
    async def get(self, task_id: TaskId) -> Task: ...
    @abstractmethod
    async def list_by_owner(self, owner_id: UserId, *, limit=100, offset=0) -> list[Task]: ...
    @abstractmethod
    async def delete(self, task_id: TaskId) -> None: ...
    @abstractmethod
    async def exists(self, task_id: TaskId) -> bool: ...

Ce qu'un port ne doit pas exposer#

Un port fuit dès qu'il laisse transparaître son implémentation. Deux exemples du projet, tous deux commentés dans le code :

• EventPublisherPort n'expose que publish — ni connect, ni close. Le cycle de vie de la connexion AMQP est un détail d'infrastructure, ouvert dans le lifespan de l'API ou au démarrage du worker.
• Il parle en événement typé, pas en dict :

  async def publish(self, event: IntegrationEvent) -> None: ...

  Une signature publish(routing_key: str, payload: dict[str, Any]) serait déjà une fuite : elle oblige le use case à sérialiser lui-même et à connaître une clé de routage, qui est du vocabulaire AMQP. Ici l'événement porte son propre type (ShareTaskNotification.name = "task.shared"), et c'est l'adaptateur qui décide d'en faire une clé de routage — un adaptateur Kafka en ferait un topic.

Un port qui exposerait execute(sql: str) ou get_session() ne serait pas un port : ce serait SQLAlchemy déguisé.

Typé ou opaque ? Le cas du cache#

Une nuance qui surprend : CachePort, lui, parle bien en dict JSON-compatible. Est-ce incohérent avec ce qu'on vient de dire ? Non — les deux ports n'ont pas la même nature de contrat.

Un cache privé peut utiliser un dictionnaire, mais son format reste un contrat. Ici _from_cache indexe des clés et parse une date : un ancien JSON valide mais incompatible provoque une erreur, pas automatiquement un cache miss. Il faut invalider ou versionner les clés lors d’une évolution, ou traiter explicitement un contenu incompatible comme un miss.

Et la pagination ? Et le tri ?#

list_by_owner(owner_id, *, limit=100, offset=0) : la pagination est dans le port. Est-ce une fuite technique ?

Non, mais la frontière mérite d'être nommée. limit/offset protègent d'un chargement non borné — c'est une policy applicative légitime (ch. 02). En revanche, ORDER BY created_at DESC NULLS LAST n'a rien à faire dans une signature de port : c'est une décision de présentation ou d'infrastructure. Si un ordre est métier (« les tâches urgentes d'abord »), nomme-le comme tel (list_urgent_first) plutôt que d'exposer une chaîne SQL.

Le use case : orchestrer#

Un use case reçoit ses ports à la construction et expose execute. Il enchaîne : charger → appeler le domaine → sauvegarder.

# application/task/use_cases/complete_task.py
class CompleteTask:
    def __init__(self, repository: TaskRepository) -> None:
        self._repository = repository

    async def execute(self, task_id: str) -> TaskDTO:
        task = await self._repository.get(TaskId.from_string(task_id))
        task.complete()                       # ← la décision est DANS le domaine
        await self._repository.save(task)
        return TaskDTO.from_entity(task)

La décision (« peut-on compléter ? ») n'est pas ici : elle est dans task.complete(). Le use case coordonne.

Orchestration inter-agrégats#

Quand une action touche plusieurs agrégats, le use case dépend de plusieurs ports :

# application/task/use_cases/create_task.py
class CreateTask:
    def __init__(self, tasks: TaskRepository, users: UserRepository) -> None:
        self._tasks = tasks
        self._users = users

    async def execute(self, command: CreateTaskCommand) -> TaskDTO:
        owner_id = UserId.from_string(command.owner_id)
        if not await self._users.exists(owner_id):
            raise UserNotFound(str(owner_id))         # cohérence référentielle
        task = Task.create(owner_id=owner_id, title=TaskTitle(command.title))
        await self._tasks.save(task)
        return TaskDTO.from_entity(task)

C'est la couche application qui coordonne des règles impliquant plusieurs agrégats — le domaine garde chaque agrégat cohérent isolément.

Combiner ports métier et ports techniques#

GetUser montre le mélange, et le contrat qui va avec :

# application/user/use_cases/get_user.py
class GetUser:
    def __init__(self, repository: UserRepository, cache: CachePort) -> None: ...

    async def execute(self, user_id: str) -> UserDTO:
        identity = UserId.from_string(user_id)   # valider AVANT de toucher au cache
        key = f"user:{identity}"

        cached = await self._cache.get(key)
        if cached is not None:
            return self._from_cache(cached)

        user = await self._repository.get(identity)
        dto = UserDTO.from_entity(user)
        await self._cache.set(key, asdict(dto))
        return dto

Trois détails qui font la différence entre « ça marche » et « c'est correct » :

• L'identité est validée avant le cache : un ID syntaxiquement invalide ne déclenche aucune lecture et ne crée aucune entrée.
• Le contrat d'invalidation est écrit dans le docstring : la clé est user:{id}, et toute mutation d'un User doit la supprimer — c'est ce que fait RenameUser. Un cache dont personne ne documente l'invalidation est un bug qui attend son heure.
• La reconstruction depuis le cache est explicite (_from_cache) : JSON ne connaît pas datetime, il faut re-parser.

Les DTO : découpler l'application de ses frontières#

Un use case ne renvoie pas une entité du domaine à l'API, et ne reçoit pas un schéma Pydantic — le premier couplerait le contrat public au modèle métier, le second ferait entrer le framework web dans l'application. On utilise des DTO : des structures neutres.

# application/task/dto.py
@dataclass(frozen=True, slots=True)
class CreateTaskCommand:        # entrée : une « commande »
    owner_id: str
    title: str
    description: str | None = None

@dataclass(frozen=True, slots=True)
class TaskDTO:                  # sortie
    id: str
    owner_id: str
    title: str
    # …
    @classmethod
    def from_entity(cls, task: Task) -> TaskDTO: ...

Pourquoi ne pas exposer l'entité directement ? Trois raisons concrètes :

1. Couplage de contrat. Renommer un attribut du domaine casserait l'API publique. Le DTO absorbe ce genre de changement.
2. Fuite d'état interne. Une entité expose des méthodes de mutation (complete, rename) qui n'ont aucun sens dans une réponse HTTP.
3. Sérialisation. Le DTO parle en types simples (str, datetime), pas en value objects — ce qui le rend trivial à sérialiser… et à mettre en cache.

Un DTO qui contiendrait owner: User raterait sa cible : il réintroduirait la navigation entre agrégats écartée au chapitre 05.

Le piège inverse : la cérémonie#

Tout ce chapitre justifie des traductions successives. Il faut aussi savoir quand elles deviennent absurdes. Compte les représentations d'un même concept dans ce projet :

CreateUserRequest    (schéma HTTP, presentation)
CreateUserCommand    (DTO d'entrée, application)
User                 (entité, domain)
UserDTO              (DTO de sortie, application)
UserResponse         (schéma HTTP, presentation)
UserModel            (modèle ORM, infrastructure)

Six classes pour transporter un nom et un e-mail. Pour un système durable avec un métier riche, c'est parfaitement raisonnable : chacune a une raison de changer différente, et c'est précisément ce qui permet de faire évoluer l'API sans toucher au domaine. Pour un CRUD administratif de 400 lignes qui ne grossira jamais, c'est absurde — et le dire fait partie du cours.

Trois questions pour trancher, à poser objet par objet et non en bloc :

1. Ces deux représentations ont-elles des raisons de changer différentes ? Si le schéma HTTP et le DTO d'entrée changeront toujours ensemble et pour la même raison, l'un des deux est du bruit.
2. Y a-t-il un deuxième appelant, réel ou probable ? L’API et le CLI de ce projet justifient CreateUserCommand : sans lui, ils dépendraient d'un objet Pydantic conçu pour HTTP.
3. Le domaine a-t-il une forme différente du transport ? User porte des value objects (Email, UserName) ; UserResponse porte des str. Quand les deux formes sont identiques champ pour champ et pour toujours, le mapper ne protège rien.

La tension entre isolation et cérémonie ne se résout pas une fois pour toutes : elle se rejuge à chaque couche ajoutée. Le chapitre 13 propose une progression graduée plutôt qu'un choix tout-ou-rien.

Le flux complet d'une requête#

Le use case est testable sans FastAPI ni base : on lui passe une CreateTaskCommand et des ports en mémoire, on vérifie le TaskDTO renvoyé. C'est exactement ce que fait tests/unit/application/ (ch. 12).

À retenir#

• Un port appartient à la couche qui exprime le besoin : domain/ pour les repositories, application/ pour cache, SMTP, broker.
• Un port ne laisse pas fuiter son implémentation (pas de connect, pas de SQL).
• Le use case orchestre : il ne porte pas les invariants, il peut porter la policy du cas d'usage.
• Une action multi-agrégats → un use case dépendant de plusieurs ports.
• Les DTO isolent l'application des frameworks des deux côtés, et évitent de réintroduire la navigation entre agrégats.