Anatomie d'un module¶

Un module métier MIC se décompose en 5 couches — controller/, interface/http/v1/, model/business/, model/persistence/ et model/repositories/ — chacune avec son test miroir. C'est le découpage canonique décrit dans la structure canonique du service ; cette page en détaille l'intérieur, fichier par fichier, sur l'exemple d'un module mail.

Pas de template « flat »

Le module n'est pas un trio model.py / domain.py / routes.py posé à plat dans src/<service>/. Ces fichiers à plat sont des anti-patterns dépréciés. Le canon = package app/ + module décomposé en 5 couches, imports par chemin direct (pas de ré-export depuis les __init__).

Structure type¶

app/modules/<domain>/
├── __init__.py                                # docstring (pas de ré-export)
├── controller/                                # use cases (1 classe par action)
│   ├── <action_a>.py                          # ex. EnqueueMailController
│   └── <action_b>.py                          # ex. ReadMailController
├── interface/                                 # adaptateurs entrants (HTTP, gRPC…)
│   └── http/v1/                               # versionné /api/v1
│       ├── <action_a>.py                      # payload/response + handler
│       ├── <action_b>.py
│       └── routes.py                          # build_<module>_v1_routes(...)
└── model/
    ├── business/<action>.py                   # résultat métier (StrictSchema)
    ├── persistence/<entity>_record.py         # SQLModel(s) table=True
    └── repositories/<entity>_repository.py    # accès persistance

Une couche = une responsabilité. Le controller/ ne sait rien du HTTP ni de la Session ; l'interface/http/v1/ ne fait pas de métier ; le model/persistence/ ne contient pas de logique ; le model/repositories/ est le seul à toucher la base. On peut tester chaque couche isolément.

`controller/` — les cas d'usage¶

Une classe par action, dans son propre fichier. Le controller reçoit son repository injecté, porte la validation métier, puis délègue la persistance ; aucune dépendance HTTP, aucun framework, jamais de Session. Les erreurs métier remontent en DomainError avec un code stable.

# app/modules/mail/controller/enqueue_mail.py
from __future__ import annotations

from mic.model.errors import DomainError

from app.modules.mail.model.business.enqueue_mail import EnqueueMailResult
from app.modules.mail.model.repositories.mail_repository import MailRepository


class EnqueueMailController:
    """Cas d'usage : valide puis met un mail en file (statut ``pending``)."""

    def __init__(self, *, repository: MailRepository) -> None:
        self._repository = repository

    def execute(
        self,
        *,
        recipient: str,
        subject: str,
        body: str,
        max_attempts: int = 3,
    ) -> EnqueueMailResult:
        if not recipient.strip() or "@" not in recipient or _has_crlf(recipient):
            raise DomainError(
                code="mail.invalid_recipient",
                message="recipient must be a non-empty email address without CRLF",
                details={"received_recipient": recipient},
            )
        if not subject.strip() or _has_crlf(subject):
            raise DomainError(
                code="mail.invalid_subject",
                message="subject must not be empty nor contain CRLF",
            )
        return self._repository.enqueue_mail(
            recipient=recipient,
            subject=subject,
            body=body,
            max_attempts=max_attempts,
        )


def _has_crlf(value: str) -> bool:
    return "\r" in value or "\n" in value

Une action de lecture est tout aussi simple — elle lève mail.not_found plutôt que de retourner None :

# app/modules/mail/controller/read_mail.py
from __future__ import annotations

from mic.model.errors import DomainError

from app.modules.mail.model.business.read_mail import ReadMailResult
from app.modules.mail.model.repositories.mail_repository import MailRepository


class ReadMailController:
    """Cas d'usage : lit un mail par id."""

    def __init__(self, *, repository: MailRepository) -> None:
        self._repository = repository

    def execute(self, *, job_id: str) -> ReadMailResult:
        job = self._repository.find_mail(job_id=job_id)
        if job is None:
            raise DomainError(
                code="mail.not_found",
                message=f"MailJob id={job_id} not found",
                details={"job_id": job_id},
            )
        return job

Règles :

Pas d'import HTTP (fastapi, litestar, httpx).
Pas de Session ni d'accès DB direct — c'est le rôle du repository.
Erreurs métier en DomainError avec code stable et lisible-machine.

`model/business/` — les résultats de cas d'usage¶

