Coverage for app/backend/src/couchers/sql.py: 96%

1from typing import TYPE_CHECKING, Any 

2 

3from sqlalchemy import ColumnElement, and_, false, or_, select, true 

4from sqlalchemy.orm import InstrumentedAttribute, aliased 

5from sqlalchemy.sql import Select, exists, union 

6 

7from couchers.models import ( 

8 ModerationState, 

9 ModerationVisibility, 

10 SignupFlow, 

11 User, 

12 UserBlock, 

13 get_moderated_models, 

14) 

15from couchers.utils import is_valid_email, is_valid_user_id, is_valid_username 

16 

17if TYPE_CHECKING: 

18 from couchers.context import CouchersContext 

19 from couchers.materialized_views import LiteUser 

20 from couchers.models.moderation import ModeratedContent 

21 

22 type _UserLike = type[User | LiteUser | SignupFlow] 

23 type _User = type[User | LiteUser] 

24 

25 

26def username_or_email(value: str, table: _UserLike = User) -> ColumnElement[bool]: 

27 if is_valid_username(value): 

28 return table.username == value 

29 elif is_valid_email(value) and hasattr(table, "email"): 

30 return table.email == value 

31 # no fields match, this will return no rows 

32 return false() 

33 

34 

35def username_or_id(value: str, table: _UserLike = User) -> ColumnElement[bool]: 

36 if is_valid_username(value): 

37 return table.username == value 

38 elif is_valid_user_id(value): 

39 return table.id == int(value) 

40 # no fields match, this will return no rows 

41 return false() 

42 

43 

44def username_or_email_or_id(value: str) -> ColumnElement[bool]: 

45 # Should only be used for admin APIs, etc. 

46 if is_valid_username(value): 

47 return User.username == value 

48 elif is_valid_email(value): 

49 return User.email == value 

50 elif is_valid_user_id(value): 50 ↛ 53line 50 didn't jump to line 53 because the condition on line 50 was always true

51 return User.id == int(value) 

52 # no fields match, this will return no rows 

53 return false() 

54 

55 

56def _shadow_clause(context: CouchersContext, table: _User) -> ColumnElement[bool]: 

57 if context.is_logged_in(): 

58 return or_(table.shadowed_at.is_(None), table.id == context.user_id) 

59 return table.shadowed_at.is_(None) 

60 

61 

62def _users_block_each_other( 

63 user_a: InstrumentedAttribute[int], user_b: InstrumentedAttribute[int] 

64) -> ColumnElement[bool]: 

65 """True when either of the two users (referenced by id columns) has blocked the other.""" 

66 return exists( 

67 select(1) 

68 .select_from(UserBlock) 

69 .where( 

70 or_( 

71 and_(UserBlock.blocking_user_id == user_a, UserBlock.blocked_user_id == user_b), 

72 and_(UserBlock.blocking_user_id == user_b, UserBlock.blocked_user_id == user_a), 

73 ) 

74 ) 

75 ) 

76 

77 

78def users_visible(context: CouchersContext, table: _User = User) -> ColumnElement[bool]: 

79 """ 

80 Filters out users that should not be visible: blocked, deleted, banned, or shadowed (to others). 

81 

82 Filters the given table, assuming it's already joined/selected from 

83 """ 

84 clauses = [table.is_visible, _shadow_clause(context, table)] 

85 if context.is_logged_in(): 

86 clauses.append(~table.id.in_(_relevant_user_blocks(context.user_id))) 

87 return and_(*clauses) 

88 

89 

90def where_users_column_visible[T: tuple[Any, ...]]( 

91 query: Select[T], context: CouchersContext, column: InstrumentedAttribute[int] 

92) -> Select[T]: 

93 """ 

94 Filters the given column, not yet joined/selected from 

95 """ 

96 # EXISTS on the bare User table rather than aliased(User): aliasing the wide User entity per call 

97 # is a major CPU hotspot (rebuilds the ORM proxy index). correlate_except keeps this User local to 

98 # the subquery so it doesn't bind to a bare User already in the outer query (e.g. jobs/handlers.py). 

99 return query.where( 

100 exists( 

101 select(1) 

102 .select_from(User) 

103 .where(User.id == column) 

104 .where(users_visible(context, User)) 

105 .correlate_except(User) 

106 ) 

107 ) 

108 

109 

110def users_visible_to_each_other(*, self_user: _User, other_user: _User) -> ColumnElement[bool]: 

111 """ 

112 Filters to ensure other_user is visible to self_user, and that they haven't blocked each other. 

113 

114 Use this when both User tables are already joined/selected in the query. 

115 """ 

116 return and_( 

117 self_user.is_visible, 

118 other_user.is_visible, 

119 other_user.shadowed_at.is_(None), 

120 ~_users_block_each_other(self_user.id, other_user.id), 

121 ) 

122 

123 

124def where_user_columns_visible_to_each_other[T: tuple[Any, ...]]( 

125 query: Select[T], *, self_column: InstrumentedAttribute[int], other_column: InstrumentedAttribute[int] 

126) -> Select[T]: 

127 """ 

128 Filters to ensure the user in other_column is visible to the user in self_column, and that they 

129 haven't blocked each other. 

130 

131 Use this when you have two user_id columns that haven't been joined yet. This will join both 

132 User tables and apply the visibility checks. 

133 """ 

