Coverage for app/backend/src/couchers/models/uploads.py: 100%

1from datetime import datetime 

2from typing import TYPE_CHECKING, Any 

3 

4from sqlalchemy import BigInteger, DateTime, Float, ForeignKey, String, UniqueConstraint, exists, func, literal, select 

5from sqlalchemy.ext.hybrid import hybrid_property 

6from sqlalchemy.orm import Mapped, Session, mapped_column, relationship 

7from sqlalchemy.sql.elements import ColumnElement 

8from sqlalchemy.sql.selectable import Subquery 

9 

10from couchers import urls 

11from couchers.models.base import Base 

12 

13if TYPE_CHECKING: 

14 from couchers.models.users import User 

15 

16 

17class InitiatedUpload(Base, kw_only=True): 

18 """ 

19 Started downloads, not necessarily complete yet. 

20 """ 

21 

22 __tablename__ = "initiated_uploads" 

23 

24 key: Mapped[str] = mapped_column(String, primary_key=True) 

25 

26 # timezones should always be UTC 

27 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) 

28 expiry: Mapped[datetime] = mapped_column(DateTime(timezone=True)) 

29 

30 initiator_user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

31 

32 initiator_user: Mapped[User] = relationship(init=False) 

33 

34 @hybrid_property 

35 def is_valid(self) -> Any: 

36 return (self.created <= func.now()) & (self.expiry >= func.now()) 

37 

38 

39class Upload(Base, kw_only=True): 

40 """ 

41 Completed uploads. 

42 

43 When adding a new foreign key to uploads.key, also update the reverse lookup in 

44 couchers/helpers/upload_uses.py. 

45 """ 

46 

47 __tablename__ = "uploads" 

48 

49 key: Mapped[str] = mapped_column(String, primary_key=True) 

50 

51 filename: Mapped[str] = mapped_column(String) 

52 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

53 creator_user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

54 

55 # photo credit, etc 

56 credit: Mapped[str | None] = mapped_column(String, default=None) 

57 

58 creator_user: Mapped[User] = relationship(init=False, backref="uploads", foreign_keys="Upload.creator_user_id") 

59 

60 def _url(self, size: str) -> str: 

61 return urls.media_url(filename=self.filename, size=size) 

62 

63 @property 

64 def thumbnail_url(self) -> str: 

65 return self._url("thumbnail") 

66 

67 @property 

68 def full_url(self) -> str: 

69 return self._url("full") 

70 

71 

72class PhotoGallery(Base, kw_only=True): 

73 """ 

74 Photo galleries for users or other entities. 

75 """ 

76 

77 __tablename__ = "photo_galleries" 

78 

79 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

80 

81 # For now, galleries are owned by users, but this could be extended 

82 # in the future for communities, events, etc. 

83 owner_user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

84 

85 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

86 last_updated: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

87 

88 owner_user: Mapped[User] = relationship(init=False, foreign_keys=[owner_user_id], back_populates="galleries") 

89 photos: Mapped[list[PhotoGalleryItem]] = relationship( 

90 init=False, 

91 back_populates="gallery", 

92 order_by="PhotoGalleryItem.position", 

93 ) 

94 

95 

96class PhotoGalleryItem(Base, kw_only=True): 

97 """ 

98 Individual photos within a gallery with ordering and captions. 

99 """ 

100 

101 __tablename__ = "photo_gallery_items" 

102 

103 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

104 

105 gallery_id: Mapped[int] = mapped_column(ForeignKey("photo_galleries.id"), index=True) 

106 upload_key: Mapped[str] = mapped_column(ForeignKey("uploads.key")) 

107 

108 # Float position for ordering - allows inserting between items without shifting 

109 position: Mapped[float] = mapped_column(Float) 

110 

111 caption: Mapped[str | None] = mapped_column(String, default=None) 

112 

113 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

114 

115 gallery: Mapped[PhotoGallery] = relationship(init=False, back_populates="photos") 

116 upload: Mapped[Upload] = relationship(init=False) 

117 

118 __table_args__ = ( 

119 # Ensure each upload is only in a gallery once 

120 UniqueConstraint("gallery_id", "upload_key", name="uix_gallery_upload"), 

121 ) 

122 

123 

124def get_avatar_photo_subquery(name: str = "avatar_photo") -> Subquery: 

125 """ 

126 Returns a subquery that selects the first photo (by position) from each photo gallery. 

127 

128 The subquery has columns: gallery_id, upload_key 

129 

130 Usage: 

131 avatar_photo = get_avatar_photo_subquery() 

132 query = select(User).outerjoin(avatar_photo, avatar_photo.c.gallery_id == User.profile_gallery_id) 

133 """ 

134 return ( 

135 select( 

136 PhotoGalleryItem.gallery_id, 

137 PhotoGalleryItem.upload_key, 

138 ) 

139 .distinct(PhotoGalleryItem.gallery_id) 

140 .order_by(PhotoGalleryItem.gallery_id, PhotoGalleryItem.position) 

141 .subquery(name=name) 

142 ) 

143 

144 

145def get_avatar_upload(session: Session, user: User) -> Upload | None: 

146 """ 

147 Returns the Upload for the user's avatar (first photo in their profile gallery), or None. 

148 """ 

149 return session.execute( 

150 select(Upload) 

151 .join(PhotoGalleryItem, PhotoGalleryItem.upload_key == Upload.key) 

152 .where(PhotoGalleryItem.gallery_id == user.profile_gallery_id) 

153 .order_by(PhotoGalleryItem.position) 

154 .limit(1) 

155 ).scalar_one_or_none() 

156 

157 

158def has_avatar_photo_expression(user: type[User] | User) -> ColumnElement[bool]: 

159 """ 

160 Returns an EXISTS expression that checks if a user has at least one photo in their profile gallery. 

161 

162 Can be used with a User instance or the User class (for SQL expressions). 

163 

164 Usage: 

165 # In a query filter 

166 statement.where(has_avatar_photo_expression(User)) 

167 

168 # With a concrete value 

169 session.execute(select(has_avatar_photo_expression(user))).scalar() 

170 """ 

171 return exists(select(literal(1)).where(PhotoGalleryItem.gallery_id == user.profile_gallery_id))