Architecture
Hexagonale

from abc import ABC, abstractmethod

# Port sortant — ce que le domaine requiert
# Le domaine définit l'interface, l'infrastructure l'implémente
class ArticleRepository(ABC):  # ← Port
    @abstractmethod
    def sauvegarder(self, article) -> None: ...
    @abstractmethod
    def trouver_par_id(self, id: int): ...
    @abstractmethod
    def lister_tous(self): ...

# Port sortant — service email
class ServiceEmail(ABC):        # ← Port
    @abstractmethod
    def envoyer(self, dest: str, sujet: str, corps: str): ...

# ═══════════════════════════════════════════════════
# Adapters — implémentations concrètes

class ArticleRepositoryMySQL(ArticleRepository):  # ← Adapter
    def __init__(self, db): self.db = db
    def sauvegarder(self, article):
        self.db.execute('INSERT INTO articles...', [article.titre])
    def trouver_par_id(self, id): ...
    def lister_tous(self): ...

class ArticleRepositoryMemoire(ArticleRepository):  # ← Adapter Tests
    def __init__(self): self._articles = []
    def sauvegarder(self, article): self._articles.append(article)
    def trouver_par_id(self, id): return next((a for a in self._articles if a.id==id), None)
    def lister_tous(self): return list(self._articles)

# Domaine — entité pure
from dataclasses import dataclass

@dataclass
class Article:
    id:      int | None
    titre:   str
    contenu: str

    def valider(self):
        if not self.titre.strip():
            raise ValueError("Titre vide")
        if len(self.contenu) < 10:
            raise ValueError("Contenu trop court")

# Application — use case
# Le service reçoit les ports via injection — il ne les crée pas
class PublierArticleService:
    def __init__(self,
                 repo: ArticleRepository,    # ← Port injecté
                 email: ServiceEmail):        # ← Port injecté
        self.repo  = repo
        self.email = email

    def executer(self, titre: str, contenu: str) -> Article:
        article = Article(id=None, titre=titre, contenu=contenu)
        article.valider()             # règle métier
        self.repo.sauvegarder(article) # port → adapter MySQL ou mémoire
        self.email.envoyer(           # port → adapter SMTP ou mock
            'admin@site.com', 'Nouvel article', titre
        )
        return article

# Infrastructure — adapter HTTP (Flask)
from flask import Blueprint, request, jsonify

def creer_router(service: PublierArticleService):
    bp = Blueprint('articles', __name__)

    @bp.route('/articles', methods=['POST'])
    def creer():
        data = request.get_json()
        try:
            article = service.executer(data['titre'], data['contenu'])
            return jsonify({'id': article.id}), 201
        except ValueError as e:
            return jsonify({'erreur': str(e)}), 400
    return bp

# app.py — composition root (assemblage final)
from flask import Flask

# Choisir les adapters selon l'environnement
import os
if os.environ.get('TEST'):
    repo  = ArticleRepositoryMemoire()
    email = ServiceEmailMock()
else:
    repo  = ArticleRepositoryMySQL(db_connection)
    email = ServiceEmailSMTP(smtp_config)

service = PublierArticleService(repo, email)

app = Flask(__name__)
app.register_blueprint(creer_router(service))

from dataclasses import dataclass, field
from datetime import datetime

# Entité pure — aucun import externe (pas de Flask, SQLAlchemy...)
@dataclass
class Article:
    titre:      str
    contenu:    str
    auteur_id:  int
    id:         int | None = None
    publie:     bool = False
    created_at: datetime = field(default_factory=datetime.now)

    def __post_init__(self):
        if not self.titre.strip():
            raise ValueError("Le titre ne peut pas être vide")
        if len(self.contenu) < 20:
            raise ValueError("Contenu trop court (min 20 caractères)")

    def publier(self):
        if self.publie:
            raise ValueError("Article déjà publié")
        self.publie = True