134 self_user = aliased(User) 

135 other_user = aliased(User) 

136 return ( 

137 query.join(self_user, self_user.id == self_column) 

138 .join(other_user, other_user.id == other_column) 

139 .where(self_user.is_visible) 

140 .where(other_user.is_visible) 

141 .where(other_user.shadowed_at.is_(None)) 

142 .where(~_users_block_each_other(self_user.id, other_user.id)) 

143 ) 

144 

145 

146def where_moderated_content_visible_to_user_column[T: tuple[Any, ...]]( 

147 query: Select[T], 

148 table: type[ModeratedContent], 

149 user_id_column: InstrumentedAttribute[int], 

150 is_list_operation: bool = False, 

151) -> Select[T]: 

152 entry = get_moderated_models()[table.__moderation_object_type__] 

153 aliased_mod_state = aliased(ModerationState) 

154 conditions = [aliased_mod_state.visibility == ModerationVisibility.visible] 

155 

156 # UNLISTED content is visible in single-item operations but not in lists 

157 if not is_list_operation: 157 ↛ 161line 157 didn't jump to line 161 because the condition on line 157 was always true

158 conditions.append(aliased_mod_state.visibility == ModerationVisibility.unlisted) 

159 

160 # Authors can always see their own SHADOWED content 

161 conditions.append( 

162 and_( 

163 aliased_mod_state.visibility == ModerationVisibility.shadowed, 

164 entry.author_column == user_id_column, 

165 ) 

166 ) 

167 

168 return query.join(aliased_mod_state, aliased_mod_state.id == entry.moderation_state_id_column).where( 

169 or_(*conditions) 

170 ) 

171 

172 

173def where_moderated_content_visible[T: tuple[Any, ...]]( 

174 query: Select[T], 

175 context: CouchersContext, 

176 table: type[ModeratedContent], 

177 is_list_operation: bool = False, 

178) -> Select[T]: 

179 entry = get_moderated_models()[table.__moderation_object_type__] 

180 aliased_mod_state = aliased(ModerationState) 

181 conditions = [aliased_mod_state.visibility == ModerationVisibility.visible] 

182 

183 # UNLISTED content is visible in single-item operations but not in lists 

184 if not is_list_operation: 

185 conditions.append(aliased_mod_state.visibility == ModerationVisibility.unlisted) 

186 

187 # Authors can always see their own SHADOWED content 

188 if context.is_logged_in(): 

189 conditions.append( 

190 and_( 

191 aliased_mod_state.visibility == ModerationVisibility.shadowed, 

192 entry.author_column == context.user_id, 

193 ) 

194 ) 

195 

196 return query.join(aliased_mod_state, aliased_mod_state.id == entry.moderation_state_id_column).where( 

197 or_(*conditions) 

198 ) 

199 

200 

201def moderation_state_column_visible( 

202 context: CouchersContext, 

203 column: InstrumentedAttribute[int | None], 

204) -> ColumnElement[bool]: 

205 """ 

206 Filters based on whether the moderation state referenced by the column is visible. 

207 

208 Use this when you have a moderation_state_id column on a table that's not the moderated 

209 content itself (e.g., Notification.moderation_state_id). 

210 

211 The condition evaluates to True when: 

212 - The column is NULL (non-moderated content), OR 

213 - The linked moderation state has visibility 'visible' or 'unlisted', OR 

214 - The linked moderation state has visibility 'shadowed' and the current user is the author 

215 """ 

216 aliased_mod_state = aliased(ModerationState) 

217 

218 # For 'shadowed' content, look up the moderated content via object_type/object_id to check the author 

219 shadowed_conditions: list[ColumnElement[bool]] = [] 

220 if context.is_logged_in(): 220 ↛ 234line 220 didn't jump to line 234 because the condition on line 220 was always true

221 for entry in get_moderated_models().values(): 

222 shadowed_conditions.append( 

223 and_( 

224 aliased_mod_state.object_type == entry.object_type, 

225 exists( 

226 select(1) 

227 .select_from(entry.model) 

228 .where(entry.object_id_column == aliased_mod_state.object_id) 

229 .where(entry.author_column == context.user_id) 

230 ), 

231 ) 

232 ) 

233 

234 return or_( 

235 column.is_(None), 

236 exists( 

237 select(aliased_mod_state.id).where( 

238 aliased_mod_state.id == column, 

239 or_( 

240 aliased_mod_state.visibility == ModerationVisibility.visible, 

241 aliased_mod_state.visibility == ModerationVisibility.unlisted, 

242 *shadowed_conditions, 

243 ), 

244 ) 

245 ), 

246 ) 

247 

248 

249def _relevant_user_blocks(user_id: int) -> Select[tuple[int]]: 

250 """ 

251 Gets a list of blocked user IDs or users that have blocked this user: those should be hidden 

252 """ 

253 blocked_users = select(UserBlock.blocked_user_id).where(UserBlock.blocking_user_id == user_id) 

254 blocking_users = select(UserBlock.blocking_user_id).where(UserBlock.blocked_user_id == user_id) 

255 

256 return select(union(blocked_users, blocking_users).subquery()) 

257 

258 

259def to_bool(value: bool) -> ColumnElement[bool]: 

260 return true() if value else false()