Coverage for src/kwai/core/events/event.py: 100%

1"""Module that defines the base class Event.""" 

2 

3import dataclasses 

4 

5from dataclasses import dataclass 

6from typing import ClassVar 

7 

8from kwai.core.domain.value_objects.timestamp import Timestamp 

9 

10 

11@dataclass(kw_only=True, frozen=True) 

12class EventMeta: 

13 """Metadata for the event. 

14 

15 The combination of version, module and name should result in a unique 

16 full name in kwai and will be used as name for a Redis stream. 

17 

18 Attributes: 

19 version: The version of the event. 

20 module: The module that the event belongs to (e.g. identity) 

21 name: The name of the event. 

22 """ 

23 

24 version: str = "v1" 

25 module: str 

26 name: str 

27 

28 @property 

29 def full_name(self) -> str: 

30 """Return the full name of the event.""" 

31 return f"kwai/{self.version}/{self.module}/{self.name}" 

32 

33 

34@dataclass(kw_only=True, frozen=True) 

35class Event: 

36 """Base class for all events.""" 

37 

38 meta: ClassVar[EventMeta] 

39 

40 @property 

41 def data(self) -> dict: 

42 """Returns a dict that can be used to serialize the event.""" 

43 return { 

44 "meta": { 

45 **dataclasses.asdict(self.__class__.meta), 

46 "date": str(Timestamp.create_now()), 

47 }, 

48 "data": {**dataclasses.asdict(self)}, 

49 }