Shared Kernel — Couche Domaine

Le Shared Kernel contient le code partagé entre tous les bounded contexts. La couche domaine définit les briques DDD fondamentales.

Module : src.share_kernel.domain

Entités de base

class src.share_kernel.domain.base_entity.Entity

Classe de base pour toutes les entités du domaine.

Chaque entité possède un identifiant UUID unique. L’égalité entre entités est basée sur l’identité (id), pas sur les attributs.

id: uuid.UUID

Identifiant unique de l’entité (UUID4).

Exemple :

from src.share_kernel.domain import Entity
import uuid

class Product(Entity):
    def __init__(self, name: str, price: float):
        super().__init__(id=uuid.uuid4())
        self.name = name
        self.price = price

p1 = Product("Widget", 9.99)
p2 = Product("Widget", 9.99)
assert p1 != p2  # Identités différentes

Agrégats (Aggregate Root)

class src.share_kernel.domain.base_aggregate.AggregateRoot(Entity)

Étend Entity avec la gestion des événements domaine.

Les agrégats collectent les événements via _record_event() et les exposent via pull_domain_events() pour publication par l’infrastructure.

_record_event(event: DomainEvent) → None

Enregistre un événement domaine dans la file interne de l’agrégat.

pull_domain_events() → list[DomainEvent]

Retourne et vide la liste des événements enregistrés. Appelée par l’infrastructure après la persistance.

Exemple :

from src.share_kernel.domain import AggregateRoot, DomainEvent
from dataclasses import dataclass

@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    event_type = "order.placed"
    order_id: str = ""
    total: float = 0.0

class Order(AggregateRoot):
    def place(self):
        self._record_event(OrderPlaced(
            order_id=str(self.id),
            total=self.total,
        ))

Value Objects

class src.share_kernel.domain.base_value_object.ValueObject

Frozen dataclass servant de base pour tous les Value Objects. Les VO sont définis par leurs attributs, sans identité propre.

__post_init__() → None

Hook de validation appelé automatiquement après l’initialisation. Les sous-classes surchargeant cette méthode pour valider leurs invariants.

Value Objects disponibles

class src.share_kernel.domain.value_objects.Address

Adresse postale avec validation du code pays ISO 3166-1.

Parameters:

• street – Rue

• city – Ville

• postal_code – Code postal

• country_code – Code pays ISO 3166-1 alpha-2 (ex: "FR", "CA")

• state_or_region – État ou région (optionnel)

from src.share_kernel.domain.value_objects import Address

addr = Address(
    street="123 Rue Principale",
    city="Montréal",
    postal_code="H2X 1Y4",
    country_code="CA",
    state_or_region="QC",
)

class src.share_kernel.domain.value_objects.Email

Adresse email avec normalisation automatique (minuscules, trim).

Parameters:

value – L’adresse email

domain: str

Retourne le domaine de l’email (après le @).

from src.share_kernel.domain.value_objects import Email

email = Email(value="User@Example.COM")
assert email.value == "user@example.com"
assert email.domain == "example.com"

class src.share_kernel.domain.value_objects.PhoneNumber

Numéro de téléphone au format E.164 (+[code pays][numéro]).

Parameters:

value – Le numéro au format E.164

from src.share_kernel.domain.value_objects import PhoneNumber

phone = PhoneNumber(value="+14155551234")

class src.share_kernel.domain.value_objects.Period

Période temporelle avec validation end_date >= start_date.

Parameters:

• start_date – Date de début

• end_date – Date de fin

contains(date) → bool

Vérifie si une date est dans la période.

overlaps(other: Period) → bool

Vérifie si deux périodes se chevauchent.

duration_days() → int

Retourne la durée en jours.

Événements Domaine

class src.share_kernel.domain.domain_event.DomainEvent

Classe de base pour tous les événements domaine. Les événements sont des dataclasses gelées (frozen) et sérialisables en JSON.

event_id: str

Identifiant unique pour l’idempotence.

correlation_id: str

Lie la requête HTTP d’origine aux tâches Celery en aval.

occurred_at: str

Horodatage UTC ISO 8601.

tenant_id: str | None

Contexte tenant (si applicable).

event_type: ClassVar[str]

Nom lisible de l’événement. Défini par les sous-classes.

to_dict() → dict[str, Any]

Sérialise l’événement en dictionnaire JSON-compatible.

classmethod from_dict(data: dict) → DomainEvent

Désérialise un événement depuis un dictionnaire. Utilise le registre global pour trouver la bonne classe.

src.share_kernel.domain.domain_event.register_event(event_cls: type[DomainEvent]) → type[DomainEvent]

Décorateur de classe pour enregistrer un événement dans le registre global.

from src.share_kernel.domain.domain_event import DomainEvent, register_event
from dataclasses import dataclass

@register_event
@dataclass(frozen=True)
class InvoiceCreated(DomainEvent):
    event_type = "invoice.created"
    invoice_id: str = ""
    amount: float = 0.0

src.share_kernel.domain.domain_event.get_event_class(event_type: str) → type[DomainEvent] | None

Recherche une classe d’événement enregistrée par son event_type.

Exceptions

Exceptions DDD de base

class src.share_kernel.domain.exceptions.DomainException(Exception)

Base pour toutes les exceptions domaine.

class src.share_kernel.domain.exceptions.EntityNotFoundException(DomainException)

Entité non trouvée par son identifiant.

Parameters:

• entity_name – Nom de l’entité

• entity_id – Identifiant recherché

class src.share_kernel.domain.exceptions.BusinessRuleViolationException(DomainException)

