Coverage for src / lexigram / contracts / domain / events.py: 45%

1"""Domain event definitions.""" 

2 

3from __future__ import annotations 

4 

5import dataclasses 

6from dataclasses import dataclass, field, replace 

7from datetime import UTC, datetime 

8from typing import Any 

9from uuid import UUID, uuid4 

10 

11 

12def _to_json_safe(value: Any) -> Any: 

13 """Convert a value to a JSON-serialisable primitive. 

14 

15 Handles UUID → str and datetime → ISO-8601 str. All other values 

16 are returned unchanged. 

17 """ 

18 if isinstance(value, UUID): 

19 return str(value) 

20 if isinstance(value, datetime): 

21 return value.isoformat() 

22 return value 

23 

24 

25@dataclass(init=False, frozen=True) 

26class DomainEvent: 

27 """Base class for domain events.""" 

28 

29 event_id: UUID = field(default_factory=uuid4, kw_only=True) 

30 occurred_at: datetime = field( 

31 default_factory=lambda: datetime.now(UTC), kw_only=True 

32 ) 

33 

34 # ---- event-sourcing fields ---- 

35 event_type: str | None = field(default=None, kw_only=True) 

36 aggregate_id: UUID | None = field(default=None, kw_only=True) 

37 aggregate_type: str | None = field(default=None, kw_only=True) 

38 sequence_number: int | None = field(default=None, kw_only=True) 

39 actor_id: str | None = field(default=None, kw_only=True) 

40 

41 # ---- schema evolution ---- 

42 schema_version: int = field( 

43 default=1, kw_only=True 

44 ) # M-07: version of the event contract 

45 

46 def __init__(self, **kwargs: Any) -> None: 

47 """Initialize the event setting base fields and any extra subclass fields. 

48 

49 Declared dataclass fields receive their default values when not 

50 supplied. Extra keyword arguments (e.g. fields declared on plain 

51 subclasses without ``@dataclass``) are set directly on the instance 

52 via ``object.__setattr__``, which bypasses the ``frozen=True`` 

53 guard safely during construction. 

54 """ 

55 for f in dataclasses.fields(self): 

56 if f.name in kwargs: 

57 object.__setattr__(self, f.name, kwargs.pop(f.name)) 

58 elif f.default is not dataclasses.MISSING: 

59 object.__setattr__(self, f.name, f.default) 

60 elif f.default_factory is not dataclasses.MISSING: 

61 object.__setattr__(self, f.name, f.default_factory()) 

62 # Extra kwargs are subclass-specific fields not declared via @dataclass. 

63 for k, v in kwargs.items(): 

64 object.__setattr__(self, k, v) 

65 self.__post_init__() 

66 

67 def __post_init__(self) -> None: 

68 # lifecycle hook 

69 if self.event_type is None: 

70 object.__setattr__(self, "event_type", self.__class__.__name__) 

71 

72 def to_dict(self) -> dict[str, Any]: 

73 """Return a JSON-serializable dictionary representation.""" 

74 return {k: _to_json_safe(v) for k, v in dataclasses.asdict(self).items()} 

75 

76 @classmethod 

77 def from_dict(cls, data: dict[str, Any]) -> DomainEvent: 

78 return cls(**data) 

79 

80 def for_aggregate(self, aggregate_id: UUID, aggregate_type: str) -> DomainEvent: 

81 return replace( 

82 self, 

83 aggregate_id=aggregate_id, 

84 aggregate_type=aggregate_type, 

85 ) 

86 

87 

88__all__ = ["DomainEvent"]