Coverage for object_streams/producers.py: 90%

1"""Producer helpers that turn source changes into outbox events.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterable 

6from collections.abc import Mapping 

7from collections.abc import Sequence 

8from typing import Any 

9 

10from django.db import models 

11 

12from object_streams.events import EventOperation 

13from object_streams.events import ObjectRef 

14from object_streams.events import SourceRef 

15from object_streams.events import StreamEvent 

16from object_streams.models import ObjectStreamEvent 

17from object_streams.outbox import create_outbox_event 

18from object_streams.outbox import enqueue_outbox_event 

19from object_streams.registry import ObjectStreamRegistry 

20from object_streams.registry import registry as default_registry 

21 

22 

23__all__ = ( 

24 "build_source_events", 

25 "create_source_events", 

26 "enqueue_source_events", 

27) 

28 

29 

30def build_source_events( 

31 instance: models.Model, 

32 *, 

33 op: EventOperation | str = EventOperation.UPDATED, 

34 changed_fields: Sequence[str] = (), 

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

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

37 metadata: Mapping[str, Any] | None = None, 

38 registry: ObjectStreamRegistry = default_registry, 

39) -> tuple[StreamEvent, ...]: 

40 """Build stream events for a changed source instance.""" 

41 

42 events = [] 

43 for registration in registry: 

44 for source in registration.sources: 

45 if not _source_matches(source, instance): 

46 continue 

47 source_changed_fields = _source_changed_fields(source, instance) or changed_fields 

48 source_ref = _source_ref(source, instance) 

49 for subject in _subjects_for_source(source, instance): 

50 if subject.model != registration.model_label: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 continue 

52 events.append( 

53 StreamEvent( 

54 subject=subject, 

55 facet=str(getattr(source, "facet", "object")), 

56 op=op, 

57 changed_fields=tuple(source_changed_fields), 

58 source=source_ref, 

59 before=before, 

60 after=after, 

61 metadata=metadata or {}, 

62 ) 

63 ) 

64 return tuple(events) 

65 

66 

67def create_source_events( 

68 instance: models.Model, 

69 *, 

70 op: EventOperation | str = EventOperation.UPDATED, 

71 changed_fields: Sequence[str] = (), 

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

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

74 metadata: Mapping[str, Any] | None = None, 

75 notify: bool = True, 

76 registry: ObjectStreamRegistry = default_registry, 

77 using: str | None = None, 

78) -> tuple[ObjectStreamEvent, ...]: 

79 """Create outbox rows for a changed source instance immediately.""" 

80 

81 return tuple( 

82 create_outbox_event(event, notify=notify, using=using) 

83 for event in build_source_events( 

84 instance, 

85 op=op, 

86 changed_fields=changed_fields, 

87 before=before, 

88 after=after, 

89 metadata=metadata, 

90 registry=registry, 

91 ) 

92 ) 

93 

94 

95def enqueue_source_events( 

96 instance: models.Model, 

97 *, 

98 op: EventOperation | str = EventOperation.UPDATED, 

99 changed_fields: Sequence[str] = (), 

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

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

102 metadata: Mapping[str, Any] | None = None, 

103 notify: bool = True, 

104 registry: ObjectStreamRegistry = default_registry, 

105 using: str | None = None, 

106) -> tuple[StreamEvent, ...]: 

107 """Schedule outbox rows for a changed source instance after commit.""" 

108 

109 events = build_source_events( 

110 instance, 

111 op=op, 

112 changed_fields=changed_fields, 

113 before=before, 

114 after=after, 

115 metadata=metadata, 

116 registry=registry, 

117 ) 

118 for event in events: 

119 enqueue_outbox_event(event, using=using, notify=notify) 

120 return events 

121 

122 

123def _source_matches(source: Any, instance: models.Model) -> bool: 

124 matches = getattr(source, "matches", None) 

125 if matches is not None: 

126 return bool(matches(instance)) 

127 

128 source_model = getattr(source, "source_model", None) 

129 if source_model is None: 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 return True 

131 if isinstance(source_model, str): 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 return source_model == instance._meta.label 

133 return isinstance(instance, source_model) 

134 

135 

136def _source_changed_fields(source: Any, instance: models.Model) -> Sequence[str]: 

137 changed_fields = getattr(source, "changed_fields", None) 

138 if changed_fields is None: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 return () 

140 return tuple(changed_fields(instance)) 

141 

142 

143def _source_ref(source: Any, instance: models.Model) -> ObjectRef | SourceRef: 

144 source_ref = getattr(source, "source_ref", None) 

145 if source_ref is not None: 

146 return source_ref(instance) 

147 return SourceRef.from_instance(instance) 

148 

149 

150def _subjects_for_source(source: Any, instance: models.Model) -> Iterable[ObjectRef]: 

151 return tuple(source.subjects_for_source(instance))