Coverage for src/couchers/servicers/galleries.py: 96%

1import grpc 

2from sqlalchemy.orm import Session 

3from sqlalchemy.sql import func 

4 

5from couchers.constants import GALLERY_MAX_PHOTOS_NOT_VERIFIED, GALLERY_MAX_PHOTOS_VERIFIED 

6from couchers.helpers.strong_verification import has_strong_verification 

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

8from couchers.proto import galleries_pb2, galleries_pb2_grpc 

9from couchers.sql import couchers_select as select 

10 

11 

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

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

14 

15 

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

17 return gallery.owner_user_id == context.user_id 

18 

19 

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

21 return galleries_pb2.PhotoGallery( 

22 gallery_id=gallery.id, 

23 photos=[ 

24 galleries_pb2.PhotoGalleryItem( 

25 item_id=item.id, 

26 full_url=item.upload.full_url, 

27 thumbnail_url=item.upload.thumbnail_url, 

28 caption=item.caption, 

29 ) 

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

31 ], 

32 can_edit=_can_edit_gallery(gallery, context), 

33 ) 

34 

35 

36class Galleries(galleries_pb2_grpc.GalleriesServicer): 

37 def GetGallery(self, request: galleries_pb2.GetGalleryReq, context, session: Session) -> galleries_pb2.PhotoGallery: 

38 gallery = session.execute( 

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

40 ).scalar_one_or_none() 

41 

42 if not gallery: 

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

44 

45 return _gallery_to_pb(gallery, context) 

46 

47 def AddPhotoToGallery( 

48 self, request: galleries_pb2.AddPhotoToGalleryReq, context, session: Session 

49 ) -> galleries_pb2.PhotoGallery: 

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

51 

52 gallery = session.execute( 

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

54 ).scalar_one_or_none() 

55 

56 if not gallery: 

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

58 

59 if not _can_edit_gallery(gallery, context): 

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

61 

62 upload_exists = session.execute( 

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

64 ).scalar_one_or_none() 

65 

66 if not upload_exists: 

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

68 

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

70 gallery_items = session.execute( 

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

72 PhotoGalleryItem.gallery_id == gallery.id 

73 ) 

74 ).all() 

75 

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

77 

78 if request.upload_key in existing_upload_keys: 

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

80 

81 max_photos = _get_max_photos_for_user(session, user) 

82 

83 if len(gallery_items) >= max_photos: 

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

85 

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

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

88 

89 item = PhotoGalleryItem( 

90 gallery_id=gallery.id, 

91 upload_key=request.upload_key, 

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

93 caption=request.caption or None, 

94 ) 

95 session.add(item) 

96 session.flush() 

97 session.refresh(gallery) 

98 

99 return _gallery_to_pb(gallery, context) 

100 

101 def RemovePhotoFromGallery( 

102 self, request: galleries_pb2.RemovePhotoFromGalleryReq, context, session: Session 

103 ) -> galleries_pb2.PhotoGallery: 

104 gallery = session.execute( 

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

106 ).scalar_one_or_none() 

107 

108 if not gallery: 

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

110 

111 if not _can_edit_gallery(gallery, context): 

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

113 

114 item = session.execute( 

115 select(PhotoGalleryItem) 

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

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

118 ).scalar_one_or_none() 

119 

120 if not item: 

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

122 

123 session.delete(item) 

124 session.flush() 

125 session.refresh(gallery) 

126 

127 return _gallery_to_pb(gallery, context) 

128 

129 def MovePhoto(self, request: galleries_pb2.MovePhotoReq, context, session: Session) -> galleries_pb2.PhotoGallery: 

130 gallery = session.execute( 

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

132 ).scalar_one_or_none() 

133 

134 if not gallery: 

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

136 

137 if not _can_edit_gallery(gallery, context): 

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

139 

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

141 items = session.execute( 

142 select(PhotoGalleryItem.id, PhotoGalleryItem.position) 

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

144 .order_by(PhotoGalleryItem.position) 

145 ).all() 

146 

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

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

149 

150 if request.item_id not in positions: 

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

152 

153 if request.after_item_id == request.item_id: 

154 # Moving after itself is a no-op 

155 return _gallery_to_pb(gallery, context) 

156 

157 if request.after_item_id == 0: 

158 # Move to first position 

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

160 else: 

161 if request.after_item_id not in positions: 

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

163 

164 after_idx = sorted_ids.index(request.after_item_id) 

165 next_idx = after_idx + 1 

166 

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

168 if next_idx < len(sorted_ids) and sorted_ids[next_idx] == request.item_id: 

169 next_idx += 1 

170 

171 if next_idx >= len(sorted_ids): 

172 # Place at end 

173 new_position = positions[request.after_item_id] + 1.0 

174 else: 

175 # Place between after_item and next_item 

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

177 

178 if new_position is not None: 

179 session.execute( 

180 PhotoGalleryItem.__table__.update() 

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

182 .values(position=new_position) 

183 ) 

184 

185 session.flush() 

186 session.refresh(gallery) 

187 

188 return _gallery_to_pb(gallery, context) 

189 

190 def UpdatePhotoCaption( 

191 self, request: galleries_pb2.UpdatePhotoCaptionReq, context, session: Session 

192 ) -> galleries_pb2.PhotoGallery: 

193 gallery = session.execute( 

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

195 ).scalar_one_or_none() 

196 

197 if not gallery: 

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

199 

200 if not _can_edit_gallery(gallery, context): 

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

202 

203 item = session.execute( 

204 select(PhotoGalleryItem) 

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

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

207 ).scalar_one_or_none() 

208 

209 if not item: 

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

211 

212 item.caption = request.caption or None 

213 session.flush() 

214 session.refresh(gallery) 

215 

216 return _gallery_to_pb(gallery, context) 

217 

218 def GetGalleryEditInfo( 

219 self, request: galleries_pb2.GetGalleryEditInfoReq, context, session: Session 

220 ) -> galleries_pb2.GetGalleryEditInfoRes: 

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

222 

223 gallery = session.execute( 

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

225 ).scalar_one_or_none() 

226 

227 if not gallery: 

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

229 

230 if not _can_edit_gallery(gallery, context): 

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

232 

233 current_photo_count = session.execute( 

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

235 ).scalar_one() 

236 

237 return galleries_pb2.GetGalleryEditInfoRes( 

238 gallery_id=gallery.id, 

239 max_photos=_get_max_photos_for_user(session, user), 

240 current_photo_count=current_photo_count, 

241 )