Coverage for object_streams/outbox.py: 88%

1"""Helpers for writing stream events to the replayable outbox.""" 

2 

3from __future__ import annotations 

4 

5from django.apps import apps 

6from django.contrib.contenttypes.models import ContentType 

7from django.db import models 

8from django.db import transaction 

9 

10from object_streams.events import ObjectRef 

11from object_streams.events import SourceRef 

12from object_streams.events import StreamEvent 

13from object_streams.models import ObjectStreamEvent 

14from object_streams.models import ObjectStreamOutboxState 

15from object_streams.postgres import notify_outbox_event 

16 

17 

18__all__ = ( 

19 "assign_outbox_cursors", 

20 "broadcasted_through_cursor", 

21 "create_outbox_event", 

22 "enqueue_outbox_event", 

23 "latest_outbox_cursor", 

24 "outbox_events_after", 

25 "outbox_events_pending_broadcast", 

26 "pruned_through_cursor", 

27 "record_broadcasted_through", 

28 "record_pruned_through", 

29 "replay_is_complete", 

30) 

31 

32 

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

34 if isinstance(model, str): 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 return model 

36 return model._meta.label 

37 

38 

39def _content_type_for_model_label(model_label: str, *, using: str | None = None) -> ContentType: 

40 app_label, model_name = model_label.split(".", 1) 

41 model_class = apps.get_model(app_label, model_name) 

42 if model_class is None: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 msg = f"No installed model matches {model_label!r}." 

44 raise LookupError(msg) 

45 manager = ContentType.objects 

46 if using is not None: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 manager = manager.db_manager(using) 

48 return manager.get_for_model(model_class) 

49 

50 

51def _source_content_type(source: ObjectRef | SourceRef | None, *, using: str | None = None) -> ContentType | None: 

52 if source is None: 

53 return None 

54 if isinstance(source, ObjectRef): 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true

55 return _content_type_for_model_label(source.model, using=using) 

56 if source.model: 56 ↛ 58line 56 didn't jump to line 58 because the condition on line 56 was always true

57 return _content_type_for_model_label(source.model, using=using) 

58 return None 

59 

60 

61def _source_object_id(source: ObjectRef | SourceRef | None) -> str: 

62 if source is None: 

63 return "" 

64 return source.pk or "" 

65 

66 

67def create_outbox_event(event: StreamEvent, *, notify: bool = True, using: str | None = None) -> ObjectStreamEvent: 

68 """Persist a stream event immediately and return the created outbox row.""" 

69 

70 source_history_content_type = None 

71 source_history_id = "" 

72 if isinstance(event.source, SourceRef) and event.source.history_model: 

73 source_history_content_type = _content_type_for_model_label(event.source.history_model, using=using) 

74 source_history_id = event.source.history_id or "" 

75 

76 manager = ObjectStreamEvent.objects 

77 if using is not None: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 manager = manager.db_manager(using) 

79 

80 row = manager.create( 

81 subject_content_type=_content_type_for_model_label(event.subject.model, using=using), 

82 subject_object_id=event.subject.pk, 

83 source_content_type=_source_content_type(event.source, using=using), 

84 source_object_id=_source_object_id(event.source), 

85 source_history_content_type=source_history_content_type, 

86 source_history_id=source_history_id, 

87 facet=event.facet, 

88 op=event.op, 

89 changed_fields=list(event.changed_fields), 

90 before=event.before, 

91 after=event.after, 

92 metadata=dict(event.metadata), 

93 ) 

94 connection = transaction.get_connection(using=using) 

95 if not connection.in_atomic_block: 

96 assign_outbox_cursors(using=using) 

97 row.refresh_from_db(fields=["cursor"], using=using) 

98 if notify: 

99 notify_outbox_event(row.pk, using=using) 

100 return row 

101 

102 

103def enqueue_outbox_event(event: StreamEvent, *, using: str | None = None, notify: bool = True) -> None: 

104 """Write an event after the current database transaction commits.""" 

105 

106 transaction.on_commit(lambda: create_outbox_event(event, notify=notify, using=using), using=using) 

107 

108 

109def _state_manager(using: str | None = None): 

110 manager = ObjectStreamOutboxState.objects 

111 if using is not None: 

112 manager = manager.db_manager(using) 

113 return manager 

114 

115 

116def assign_outbox_cursors(*, using: str | None = None, limit: int | None = None) -> tuple[ObjectStreamEvent, ...]: 

117 """Assign commit-visible delivery cursors to captured outbox rows.""" 

118 

119 manager = ObjectStreamEvent.objects 

120 if using is not None: 

121 manager = manager.db_manager(using) 

122 

123 with transaction.atomic(using=using): 

