Coverage for src / lexigram / contracts / domain / aggregates.py: 0%

1"""Aggregate root protocol. 

2 

3The concrete ``AggregateRoot`` base class lives in ``lexigram.domain.models.aggregate`` 

4(the core ``lexigram`` package). Contracts expose only the structural protocol 

5so that any package can type-check against it without pulling in implementation 

6code. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, Protocol, runtime_checkable 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts.domain.events import DomainEvent 

15 

16 

17@runtime_checkable 

18class AggregateRootProtocol(Protocol): 

19 """Structural protocol for aggregate root objects. 

20 

21 Any class that implements event buffering and consistency boundaries 

22 satisfies this protocol. Type-check against it instead of the concrete 

23 ``AggregateRoot`` base class where possible. 

24 """ 

25 

26 def add_event(self, event: DomainEvent) -> None: 

27 """Register a domain event with the aggregate.""" 

28 ... 

29 

30 def collect_events(self) -> list[DomainEvent]: 

31 """Return and clear all buffered domain events.""" 

32 ... 

33 

34 def pull_events(self) -> list[DomainEvent]: 

35 """Return buffered events without clearing them.""" 

36 ... 

37 

38 def clear_events(self) -> None: 

39 """Discard all buffered domain events.""" 

40 ... 

41 

42 @property 

43 def has_uncommitted_events(self) -> bool: 

44 """Return True when there are buffered events not yet published.""" 

45 ... 

46 

47 

48__all__ = ["AggregateRootProtocol"]