Règle métier violée.

class src.share_kernel.domain.exceptions.InvalidValueObjectException(DomainException)

Value Object avec valeur invalide.

class src.share_kernel.domain.exceptions.ConfigurationError(DomainException)

Configuration invalide de la bibliothèque.

Exceptions Multi-tenant

class src.share_kernel.domain.exceptions.DjangoMultitenantError(Exception)

Racine pour toutes les erreurs multi-tenant.

class src.share_kernel.domain.exceptions.TenantError(DjangoMultitenantError)

Base pour les erreurs liées aux tenants.

class src.share_kernel.domain.exceptions.TenantNotFoundError(TenantError)

Tenant introuvable.

class src.share_kernel.domain.exceptions.TenantInactiveError(TenantError)

Accès à un tenant suspendu ou archivé.

class src.share_kernel.domain.exceptions.TenantSchemaError(TenantError)

Erreur de provisionnement ou migration de schéma.

class src.share_kernel.domain.exceptions.NoTenantContextError(TenantError)

Opération nécessitant un contexte tenant, mais aucun n’est défini.

class src.share_kernel.domain.exceptions.AuthorizationError(DjangoMultitenantError)

Base pour les erreurs d’autorisation.

class src.share_kernel.domain.exceptions.AccessDeniedError(AuthorizationError)

Accès refusé par l’évaluation ABAC.

Parameters:

• reason – Raison du refus

• policy_id – Identifiant de la politique (optionnel)

class src.share_kernel.domain.exceptions.ApprovalRequiredError(AuthorizationError)

Action nécessitant une approbation. Le PEP intercepte cette exception et retourne HTTP 202.

Parameters:

• approval_request_id – ID de la demande d’approbation créée

• message – Message descriptif

class src.share_kernel.domain.exceptions.MembershipError(DjangoMultitenantError)

Base pour les erreurs de membership.

class src.share_kernel.domain.exceptions.MembershipNotFoundError(MembershipError)

L’utilisateur n’a pas de membership active dans le tenant courant.

Interfaces (ABC)

class src.share_kernel.domain.interfaces.TenantIsolationBackend(ABC)

Interface pour les stratégies d’isolation des tenants.

abstractmethod activate_tenant(tenant, connection) → None

Configure la connexion DB pour le tenant actif.

abstractmethod deactivate_tenant(connection) → None

Restaure la connexion DB à son état par défaut.

abstractmethod create_tenant_storage(tenant) → None

Provisionne le stockage pour un nouveau tenant.

abstractmethod destroy_tenant_storage(tenant) → None

Détruit le stockage d’un tenant.

abstractmethod migrate_tenant(tenant) → None

Applique les migrations pour un tenant spécifique.

class src.share_kernel.domain.interfaces.ConditionEngine(ABC)

Interface pour les moteurs d’évaluation de conditions ABAC.

abstractmethod evaluate(conditions: dict, attributes: dict) → bool

Évalue les conditions par rapport aux attributs fournis.

class src.share_kernel.domain.interfaces.AttributeProvider(ABC)

Interface pour les fournisseurs d’attributs personnalisés utilisés par le PIP.

abstractmethod get_attributes(context: dict) → dict

Retourne un dictionnaire d’attributs supplémentaires.

class src.share_kernel.domain.interfaces.NotificationChannel(ABC)

Interface pour les canaux de notification des approbations.

abstractmethod send_notification(recipients, subject, message, context=None) → None

Envoie une notification aux destinataires.

Enums et Value Objects Multi-tenant

class src.share_kernel.domain.multitenant_value_objects.TenantId

Wrapper typé autour d’un UUID de tenant.

class src.share_kernel.domain.multitenant_value_objects.UserId

Wrapper typé autour d’un UUID d’utilisateur.

class src.share_kernel.domain.multitenant_value_objects.PolicyId

Wrapper typé autour d’un UUID de politique.

class src.share_kernel.domain.multitenant_value_objects.PolicyEffect(StrEnum)

• ALLOW : La politique autorise l’accès

• DENY : La politique refuse l’accès

class src.share_kernel.domain.multitenant_value_objects.ActionType(StrEnum)

Actions ABAC standard mappées depuis les actions DRF :

• LIST, READ, CREATE, UPDATE, DELETE, WILDCARD

class src.share_kernel.domain.multitenant_value_objects.TenantStatus(StrEnum)

• ACTIVE : Tenant actif

• SUSPENDED : Tenant suspendu

• ARCHIVED : Tenant archivé

class src.share_kernel.domain.multitenant_value_objects.MembershipStatus(StrEnum)

• ACTIVE : Membership active

• INVITED : Invitation en attente

• SUSPENDED : Membership suspendue

class src.share_kernel.domain.multitenant_value_objects.ApprovalStatus(StrEnum)

• PENDING, APPROVED, REJECTED, EXPIRED, CANCELLED, EXECUTION_FAILED

class src.share_kernel.domain.multitenant_value_objects.PolicyScope(StrEnum)

• GLOBAL : Politique globale (tous les tenants)

• TENANT : Politique spécifique au tenant

class src.share_kernel.domain.multitenant_value_objects.IsolationBackendType(StrEnum)

• SCHEMA, RLS, SHARED_FK

src.share_kernel.domain.multitenant_value_objects.DRF_ACTION_MAPPING

Mapping des actions DRF vers les actions ABAC :

{
    "list": "list",
    "retrieve": "read",
    "create": "create",
    "update": "update",
    "partial_update": "update",
    "destroy": "delete",
}