Coverage for app/backend/src/couchers/helpers/upload_uses.py: 100%

1import enum 

2from collections import defaultdict 

3from dataclasses import dataclass 

4 

5from sqlalchemy import select 

6from sqlalchemy.orm import Session 

7 

8from couchers import urls 

9from couchers.models import ( 

10 Event, 

11 EventOccurrence, 

12 Page, 

13 PageType, 

14 PageVersion, 

15 PhotoGallery, 

16 PhotoGalleryItem, 

17 User, 

18 get_avatar_photo_subquery, 

19) 

20 

21 

22class UploadUseType(enum.Enum): 

23 profile_gallery_photo = enum.auto() 

24 profile_gallery_photo_avatar = enum.auto() 

25 event = enum.auto() 

26 page = enum.auto() 

27 

28 

29@dataclass 

30class UploadUse: 

31 use_type: UploadUseType 

32 # whether this use is the one currently shown (vs. a superseded page version or deleted occurrence) 

33 is_current: bool 

34 # only the identifier relevant to use_type is set 

35 user_id: int | None = None 

36 event_id: int | None = None 

37 page_id: int | None = None 

38 # link to the use (user profile, event, community page); not set for places/guides (no frontend route yet) 

39 url: str | None = None 

40 

41 

42def get_upload_uses(session: Session, key: str) -> list[UploadUse]: 

43 """ 

44 Returns every place a given upload (by key) is used across the platform. 

45 

46 This is a reverse lookup over all foreign keys to uploads.key. Any new reference to uploads.key 

47 must be handled here; test_upload_uses.py guards against drift. 

48 """ 

49 return get_upload_uses_for_keys(session, [key]).get(key, []) 

50 

51 

52def get_upload_uses_for_keys(session: Session, keys: list[str]) -> dict[str, list[UploadUse]]: 

53 """ 

54 Batched version of get_upload_uses: maps each given key to its list of uses. 

55 """ 

56 uses: dict[str, list[UploadUse]] = defaultdict(list) 

57 if not keys: 

58 return {} 

59 

60 # the first gallery photo by position is the user's avatar 

61 avatar_photo = get_avatar_photo_subquery() 

62 gallery_rows = session.execute( 

63 select( 

64 PhotoGalleryItem.upload_key, 

65 PhotoGallery.owner_user_id, 

66 User.username, 

67 avatar_photo.c.upload_key == PhotoGalleryItem.upload_key, 

68 ) 

69 .select_from(PhotoGalleryItem) 

70 .join(PhotoGallery, PhotoGalleryItem.gallery_id == PhotoGallery.id) 

71 .join(User, User.id == PhotoGallery.owner_user_id) 

72 .join(avatar_photo, avatar_photo.c.gallery_id == PhotoGallery.id) 

73 .where(PhotoGalleryItem.upload_key.in_(keys)) 

74 ).all() 

75 for upload_key, owner_user_id, username, is_avatar in gallery_rows: 

76 uses[upload_key].append( 

77 UploadUse( 

78 use_type=( 

79 UploadUseType.profile_gallery_photo_avatar if is_avatar else UploadUseType.profile_gallery_photo 

80 ), 

81 is_current=True, 

82 user_id=owner_user_id, 

83 url=urls.user_link(username=username), 

84 ) 

85 ) 

86 

87 # the occurrence id is what the rest of the API exposes as the "event id" 

88 event_rows = session.execute( 

89 select(EventOccurrence.photo_key, EventOccurrence.id, EventOccurrence.is_deleted, Event.slug) 

90 .join(Event, Event.id == EventOccurrence.event_id) 

91 .where(EventOccurrence.photo_key.in_(keys)) 

92 ).all() 

93 for photo_key, occurrence_id, is_deleted, slug in event_rows: 

94 uses[photo_key].append( 

95 UploadUse( 

96 use_type=UploadUseType.event, 

97 is_current=not is_deleted, 

98 event_id=occurrence_id, 

99 url=urls.event_link(occurrence_id=occurrence_id, slug=slug), 

100 ) 

101 ) 

102 

103 # the link points at the page as it stands now, so use the current version's slug 

104 current_version = ( 

105 select(PageVersion.page_id, PageVersion.id.label("current_id"), PageVersion.slug.label("current_slug")) 

106 .distinct(PageVersion.page_id) 

107 .order_by(PageVersion.page_id, PageVersion.id.desc()) 

108 .subquery() 

109 ) 

110 page_rows = session.execute( 

111 select( 

112 PageVersion.photo_key, 

113 Page.id, 

114 Page.type, 

115 Page.parent_node_id, 

116 current_version.c.current_slug, 

117 PageVersion.id == current_version.c.current_id, 

118 ) 

119 .select_from(PageVersion) 

120 .join(Page, Page.id == PageVersion.page_id) 

121 .join(current_version, current_version.c.page_id == PageVersion.page_id) 

122 .where(PageVersion.photo_key.in_(keys)) 

123 ).all() 

124 for photo_key, page_id, page_type, parent_node_id, current_slug, is_current in page_rows: 

125 # only community pages have a frontend route; places and guides aren't surfaced yet 

126 url = ( 

127 urls.community_link(node_id=parent_node_id, slug=current_slug) if page_type == PageType.main_page else None 

128 ) 

129 uses[photo_key].append(UploadUse(use_type=UploadUseType.page, is_current=is_current, page_id=page_id, url=url)) 

130 

131 return dict(uses)