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 session.flush() 

100 session.refresh(gallery) 

101 

102 return _gallery_to_pb(gallery, context) 

103 

104 def RemovePhotoFromGallery( 

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

106 ) -> galleries_pb2.PhotoGallery: 

107 gallery = session.execute( 

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

109 ).scalar_one_or_none() 

110 

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

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

113 

114 if not _can_edit_gallery(gallery, context): 

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

116 

117 item = session.execute( 

118 select(PhotoGalleryItem) 

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

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

121 ).scalar_one_or_none() 

122 

123 if not item: 

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

125 

126 session.delete(item) 

127 session.flush() 

128 session.refresh(gallery) 

129 

130 return _gallery_to_pb(gallery, context) 

131 

132 def MovePhoto( 

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

134 ) -> galleries_pb2.PhotoGallery: 

135 gallery = session.execute( 

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

137 ).scalar_one_or_none() 

138 

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

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

141 

142 if not _can_edit_gallery(gallery, context): 

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

144 

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

146 items = session.execute( 

147 select(PhotoGalleryItem.id, PhotoGalleryItem.position) 

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

149 .order_by(PhotoGalleryItem.position) 

150 ).all() 

151 

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

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

154 

155 if request.item_id not in positions: 

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

157 

158 if request.after_item_id == request.item_id: 

159 # Moving after itself is a no-op 

160 return _gallery_to_pb(gallery, context) 

161 

162 if request.after_item_id == 0: 

163 # Move to first position 

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

165 else: 

166 if request.after_item_id not in positions: 

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

168 

169 after_idx = sorted_ids.index(request.after_item_id) 

170 next_idx = after_idx + 1 

171 

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

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

174 next_idx += 1 

175 

176 if next_idx >= len(sorted_ids): 

177 # Place at end 

178 new_position = positions[request.after_item_id] + 1.0 

179 else: 

180 # Place between after_item and next_item 

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

182 

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

184 session.execute( 

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

186 ) 

187 

188 session.flush() 

189 session.refresh(gallery) 

190 

191 return _gallery_to_pb(gallery, context) 

192 

193 def UpdatePhotoCaption( 

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

195 ) -> galleries_pb2.PhotoGallery: 

196 gallery = session.execute( 

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

198 ).scalar_one_or_none() 

199 

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

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

202 

203 if not _can_edit_gallery(gallery, context): 

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

205 

206 item = session.execute( 

207 select(PhotoGalleryItem) 

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

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

210 ).scalar_one_or_none() 

211 

212 if not item: 

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

214 

215 item.caption = request.caption or None 

216 session.flush() 

217 session.refresh(gallery) 

218 

219 return _gallery_to_pb(gallery, context) 

220 

221 def GetGalleryEditInfo( 

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

223 ) -> galleries_pb2.GetGalleryEditInfoRes: 

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

225 

226 gallery = session.execute( 

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

228 ).scalar_one_or_none() 

229 

230 if not gallery: 

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

232 

233 if not _can_edit_gallery(gallery, context): 

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

235 

236 current_photo_count = session.execute( 

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

238 ).scalar_one() 

239 

240 return galleries_pb2.GetGalleryEditInfoRes( 

241 gallery_id=gallery.id, 

242 max_photos=_get_max_photos_for_user(session, user), 

243 current_photo_count=current_photo_count, 

244 )