Chaque cas d'usage expose un résultat métier (StrictSchema), découplé de l'enregistrement de persistance et de la réponse HTTP. C'est le contrat partagé entre le repository (qui le produit) et le controller (qui le renvoie).

# app/modules/mail/model/business/enqueue_mail.py
from __future__ import annotations

from mic.contracts import StrictSchema


class EnqueueMailResult(StrictSchema):
    id: str
    status: str

# app/modules/mail/model/business/read_mail.py
from __future__ import annotations

from datetime import datetime

from mic.contracts import StrictSchema


class ReadMailResult(StrictSchema):
    id: str
    recipient: str
    subject: str
    status: str
    attempts: int
    created_at: datetime

`model/persistence/` — l'enregistrement SQLModel¶

<entity>_record.py : SQLModel table=True, suffixe Record, identité sous forme de chaîne UUID générée côté app. Pas de logique.

# app/modules/mail/model/persistence/mail_job_record.py
from __future__ import annotations

import uuid
from datetime import UTC, datetime
from enum import StrEnum

from sqlmodel import Field, SQLModel


class MailStatus(StrEnum):
    PENDING = "pending"
    IN_FLIGHT = "in_flight"
    SENT = "sent"
    FAILED = "failed"
    DEAD_LETTER = "dead_letter"


def _utcnow() -> datetime:
    return datetime.now(tz=UTC)


class MailJobRecord(SQLModel, table=True):
    __tablename__ = "mail_jobs"

    id: str = Field(default_factory=lambda: str(uuid.uuid4()), primary_key=True)
    recipient: str = Field(index=True)
    subject: str
    body: str
    status: MailStatus = Field(default=MailStatus.PENDING, index=True)
    attempts: int = Field(default=0)
    max_attempts: int = Field(default=3)
    last_error: str | None = Field(default=None)
    created_at: datetime = Field(default_factory=_utcnow)
    updated_at: datetime = Field(default_factory=_utcnow)

SQLModel + table=True

Pydantic v2 ne valide pas à l'instanciation quand table=True. Mettez vos validations métier dans le controller/, et les bornes de format dans interface/http/v1/<action>.py — jamais dans le modèle.

`model/repositories/` — l'accès persistance¶

<entity>_repository.py : seul point d'accès à la base. Il encapsule la Session (via un session_factory injecté) et renvoie des schémas model/business — jamais des records nus.

# app/modules/mail/model/repositories/mail_repository.py
from __future__ import annotations

from collections.abc import Callable
from contextlib import AbstractContextManager

from sqlmodel import Session

from app.modules.mail.model.business.enqueue_mail import EnqueueMailResult
from app.modules.mail.model.business.read_mail import ReadMailResult
from app.modules.mail.model.persistence.mail_job_record import (
    MailJobRecord,
    MailStatus,
)


class MailRepository:
    def __init__(
        self,
        *,
        session_factory: Callable[[], AbstractContextManager[Session]],
    ) -> None:
        self._session_factory = session_factory

    def enqueue_mail(
        self,
        *,
        recipient: str,
        subject: str,
        body: str,
        max_attempts: int = 3,
    ) -> EnqueueMailResult:
        with self._session_factory() as session:
            record = MailJobRecord(
                recipient=recipient,
                subject=subject,
                body=body,
                max_attempts=max_attempts,
                status=MailStatus.PENDING,
            )
            session.add(record)
            session.commit()
            session.refresh(record)
            return EnqueueMailResult(id=record.id, status=record.status.value)

    def find_mail(self, *, job_id: str) -> ReadMailResult | None:
        with self._session_factory() as session:
            record = session.get(MailJobRecord, job_id)
            if record is None:
                return None
            return ReadMailResult(
                id=record.id,
                recipient=record.recipient,
                subject=record.subject,
                status=record.status.value,
                attempts=record.attempts,
                created_at=record.created_at,
            )

`interface/http/v1/` — traduction HTTP ↔ controller¶

Versionnée sous /api/v1. Un fichier par action (payload/response StrictSchema + handle_<action>), plus un routes.py d'agrégation.

`<action>.py` — payload, response, handler¶

La validation de format vit ici (StrictSchema refuse les champs inconnus) ; le handler traduit payload → use case → response.

# app/modules/mail/interface/http/v1/enqueue_mail.py
from __future__ import annotations

from pydantic import EmailStr, Field

from mic.contracts import StrictSchema

