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 

44 __tablename__ = "uploads" 

45 

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

47 

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

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

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

51 

52 # photo credit, etc 

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

54 

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

56 

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

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

59 

60 @property 

61 def thumbnail_url(self) -> str: 

62 return self._url("thumbnail") 

63 

64 @property 

65 def full_url(self) -> str: 

66 return self._url("full") 

67 

68 

69class PhotoGallery(Base, kw_only=True): 

70 """ 

71 Photo galleries for users or other entities. 

72 """ 

73 

74 __tablename__ = "photo_galleries" 

75 

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

77 

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

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

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

81 

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

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

84 

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

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

87 init=False, 

88 back_populates="gallery", 

89 order_by="PhotoGalleryItem.position", 

90 ) 

91 

92 

93class PhotoGalleryItem(Base, kw_only=True): 

94 """ 

95 Individual photos within a gallery with ordering and captions. 

96 """ 

97 

98 __tablename__ = "photo_gallery_items" 

99 

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

101 

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

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

104 

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

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

107 

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

109 

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

111 

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

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

114 

115 __table_args__ = ( 

116 # Ensure each upload is only in a gallery once 

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

118 ) 

119 

120 

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

122 """ 

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

124 

125 The subquery has columns: gallery_id, upload_key 

126 

127 Usage: 

128 avatar_photo = get_avatar_photo_subquery() 

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

130 """ 

131 return ( 

132 select( 

133 PhotoGalleryItem.gallery_id, 

134 PhotoGalleryItem.upload_key, 

135 ) 

136 .distinct(PhotoGalleryItem.gallery_id) 

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

138 .subquery(name=name) 

139 ) 

140 

141 

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

143 """ 

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

145 """ 

146 return session.execute( 

147 select(Upload) 

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

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

150 .order_by(PhotoGalleryItem.position) 

151 .limit(1) 

152 ).scalar_one_or_none() 

153 

154 

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

156 """ 

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

158 

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

160 

161 Usage: 

162 # In a query filter 

163 statement.where(has_avatar_photo_expression(User)) 

164 

165 # With a concrete value 

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

167 """ 

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