// Domaine — aucun require() externe !
class Article {
  constructor({ id = null, titre, contenu, auteurId }) {
    this.id       = id;
    this.titre    = titre;
    this.contenu  = contenu;
    this.auteurId = auteurId;
    this.publie   = false;
    this._valider();
  }

  _valider() {
    if (!this.titre?.trim())
      throw new Error('TITRE_VIDE');
    if (this.contenu.length < 20)
      throw new Error('CONTENU_TROP_COURT');
  }

  publier() {
    if (this.publie) throw new Error('DEJA_PUBLIE');
    this.publie = true;
  }
}

module.exports = Article;

const Article = require('../domain/Article'); // ← domain seulement

class PublierArticleService {
  constructor(repo, emailService) {
    this.repo = repo;             // port injecté
    this.emailService = emailService;
  }

  async executer({ titre, contenu, auteurId }) {
    const article = new Article({ titre, contenu, auteurId });
    article.publier();
    await this.repo.sauvegarder(article);    // port
    await this.emailService.notifier(auteurId, article.titre);
    return article;
  }
}

module.exports = PublierArticleService;

class ArticleRepositoryMySQL {
  constructor(db) { this.db = db; }

  async sauvegarder(article) {
    const [r] = await this.db.query(
      'INSERT INTO articles (titre,contenu,auteur_id,publie) VALUES (?,?,?,?)',
      [article.titre, article.contenu, article.auteurId, article.publie]
    );
    article.id = r.insertId;
  }

  async findById(id) { /* ... */ }
  async findAll()   { /* ... */ }
}

// Adapter de test — en mémoire
class ArticleRepositoryMemoire {
  constructor() { this.articles = []; this.nextId = 1; }
  async sauvegarder(article) { article.id = this.nextId++; this.articles.push(article); }
  async findAll() { return this.articles; }
}

module.exports = { ArticleRepositoryMySQL, ArticleRepositoryMemoire };

# tests/test_publier_article.py
import pytest
from domain.entities.article import Article
from application.services.publier_article import PublierArticleService
from infrastructure.adapters.in_memory import (
    ArticleRepositoryMemoire, ServiceEmailMock
)

def test_publier_article_valide():
    repo  = ArticleRepositoryMemoire()
    email = ServiceEmailMock()
    service = PublierArticleService(repo, email)

    article = service.executer(
        titre="Mon article",
        contenu="Contenu suffisamment long pour passer la validation",
        auteur_id=1
    )

    assert article.publie == True
    assert repo.lister_tous()[0].titre == "Mon article"
    assert email.notifications_envoyees == 1

def test_publier_article_titre_vide():
    repo  = ArticleRepositoryMemoire()
    email = ServiceEmailMock()
    service = PublierArticleService(repo, email)

    with pytest.raises(ValueError, match="Titre"):
        service.executer(titre="", contenu="...", auteur_id=1)

    assert len(repo.lister_tous()) == 0  # rien sauvegardé
    assert email.notifications_envoyees == 0  # pas d'email

// tests/PublierArticle.test.js
const PublierArticleService = require('../application/PublierArticleService');
const { ArticleRepositoryMemoire } = require('../infrastructure/adapters/ArticleRepositoryMySQL');

const emailMock = {
  appels: [],
  async notifier(id, titre) { this.appels.push({ id, titre }); },
};

test('publier un article valide', async () => {
  const repo    = new ArticleRepositoryMemoire();
  const service = new PublierArticleService(repo, emailMock);

  const article = await service.executer({
    titre:    'Test',
    contenu:  'Contenu suffisamment long pour être valide',
    auteurId: 1,
  });

  expect(article.publie).toBe(true);
  expect(await repo.findAll()).toHaveLength(1);
  expect(emailMock.appels).toHaveLength(1);
});

test('rejeter un titre vide', async () => {
  const repo    = new ArticleRepositoryMemoire();
  const service = new PublierArticleService(repo, emailMock);

  await expect(service.executer({ titre: '', contenu: '...', auteurId: 1 }))
    .rejects.toThrow('TITRE_VIDE');
});