Coverage for src/kwai/core/domain/entity.py: 92%

1"""Module that defines a generic entity.""" 

2 

3import inspect 

4 

5from dataclasses import dataclass, field, fields, replace 

6from typing import ( 

7 Any, 

8 ClassVar, 

9 Self, 

10) 

11 

12from kwai.core.domain.value_objects.identifier import Identifier, IntIdentifier 

13from kwai.core.domain.value_objects.traceable_time import TraceableTime 

14 

15 

16@dataclass(frozen=True, slots=True, eq=False) 

17class DataclassEntity: 

18 """A base class for an entity. 

19 

20 An entity is immutable, so it cannot be modified. A method of an entity that 

21 changes the entity must allways return a new entity instance with the changed 

22 data. The method replace of dataclasses can be used for this. 

23 

24 Currently, this is a separate class to make it possible to migrate to this 

25 new class. In the future, the Entity class will be removed and this class 

26 will be renamed to Entity. 

27 

28 By default, id is of type IntIdentifier. Overwrite ID in an entity class if 

29 another identifier should be used. 

30 

31 Attributes: 

32 id: The id of the entity. 

33 traceable_time: Keeps track of the creation and update timestamp of the entity. 

34 """ 

35 

36 ID: ClassVar = IntIdentifier 

37 

38 id: ID = None 

39 traceable_time: TraceableTime = field(default_factory=TraceableTime) 

40 version = 0 

41 

42 def __post_init__(self): 

43 """When is id is not set, a default id is created.""" 

44 if self.id is None: 

45 object.__setattr__(self, "id", self.ID()) 

46 

47 def set_id(self, id_: ID) -> Self: 

48 """Set the id for this entity. 

49 

50 This will raise a ValueError if the id was already set. 

51 If you need an entity with the same data but with another id, you should create 

52 a new entity with dataclasses.replace and replace the id. 

53 """ 

54 if not self.id.is_empty(): 

55 raise ValueError(f"{self.__class__.__name__} has already an ID: {self.id}") 

56 return replace(self, id=id_) 

57 

58 def shallow_dict(self) -> dict[str, Any]: 

59 """Return a dictionary representation of the entity. 

60 

61 !!! Note 

62 This method is not recursive. Use asdict from dataclasses when also 

63 nested fields must be returned as a dict. 

64 """ 

65 return {f.name: getattr(self, f.name) for f in fields(self)} 

66 

67 def __eq__(self, other: Any) -> bool: 

68 """Check if two entities are equal. 

69 

70 An entity equals another entity when the id is the same. 

71 """ 

72 return isinstance(other, type(self)) and other.id == self.id 

73 

74 def __hash__(self) -> int: 

75 """Generate a hash for this entity.""" 

76 return hash(self.id) 

77 

78 

79class Entity[T: Identifier]: 

80 """A base class for an entity.""" 

81 

82 def __init__(self, id_: T): 

83 self._id = id_ 

84 

85 @property 

86 def id(self) -> T: 

87 """Return the id of the entity.""" 

88 return self._id 

89 

90 def has_id(self) -> bool: 

91 """Check if this entity has a valid id. 

92 

93 Returns: 

94 bool: True when the id is not empty. 

95 """ 

96 return not self._id.is_empty() 

97 

98 @classmethod 

99 def replace(cls, entity: Self, **changes: Any) -> Any: 

100 """Return a new entity from the existing entity. 

101 

102 Args: 

103 entity: The entity to copy the values from 

104 changes: the values to override when creating the new entity. 

105 

106 Use the same keyword arguments as used on the class constructor (__init__) to 

107 replace the existing value of an attribute. 

108 The class constructor will be called to create this new entity. 

109 The arguments of the constructor that are not passed in "changes", will get 

110 the value from the current entity. 

111 

112 Note: 

113 To make it clear that the attributes of an entity are private, they are 

114 prefixed with an underscore. The name of the keyword argument does not 

115 contain this underscore. This method will try to find the attribute first 

116 without underscore. When no attribute exists with that name, it will try 

117 to find it with an underscore. 

118 When an argument of the constructor contains an underscore as suffix (to 

119 avoid naming conflicts for example), the underscore will be removed to find 

120 the attribute of the class. 

121 """ 

122 ctor_arguments = inspect.signature(entity.__class__.__init__).parameters 

123 for argument in ctor_arguments: 

124 # self is not needed 

125 if argument == "self": 

126 continue 

127 

128 # We already have the value when the argument is passed 

129 if argument in changes: 

130 continue 

131 

132 # The attribute is not passed, so we need to use the original value. 

133 

134 # Try to get the value of a public attribute 

135 attribute_name = argument.removesuffix("_") 

136 if attribute_name in entity.__dict__: 

137 changes[argument] = entity.__dict__[attribute_name] 

138 continue 

139 

140 # Try to get the value of the private attribute 

141 private_attribute_name = "_" + attribute_name 

142 if private_attribute_name in entity.__dict__: 

143 changes[argument] = entity.__dict__[private_attribute_name] 

144 continue 

145 

146 return entity.__class__(**changes)