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 

41 def __post_init__(self): 

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

43 if self.id is None: 

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

45 

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

47 """Set the id for this entity. 

48 

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

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

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

52 """ 

53 if not self.id.is_empty(): 

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

55 return replace(self, id=id_) 

56 

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

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

59 

60 !!! Note 

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

62 nested fields must be returned as a dict. 

63 """ 

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

65 

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

67 """Check if two entities are equal. 

68 

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

70 """ 

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

72 

73 def __hash__(self) -> int: 

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

75 return hash(self.id) 

76 

77 

78class Entity[T: Identifier]: 

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

80 

81 def __init__(self, id_: T): 

82 self._id = id_ 

83 

84 @property 

85 def id(self) -> T: 

86 """Return the id of the entity.""" 

87 return self._id 

88 

89 def has_id(self) -> bool: 

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

91 

92 Returns: 

93 bool: True when the id is not empty. 

94 """ 

95 return not self._id.is_empty() 

96 

97 @classmethod 

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

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

100 

101 Args: 

102 entity: The entity to copy the values from 

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

104 

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

106 replace the existing value of an attribute. 

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

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

109 the value from the current entity. 

110 

111 Note: 

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

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

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

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

116 to find it with an underscore. 

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

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

119 the attribute of the class. 

120 """ 

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

122 for argument in ctor_arguments: 

123 # self is not needed 

124 if argument == "self": 

125 continue 

126 

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

128 if argument in changes: 

129 continue 

130 

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

132 

133 # Try to get the value of a public attribute 

134 attribute_name = argument.removesuffix("_") 

135 if attribute_name in entity.__dict__: 

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

137 continue 

138 

139 # Try to get the value of the private attribute 

140 private_attribute_name = "_" + attribute_name 

141 if private_attribute_name in entity.__dict__: 

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

143 continue 

144 

145 return entity.__class__(**changes)