Coverage for app/backend/src/couchers/servicers/galleries.py: 94%

1import grpc 

2from sqlalchemy import select, update 

3from sqlalchemy.orm import Session 

4from sqlalchemy.sql import func 

5 

6from couchers.constants import GALLERY_MAX_PHOTOS_NOT_VERIFIED, GALLERY_MAX_PHOTOS_VERIFIED 

7from couchers.context import CouchersContext 

8from couchers.helpers.strong_verification import has_strong_verification 

9from couchers.models import PhotoGallery, PhotoGalleryItem, Upload, User 

10from couchers.proto import galleries_pb2, galleries_pb2_grpc 

11 

12 

13def _get_max_photos_for_user(session: Session, user: User) -> int: 

14 return GALLERY_MAX_PHOTOS_VERIFIED if has_strong_verification(session, user) else GALLERY_MAX_PHOTOS_NOT_VERIFIED 

15 

16 

17def _can_edit_gallery(gallery: PhotoGallery, context: CouchersContext) -> bool: 

18 return gallery.owner_user_id == context.user_id 

19 

20 

21def _gallery_to_pb(gallery: PhotoGallery, context: CouchersContext) -> galleries_pb2.PhotoGallery: 

22 return galleries_pb2.PhotoGallery( 

23 gallery_id=gallery.id, 

24 photos=[ 

25 galleries_pb2.PhotoGalleryItem( 

26 item_id=item.id, 

27 full_url=item.upload.full_url, 

28 thumbnail_url=item.upload.thumbnail_url, 

29 caption=item.caption, 

30 ) 

31 for item in gallery.photos # Already ordered by position via relationship 

32 ], 

33 can_edit=_can_edit_gallery(gallery, context), 

34 ) 

35 

36 

37class Galleries(galleries_pb2_grpc.GalleriesServicer): 

38 def GetGallery( 

39 self, request: galleries_pb2.GetGalleryReq, context: CouchersContext, session: Session 

40 ) -> galleries_pb2.PhotoGallery: 

41 gallery = session.execute( 

42 select(PhotoGallery).where(PhotoGallery.id == request.gallery_id) 

43 ).scalar_one_or_none() 

44 

45 if not gallery: 

46 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_not_found") 

47 

48 return _gallery_to_pb(gallery, context) 

49 

50 def AddPhotoToGallery( 

51 self, request: galleries_pb2.AddPhotoToGalleryReq, context: CouchersContext, session: Session 

52 ) -> galleries_pb2.PhotoGallery: 

53 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

54 

55 gallery = session.execute( 

56 select(PhotoGallery).where(PhotoGallery.id == request.gallery_id) 

57 ).scalar_one_or_none() 

58 

59 if not gallery: 

60 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_not_found") 

61 

62 if not _can_edit_gallery(gallery, context): 

63 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "permission_denied_to_edit_gallery") 

64 

65 upload_exists = session.execute( 

66 select(Upload.key).where(Upload.key == request.upload_key).where(Upload.creator_user_id == user.id) 

67 ).scalar_one_or_none() 

68 

69 if not upload_exists: 

70 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "upload_not_found_or_not_owned") 

71 

72 # Get all gallery items' upload keys and positions in one query 

73 gallery_items = session.execute( 

74 select(PhotoGalleryItem.upload_key, PhotoGalleryItem.position).where( 

75 PhotoGalleryItem.gallery_id == gallery.id 

76 ) 

77 ).all() 

78 

79 existing_upload_keys = {item.upload_key for item in gallery_items} 

80 

81 if request.upload_key in existing_upload_keys: 

82 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "photo_already_in_gallery") 

83 

84 max_photos = _get_max_photos_for_user(session, user) 

85 

86 if len(gallery_items) >= max_photos: 

87 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "gallery_at_max_capacity") 

88 

89 # Get max position and add 1.0 (or start at 0.0 if empty) 

90 max_position = max((item.position for item in gallery_items), default=None) 

91 

92 item = PhotoGalleryItem( 

93 gallery_id=gallery.id, 

94 upload_key=request.upload_key, 

95 position=0.0 if max_position is None else max_position + 1.0, 

96 caption=request.caption or None, 

97 ) 

98 session.add(item) 

99 gallery.last_updated = func.now() 

100 session.flush() 

101 session.refresh(gallery) 

102 

103 return _gallery_to_pb(gallery, context) 

104 

105 def RemovePhotoFromGallery( 

106 self, request: galleries_pb2.RemovePhotoFromGalleryReq, context: CouchersContext, session: Session 

107 ) -> galleries_pb2.PhotoGallery: 

108 gallery = session.execute( 

109 select(PhotoGallery).where(PhotoGallery.id == request.gallery_id) 

110 ).scalar_one_or_none() 

111 

112 if not gallery: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true

113 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_not_found") 

114 

115 if not _can_edit_gallery(gallery, context): 

116 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "permission_denied_to_edit_gallery") 

117 

