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

1from datetime import datetime 

2from typing import TYPE_CHECKING, Any 

3 

4from sqlalchemy import ( 

5 BigInteger, 

6 DateTime, 

7 Float, 

8 ForeignKey, 

9 Integer, 

10 String, 

11 UniqueConstraint, 

12 exists, 

13 func, 

14 literal, 

15 select, 

16) 

17from sqlalchemy import LargeBinary as Binary 

18from sqlalchemy.dialects.postgresql import JSONB 

19from sqlalchemy.ext.hybrid import hybrid_property 

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

21from sqlalchemy.sql.elements import ColumnElement 

22from sqlalchemy.sql.selectable import Subquery 

23 

24from couchers import urls 

25from couchers.models.base import Base 

26 

27if TYPE_CHECKING: 

28 from couchers.models.users import User 

29 

30 

31class InitiatedUpload(Base, kw_only=True): 

32 """ 

33 Started downloads, not necessarily complete yet. 

34 """ 

35 

36 __tablename__ = "initiated_uploads" 

37 

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

39 

40 # timezones should always be UTC 

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

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

43 

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

45 

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

47 

48 @hybrid_property 

49 def is_valid(self) -> Any: 

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

51 

52 

53class Upload(Base, kw_only=True): 

54 """ 

55 Completed uploads. 

56 

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

58 couchers/helpers/upload_uses.py. 

59 """ 

60 

61 __tablename__ = "uploads" 

62 

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

64 

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

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

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

68 

69 # photo credit, etc 

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

71 

72 metadata_exif: Mapped[bytes | None] = mapped_column(Binary, deferred=True, default=None) 

73 metadata_xmp: Mapped[bytes | None] = mapped_column(Binary, deferred=True, default=None) 

74 metadata_iptc: Mapped[bytes | None] = mapped_column(Binary, deferred=True, default=None) 

75 metadata_parsed: Mapped[Any | None] = mapped_column(JSONB(none_as_null=True), deferred=True, default=None) 

76 

77 metadata_parse_error: Mapped[str | None] = mapped_column(String, deferred=True, default=None) 

78 

79 original_filename: Mapped[str | None] = mapped_column(String, deferred=True, default=None) 

80 original_format: Mapped[str | None] = mapped_column(String, deferred=True, default=None) 

81 original_size: Mapped[int | None] = mapped_column(BigInteger, deferred=True, default=None) 

82 original_width: Mapped[int | None] = mapped_column(Integer, deferred=True, default=None) 

83 original_height: Mapped[int | None] = mapped_column(Integer, deferred=True, default=None) 

84 

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

86 

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

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

89 

90 @property 

91 def thumbnail_url(self) -> str: 

92 return self._url("thumbnail") 

93 

94 @property 

95 def full_url(self) -> str: 

96 return self._url("full") 

97 

98 

99class PhotoGallery(Base, kw_only=True): 

100 """ 

101 Photo galleries for users or other entities. 

102 """ 

103 

104 __tablename__ = "photo_galleries" 

105 

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

107 

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

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

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

111 

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

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

114 

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

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

117 init=False, 

118 back_populates="gallery", 

119 order_by="PhotoGalleryItem.position", 

120 ) 

121 

122 

123class PhotoGalleryItem(Base, kw_only=True): 

124 """ 

125 Individual photos within a gallery with ordering and captions. 

126 """ 

127 

128 __tablename__ = "photo_gallery_items" 

129 

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

131 

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

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

134 

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

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

137 

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

139 

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

141 

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

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

144 

145 __table_args__ = ( 

146 # Ensure each upload is only in a gallery once 

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

148 ) 

149 

150 

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

152 """ 

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

154 

155 The subquery has columns: gallery_id, upload_key 

156 

157 Usage: 

158 avatar_photo = get_avatar_photo_subquery() 

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

160 """ 

161 return ( 

162 select( 

163 PhotoGalleryItem.gallery_id, 

164 PhotoGalleryItem.upload_key, 

165 ) 

166 .distinct(PhotoGalleryItem.gallery_id) 

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

168 .subquery(name=name) 

169 ) 

170 

171 

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

173 """ 

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

175 """ 

176 return session.execute( 

177 select(Upload) 

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

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

180 .order_by(PhotoGalleryItem.position) 

181 .limit(1) 

182 ).scalar_one_or_none() 

183 

184 

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

186 """ 

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

188 

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

190 

191 Usage: 

192 # In a query filter 

193 statement.where(has_avatar_photo_expression(User)) 

194 

195 # With a concrete value 

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

197 """ 

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