124 state, _ = _state_manager(using).select_for_update().get_or_create(pk=1) 

125 queryset = manager.select_for_update().filter(cursor__isnull=True).order_by("id") 

126 if limit is not None: 

127 queryset = queryset[:limit] 

128 rows = list(queryset) 

129 if not rows: 

130 return () 

131 

132 next_cursor = state.next_cursor 

133 for row in rows: 

134 row.cursor = next_cursor 

135 next_cursor += 1 

136 manager.bulk_update(rows, ["cursor"]) 

137 state.next_cursor = next_cursor 

138 state.save(update_fields=["next_cursor"]) 

139 return tuple(rows) 

140 

141 

142def latest_outbox_cursor(*, using: str | None = None) -> int: 

143 """Return the latest global outbox cursor, or 0 when the outbox is empty.""" 

144 

145 manager = ObjectStreamEvent.objects 

146 if using is not None: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 manager = manager.db_manager(using) 

148 return manager.filter(cursor__isnull=False).order_by("-cursor").values_list("cursor", flat=True).first() or 0 

149 

150 

151def pruned_through_cursor(*, using: str | None = None) -> int: 

152 """Return the highest outbox cursor removed by retention pruning.""" 

153 

154 return _state_manager(using).values_list("pruned_through", flat=True).first() or 0 

155 

156 

157def broadcasted_through_cursor(*, using: str | None = None) -> int: 

158 """Return the highest delivery cursor successfully fanned out by the listener.""" 

159 

160 return _state_manager(using).values_list("broadcasted_through", flat=True).first() or 0 

161 

162 

163def record_pruned_through(cursor: int, *, using: str | None = None) -> None: 

164 """Raise the pruning watermark to a cursor that has been deleted.""" 

165 

166 state, created = _state_manager(using).get_or_create(pk=1, defaults={"pruned_through": cursor}) 

167 if not created and cursor > state.pruned_through: 167 ↛ exitline 167 didn't return from function 'record_pruned_through' because the condition on line 167 was always true

168 state.pruned_through = cursor 

169 state.save(update_fields=["pruned_through"]) 

170 

171 

172def record_broadcasted_through(cursor: int, *, using: str | None = None) -> None: 

173 """Advance the durable listener watermark after successful fanout.""" 

174 

175 with transaction.atomic(using=using): 

176 state, _ = _state_manager(using).select_for_update().get_or_create(pk=1) 

177 if cursor > state.broadcasted_through: 177 ↛ exitline 177 didn't jump to the function exit

178 state.broadcasted_through = cursor 

179 state.save(update_fields=["broadcasted_through"]) 

180 

181 

182def replay_is_complete(cursor: int) -> bool: 

183 """Return whether every outbox row after a cursor is still retained.""" 

184 

185 return cursor >= pruned_through_cursor() 

186 

187 

188def outbox_events_after( 

189 cursor: int, 

190 *, 

191 model: type[models.Model] | str | None = None, 

192 subject: ObjectRef | None = None, 

193 through_cursor: int | None = None, 

194 limit: int | None = None, 

195) -> models.QuerySet: 

196 """Return outbox rows after a cursor, optionally scoped to a model or subject.""" 

197 

198 queryset = ObjectStreamEvent.objects.filter(cursor__gt=cursor).order_by("cursor") 

199 if through_cursor is not None: 

200 queryset = queryset.filter(cursor__lte=through_cursor) 

201 if subject is not None: 

202 queryset = queryset.filter( 

203 subject_content_type=_content_type_for_model_label(subject.model), 

204 subject_object_id=subject.pk, 

205 ) 

206 elif model is not None: 

207 queryset = queryset.filter(subject_content_type=_content_type_for_model_label(_model_label(model))) 

208 if limit is not None: 

209 return queryset[:limit] 

210 return queryset 

211 

212 

213def outbox_events_pending_broadcast( 

214 *, 

215 using: str | None = None, 

216 limit: int | None = None, 

217) -> models.QuerySet: 

218 """Return cursor-assigned rows after the listener's durable watermark.""" 

219 

220 manager = ObjectStreamEvent.objects 

221 if using is not None: 221 ↛ 223line 221 didn't jump to line 223 because the condition on line 221 was always true

222 manager = manager.db_manager(using) 

223 queryset = ( 

224 manager.select_related( 

225 "subject_content_type", 

226 "source_content_type", 

227 "source_history_content_type", 

228 ) 

229 .filter(cursor__gt=broadcasted_through_cursor(using=using)) 

230 .order_by("cursor") 

231 ) 

232 if limit is not None: 232 ↛ 234line 232 didn't jump to line 234 because the condition on line 232 was always true

233 return queryset[:limit] 

234 return queryset