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 

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

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

86 init=False, 

87 back_populates="gallery", 

88 order_by="PhotoGalleryItem.position", 

89 ) 

90 

91 

92class PhotoGalleryItem(Base, kw_only=True): 

93 """ 

94 Individual photos within a gallery with ordering and captions. 

95 """ 

96 

97 __tablename__ = "photo_gallery_items" 

98 

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

100 

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

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

103 

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

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

106 

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

108 

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

110 

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

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

113 

114 __table_args__ = ( 

115 # Ensure each upload is only in a gallery once 

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

117 ) 

118 

119 

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

121 """ 

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

123 

124 The subquery has columns: gallery_id, upload_key 

125 

126 Usage: 

127 avatar_photo = get_avatar_photo_subquery() 

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

129 """ 

130 return ( 

131 select( 

132 PhotoGalleryItem.gallery_id, 

133 PhotoGalleryItem.upload_key, 

134 ) 

135 .distinct(PhotoGalleryItem.gallery_id) 

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

137 .subquery(name=name) 

138 ) 

139 

140 

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

142 """ 

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

144 """ 

145 return session.execute( 

146 select(Upload) 

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

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

149 .order_by(PhotoGalleryItem.position) 

150 .limit(1) 

151 ).scalar_one_or_none() 

152 

153 

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

155 """ 

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

157 

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

159 

160 Usage: 

161 # In a query filter 

162 statement.where(has_avatar_photo_expression(User)) 

163 

164 # With a concrete value 

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

166 """ 

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