118 item = session.execute( 

119 select(PhotoGalleryItem) 

120 .where(PhotoGalleryItem.gallery_id == gallery.id) 

121 .where(PhotoGalleryItem.id == request.item_id) 

122 ).scalar_one_or_none() 

123 

124 if not item: 

125 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_item_not_found") 

126 

127 session.delete(item) 

128 gallery.last_updated = func.now() 

129 session.flush() 

130 session.refresh(gallery) 

131 

132 return _gallery_to_pb(gallery, context) 

133 

134 def MovePhoto( 

135 self, request: galleries_pb2.MovePhotoReq, context: CouchersContext, session: Session 

136 ) -> galleries_pb2.PhotoGallery: 

137 gallery = session.execute( 

138 select(PhotoGallery).where(PhotoGallery.id == request.gallery_id) 

139 ).scalar_one_or_none() 

140 

141 if not gallery: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_not_found") 

143 

144 if not _can_edit_gallery(gallery, context): 

145 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "permission_denied_to_edit_gallery") 

146 

147 # Get all items' (id, position) sorted by position 

148 items = session.execute( 

149 select(PhotoGalleryItem.id, PhotoGalleryItem.position) 

150 .where(PhotoGalleryItem.gallery_id == gallery.id) 

151 .order_by(PhotoGalleryItem.position) 

152 ).all() 

153 

154 positions = {item.id: item.position for item in items} 

155 sorted_ids = [item.id for item in items] 

156 

157 if request.item_id not in positions: 

158 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_item_not_found") 

159 

160 if request.after_item_id == request.item_id: 

161 # Moving after itself is a no-op 

162 return _gallery_to_pb(gallery, context) 

163 

164 if request.after_item_id == 0: 

165 # Move to first position 

166 new_position = positions[sorted_ids[0]] - 1.0 if sorted_ids[0] != request.item_id else None 

167 else: 

168 if request.after_item_id not in positions: 

169 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "after_item_not_found") 

170 

171 after_idx = sorted_ids.index(request.after_item_id) 

172 next_idx = after_idx + 1 

173 

174 # Skip over the item being moved if it's the next one 

175 if next_idx < len(sorted_ids) and sorted_ids[next_idx] == request.item_id: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true

176 next_idx += 1 

177 

178 if next_idx >= len(sorted_ids): 

179 # Place at end 

180 new_position = positions[request.after_item_id] + 1.0 

181 else: 

182 # Place between after_item and next_item 

183 new_position = (positions[request.after_item_id] + positions[sorted_ids[next_idx]]) / 2.0 

184 

185 if new_position is not None: 185 ↛ 191line 185 didn't jump to line 191 because the condition on line 185 was always true

186 session.execute( 

187 update(PhotoGalleryItem).where(PhotoGalleryItem.id == request.item_id).values(position=new_position) 

188 ) 

189 gallery.last_updated = func.now() 

190 

191 session.flush() 

192 session.refresh(gallery) 

193 

194 return _gallery_to_pb(gallery, context) 

195 

196 def UpdatePhotoCaption( 

197 self, request: galleries_pb2.UpdatePhotoCaptionReq, context: CouchersContext, session: Session 

198 ) -> galleries_pb2.PhotoGallery: 

199 gallery = session.execute( 

200 select(PhotoGallery).where(PhotoGallery.id == request.gallery_id) 

201 ).scalar_one_or_none() 

202 

203 if not gallery: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true

204 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_not_found") 

205 

206 if not _can_edit_gallery(gallery, context): 

207 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "permission_denied_to_edit_gallery") 

208 

209 item = session.execute( 

210 select(PhotoGalleryItem) 

211 .where(PhotoGalleryItem.gallery_id == gallery.id) 

212 .where(PhotoGalleryItem.id == request.item_id) 

213 ).scalar_one_or_none() 

214 

215 if not item: 

216 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_item_not_found") 

217 

218 item.caption = request.caption or None 

219 gallery.last_updated = func.now() 

220 session.flush() 

221 session.refresh(gallery) 

222 

223 return _gallery_to_pb(gallery, context) 

224 

225 def GetGalleryEditInfo( 

226 self, request: galleries_pb2.GetGalleryEditInfoReq, context: CouchersContext, session: Session 

227 ) -> galleries_pb2.GetGalleryEditInfoRes: 

228 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

229 

230 gallery = session.execute( 

231 select(PhotoGallery).where(PhotoGallery.id == request.gallery_id) 

232 ).scalar_one_or_none() 

233 

234 if not gallery: 

235 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "gallery_not_found") 

236 

237 if not _can_edit_gallery(gallery, context): 

238 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "permission_denied_to_edit_gallery") 

239 

240 current_photo_count = session.execute( 

241 select(func.count()).where(PhotoGalleryItem.gallery_id == gallery.id) 

242 ).scalar_one() 

243 

244 return galleries_pb2.GetGalleryEditInfoRes( 

245 gallery_id=gallery.id, 

246 max_photos=_get_max_photos_for_user(session, user), 

247 current_photo_count=current_photo_count, 

248 )