from app.modules.mail.controller.enqueue_mail import EnqueueMailController


class EnqueueMailPayload(StrictSchema):
    """Corps attendu pour POST /api/v1/messages."""

    recipient: EmailStr
    subject: str = Field(min_length=1, max_length=998)
    body: str = Field(max_length=1_000_000)
    max_attempts: int = Field(default=3, ge=1, le=10)


class EnqueueMailResponse(StrictSchema):
    id: str
    status: str


def handle_enqueue_mail(
    payload: EnqueueMailPayload,
    *,
    controller: EnqueueMailController,
) -> EnqueueMailResponse:
    result = controller.execute(
        recipient=payload.recipient,
        subject=payload.subject,
        body=payload.body,
        max_attempts=payload.max_attempts,
    )
    return EnqueueMailResponse(id=result.id, status=result.status)

`routes.py` — `build_<module>_v1_routes(...)`¶

Reçoit les controllers déjà câblés et renvoie un tuple[HttpRouteSpec, ...] sous /api/v1. Les HttpRouteSpec sont agnostiques du framework — c'est build_fastapi_router / build_litestar_router qui les montent (cf. adaptateurs HTTP).

# app/modules/mail/interface/http/v1/routes.py
from __future__ import annotations

from mic.contracts import wrap_http_success
from mic.http import HttpResponseSpec, HttpRouteSpec

from app.modules.mail.controller.enqueue_mail import EnqueueMailController
from app.modules.mail.controller.read_mail import ReadMailController
from app.modules.mail.interface.http.v1.enqueue_mail import (
    EnqueueMailPayload,
    handle_enqueue_mail,
)
from app.modules.mail.interface.http.v1.read_mail import handle_read_mail


def build_mail_v1_routes(
    *,
    enqueue_mail_controller: EnqueueMailController,
    read_mail_controller: ReadMailController,
) -> tuple[HttpRouteSpec, ...]:

    def _enqueue(*, payload: EnqueueMailPayload) -> HttpResponseSpec:
        response = handle_enqueue_mail(payload, controller=enqueue_mail_controller)
        return HttpResponseSpec(status_code=201, body=wrap_http_success(response))

    def _read(*, message_id: str) -> HttpResponseSpec:
        response = handle_read_mail(message_id, controller=read_mail_controller)
        return HttpResponseSpec(status_code=200, body=wrap_http_success(response))

    return (
        HttpRouteSpec(
            method="POST", path="/api/v1/messages",
            name="mail_v1_enqueue_mail", category="mail", handler=_enqueue,
        ),
        HttpRouteSpec(
            method="GET", path="/api/v1/messages/{message_id}",
            name="mail_v1_read_mail", category="mail", handler=_read,
        ),
    )

Règles :

Handlers courts, pas de logique métier.
Validations de format via StrictSchema (les bornes vivent dans <action>.py).
Réponses enveloppées via wrap_http_success(...).
Mêmes controllers réutilisables derrière interface/grpc/ ou un CLI.

`init.py` — pas de ré-export¶

Les __init__.py du module portent une docstring et rien d'autre : on importe toujours par chemin direct (app.modules.mail.controller.enqueue_mail), comme dans rdlink-api. Le wiring assemble repository → controllers → routes.

# app/modules/mail/__init__.py
"""Module ``mail`` — file d'envoi, décomposé en 5 couches MIC."""

Tests miroir¶

L'arborescence tests/ reflète app/ — un test par couche :

tests/modules/<domain>/
├── __init__.py
├── test_model.py      # record : defaults, status, timestamps, id UUID
├── test_domain.py     # controller + repository : happy path + DomainError
└── test_routes.py     # via TestClient : codes HTTP + enveloppe (/api/v1/…)

test_domain.py construit un MailRepository sur la SqlDataStorage de test puis exerce les controllers — la chaîne controller → repository → persistence sans TestClient ni framework.

Voir tests miroir · structure canonique.

Anatomie d'un module¶

Structure type¶

controller/ — les cas d'usage¶

model/business/ — les résultats de cas d'usage¶

model/persistence/ — l'enregistrement SQLModel¶

model/repositories/ — l'accès persistance¶

interface/http/v1/ — traduction HTTP ↔ controller¶

<action>.py — payload, response, handler¶

routes.py — build_<module>_v1_routes(...)¶

__init__.py — pas de ré-export¶