Coverage for object_streams/events.py: 85%

1"""Serializable event primitives used by subscriptions and the outbox.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Mapping 

6from collections.abc import Sequence 

7from dataclasses import dataclass 

8from dataclasses import field 

9from enum import StrEnum 

10from typing import Any 

11 

12from django.db import models 

13 

14 

15__all__ = ( 

16 "EventOperation", 

17 "ListAction", 

18 "ObjectRef", 

19 "SourceRef", 

20 "StreamEvent", 

21) 

22 

23 

24class EventOperation(StrEnum): 

25 CREATED = "created" 

26 UPDATED = "updated" 

27 DELETED = "deleted" 

28 

29 

30class ListAction(StrEnum): 

31 ADDED = "added" 

32 CHANGED = "changed" 

33 REMOVED = "removed" 

34 DELETED = "deleted" 

35 RESYNC_REQUIRED = "resync_required" 

36 

37 

38def _model_label(model: type[models.Model] | str) -> str: 

39 if isinstance(model, str): 

40 return model 

41 return model._meta.label 

42 

43 

44def _coerce_pk(pk: Any) -> str: 

45 if pk is None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 msg = "Object stream references require a primary key." 

47 raise ValueError(msg) 

48 return str(pk) 

49 

50 

51@dataclass(frozen=True, slots=True) 

52class ObjectRef: 

53 """A serializable reference to a Django model object.""" 

54 

55 model: str 

56 pk: str 

57 

58 def __post_init__(self): 

59 object.__setattr__(self, "model", _model_label(self.model)) 

60 object.__setattr__(self, "pk", _coerce_pk(self.pk)) 

61 

62 @classmethod 

63 def for_model(cls, model: type[models.Model] | str, pk: Any) -> ObjectRef: 

64 return cls(model=_model_label(model), pk=_coerce_pk(pk)) 

65 

66 @classmethod 

67 def from_instance(cls, instance: models.Model) -> ObjectRef: 

68 return cls.for_model(instance.__class__, instance.pk) 

69 

70 def as_dict(self) -> dict[str, str]: 

71 return { 

72 "model": self.model, 

73 "pk": self.pk, 

74 } 

75 

76 

77@dataclass(frozen=True, slots=True) 

78class SourceRef: 

79 """A serializable reference to the source that produced a subject event.""" 

80 

81 model: str | None = None 

82 pk: str | None = None 

83 history_model: str | None = None 

84 history_id: str | None = None 

85 metadata: Mapping[str, Any] = field(default_factory=dict) 

86 

87 def __post_init__(self): 

88 if self.model is not None: 88 ↛ 90line 88 didn't jump to line 90 because the condition on line 88 was always true

89 object.__setattr__(self, "model", _model_label(self.model)) 

90 if self.pk is not None: 90 ↛ 92line 90 didn't jump to line 92 because the condition on line 90 was always true

91 object.__setattr__(self, "pk", str(self.pk)) 

92 if self.history_model is not None: 

93 object.__setattr__(self, "history_model", _model_label(self.history_model)) 

94 if self.history_id is not None: 

95 object.__setattr__(self, "history_id", str(self.history_id)) 

96 object.__setattr__(self, "metadata", dict(self.metadata)) 

97 

98 @classmethod 

99 def from_instance(cls, instance: models.Model, *, metadata: Mapping[str, Any] | None = None) -> SourceRef: 

100 return cls(model=instance.__class__, pk=instance.pk, metadata=metadata or {}) 

101 

102 def as_dict(self) -> dict[str, Any]: 

103 value: dict[str, Any] = {} 

104 if self.model: 104 ↛ 106line 104 didn't jump to line 106 because the condition on line 104 was always true

105 value["model"] = self.model 

106 if self.pk: 106 ↛ 108line 106 didn't jump to line 108 because the condition on line 106 was always true

107 value["pk"] = self.pk 

108 if self.history_model: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 value["history_model"] = self.history_model 

110 if self.history_id: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 value["history_id"] = self.history_id 

112 if self.metadata: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true

113 value["metadata"] = dict(self.metadata) 

114 return value 

115 

116 

117@dataclass(frozen=True, slots=True) 

118class StreamEvent: 

119 """A subscription-relative event suitable for WebSocket or outbox delivery.""" 

120 

121 subject: ObjectRef 

122 facet: str = "object" 

123 op: EventOperation | str = EventOperation.UPDATED 

124 cursor: int | None = None 

125 subscription_id: str | None = None 

126 list_action: ListAction | str | None = None 

127 changed_fields: Sequence[str] = field(default_factory=tuple) 

128 source: ObjectRef | SourceRef | None = None 

129 fetch: bool = True 

130 before: Mapping[str, Any] | None = None 

131 after: Mapping[str, Any] | None = None 

132 metadata: Mapping[str, Any] = field(default_factory=dict) 

133 

134 def __post_init__(self): 

135 object.__setattr__(self, "op", str(self.op)) 

136 if self.list_action is not None: 

137 object.__setattr__(self, "list_action", str(self.list_action)) 

138 object.__setattr__(self, "changed_fields", tuple(self.changed_fields)) 

139 object.__setattr__(self, "metadata", dict(self.metadata)) 

140 if self.before is not None: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 object.__setattr__(self, "before", dict(self.before)) 

142 if self.after is not None: 

143 object.__setattr__(self, "after", dict(self.after)) 

144 

145 def as_dict(self) -> dict[str, Any]: 

146 value: dict[str, Any] = { 

147 "type": "event", 

148 "subject": self.subject.as_dict(), 

149 "facet": self.facet, 

150 "op": self.op, 

151 "changed_fields": list(self.changed_fields), 

152 "fetch": self.fetch, 

153 } 

154 if self.subscription_id is not None: 154 ↛ 156line 154 didn't jump to line 156 because the condition on line 154 was always true

155 value["subscription_id"] = self.subscription_id 

156 if self.cursor is not None: 156 ↛ 158line 156 didn't jump to line 158 because the condition on line 156 was always true

157 value["cursor"] = self.cursor 

158 if self.list_action is not None: 158 ↛ 160line 158 didn't jump to line 160 because the condition on line 158 was always true

159 value["list_action"] = self.list_action 

160 if self.source is not None: 

161 value["source"] = self.source.as_dict() 

162 if self.before is not None: 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true

163 value["before"] = dict(self.before) 

164 if self.after is not None: 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true

165 value["after"] = dict(self.after) 

166 if self.metadata: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 value["metadata"] = dict(self.metadata) 

168 return value