Coverage for formkit_ninja / form_submission / querysets.py: 44.83%

1""" 

2Custom querysets for Submission annotations. 

3 

4Provides efficient annotations for: 

5- Import failure detection (latest SeparatedSubmissionImport per SeparatedSubmission) 

6- Unresolved flag detection and JSON aggregation 

7 

8Requires PostgreSQL: uses JSONBAgg and JSONObject (django.contrib.postgres). 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TypeVar 

14 

15from django.contrib.postgres.aggregates import JSONBAgg 

16from django.db import models 

17from django.db.models import Case, Exists, OuterRef, Subquery, Value, When 

18from django.db.models.functions import JSONObject 

19 

20_QS = TypeVar("_QS", bound=models.QuerySet) 

21 

22 

23class SubmissionQuerySet(models.QuerySet): 

24 """ 

25 Custom queryset for Submission with annotation helpers. 

26 

27 Usage: 

28 Submission.objects.with_import_failure() 

29 Submission.objects.with_unresolved_flags() 

30 Submission.objects.with_import_failure().with_unresolved_flags() 

31 """ 

32 

33 def with_import_failure(self: _QS) -> _QS: 

34 """ 

35 Annotate each Submission with ``has_import_failure`` (bool). 

36 

37 ``True`` when **any** related SeparatedSubmission has a latest 

38 SeparatedSubmissionImport where ``success=False``. 

39 """ 

40 from formkit_ninja.form_submission.models import ( 

41 SeparatedSubmission, 

42 SeparatedSubmissionImport, 

43 ) 

44 

45 # Latest import result per SeparatedSubmission 

46 latest_import_success = SeparatedSubmissionImport.objects.filter(submission=OuterRef("pk")).order_by("-created").values("success")[:1] 

47 

48 # SeparatedSubmissions whose latest import failed 

49 failed_subs = SeparatedSubmission.objects.filter(submission=OuterRef("pk")).annotate(latest_success=Subquery(latest_import_success)).filter(latest_success=False) 

50 

51 return self.annotate(has_import_failure=Exists(failed_subs)) 

52 

53 def with_unresolved_flags(self: _QS) -> _QS: 

54 """ 

55 Annotate each Submission with: 

56 

57 - ``has_unresolved_flags`` (bool) — True if any unresolved Flag exists 

58 - ``unresolved_flags_json`` (JSON array) — ``[{"flag_type", "message", "severity"}, ...]`` 

59 Ordered by flag ``created`` descending (newest first). When there are no 

60 unresolved flags, ``unresolved_flags_json`` is ``None`` (not ``[]``). 

61 """ 

62 from formkit_ninja.form_submission.models import Flag 

63 

64 unresolved = Flag.objects.filter( 

65 separated_submission__submission=OuterRef("pk"), 

66 resolved_at__isnull=True, 

67 ) 

68 

69 flags_json = ( 

70 Flag.objects.filter( 

71 separated_submission__submission=OuterRef("pk"), 

72 resolved_at__isnull=True, 

73 ) 

74 .order_by() # clear default ordering 

75 .values("separated_submission__submission") # grouping key 

76 .annotate( 

77 flags=JSONBAgg( 

78 JSONObject( 

79 flag_type="flag_type", 

80 message="message", 

81 severity="severity", 

82 ), 

83 ordering="-created", 

84 ) 

85 ) 

86 .values("flags")[:1] 

87 ) 

88 

89 return self.annotate( 

90 has_unresolved_flags=Exists(unresolved), 

91 unresolved_flags_json=Subquery(flags_json), 

92 ) 

93 

94 

95class SeparatedSubmissionQuerySet(models.QuerySet): 

96 """ 

97 Custom queryset for SeparatedSubmission with annotation helpers. 

98 

99 Usage: 

100 SeparatedSubmission.objects.with_import_failure() 

101 SeparatedSubmission.objects.with_unresolved_flags() 

102 SeparatedSubmission.objects.with_import_failure().with_unresolved_flags() 

103 """ 

104 

105 def with_import_failure(self: _QS) -> _QS: 

106 """ 

107 Annotate each SeparatedSubmission with ``has_import_failure`` (bool). 

108 

109 ``True`` when its latest ``SeparatedSubmissionImport`` has ``success=False``. 

110 Also annotates ``latest_import_success`` (bool | None) for the latest 

111 import; ``None`` when there are no imports. 

112 """ 

113 from django.db.models import BooleanField 

114 

115 from formkit_ninja.form_submission.models import SeparatedSubmissionImport 

116 

117 # Subquery: success of the latest import for this SeparatedSubmission (no nesting, 

118 # so OuterRef("pk") correctly refers to SeparatedSubmission.id / UUID). 

119 latest_success = ( 

120 SeparatedSubmissionImport.objects.filter( 

121 submission=OuterRef("pk"), 

122 ) 

123 .order_by("-created") 

124 .values("success")[:1] 

125 ) 

126 

127 return self.annotate( 

128 latest_import_success=Subquery(latest_success), 

129 ).annotate( 

130 has_import_failure=Case( 

131 When(latest_import_success=False, then=Value(True)), 

132 default=Value(False), 

133 output_field=BooleanField(), 

134 ), 

135 ) 

136 

137 def with_unresolved_flags(self: _QS) -> _QS: 

138 """ 

139 Annotate each SeparatedSubmission with: 

140 

141 - ``has_unresolved_flags`` (bool) — True if any unresolved Flag exists 

142 - ``unresolved_flags_json`` (JSON array) — ``[{"flag_type", "message", "severity"}, ...]`` 

143 Ordered by flag ``created`` descending (newest first). When there are no 

144 unresolved flags, ``unresolved_flags_json`` is ``None`` (not ``[]``). 

145 """ 

146 from formkit_ninja.form_submission.models import Flag 

147 

148 unresolved = Flag.objects.filter( 

149 separated_submission=OuterRef("pk"), 

150 resolved_at__isnull=True, 

151 ) 

152 

153 flags_json = ( 

154 Flag.objects.filter( 

155 separated_submission=OuterRef("pk"), 

156 resolved_at__isnull=True, 

157 ) 

158 .order_by() # clear default ordering 

159 .values("separated_submission") # grouping key 

160 .annotate( 

161 flags=JSONBAgg( 

162 JSONObject( 

163 flag_type="flag_type", 

164 message="message", 

165 severity="severity", 

166 ), 

167 ordering="-created", 

168 ) 

169 ) 

170 .values("flags")[:1] 

171 ) 

172 

173 return self.annotate( 

174 has_unresolved_flags=Exists(unresolved), 

175 unresolved_flags_json=Subquery(flags_json), 

176 )