Coverage for src/couchers/models/events.py: 100%

1import enum 

2from datetime import datetime 

3from typing import Any 

4 

5from geoalchemy2 import Geometry 

6from psycopg2.extras import DateTimeTZRange 

7from sqlalchemy import ( 

8 BigInteger, 

9 Boolean, 

10 CheckConstraint, 

11 DateTime, 

12 Enum, 

13 ForeignKey, 

14 Index, 

15 String, 

16 UniqueConstraint, 

17 and_, 

18 func, 

19) 

20from sqlalchemy.dialects.postgresql import TSTZRANGE, ExcludeConstraint 

21from sqlalchemy.ext.hybrid import hybrid_property 

22from sqlalchemy.orm import Mapped, backref, column_property, mapped_column, relationship 

23from sqlalchemy.sql import expression 

24 

25from couchers.models.base import Base, Geom, communities_seq 

26from couchers.utils import get_coordinates 

27 

28 

29class ClusterEventAssociation(Base): 

30 """ 

31 events related to clusters 

32 """ 

33 

34 __tablename__ = "cluster_event_associations" 

35 __table_args__ = (UniqueConstraint("event_id", "cluster_id"),) 

36 

37 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

38 

39 event_id: Mapped[int] = mapped_column(ForeignKey("events.id"), index=True) 

40 cluster_id: Mapped[int] = mapped_column(ForeignKey("clusters.id"), index=True) 

41 

42 event = relationship("Event", backref="cluster_event_associations") 

43 cluster = relationship("Cluster", backref="cluster_event_associations") 

44 

45 

46class Event(Base): 

47 """ 

48 An event is composed of two parts: 

49 

50 * An event template (Event) 

51 * An occurrence (EventOccurrence) 

52 

53 One-off events will have one of each; repeating events will have one Event, 

54 multiple EventOccurrences, one for each time the event happens. 

55 """ 

56 

57 __tablename__ = "events" 

58 

59 id: Mapped[int] = mapped_column( 

60 BigInteger, communities_seq, primary_key=True, server_default=communities_seq.next_value() 

61 ) 

62 parent_node_id: Mapped[int] = mapped_column(ForeignKey("nodes.id"), index=True) 

63 

64 title: Mapped[str] = mapped_column(String) 

65 

66 slug = column_property(func.slugify(title)) 

67 

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

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

70 owner_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), nullable=True, index=True) 

71 owner_cluster_id: Mapped[int | None] = mapped_column(ForeignKey("clusters.id"), nullable=True, index=True) 

72 thread_id: Mapped[int] = mapped_column(ForeignKey("threads.id"), unique=True) 

73 

74 parent_node = relationship( 

75 "Node", backref="child_events", remote_side="Node.id", foreign_keys="Event.parent_node_id" 

76 ) 

77 thread = relationship("Thread", backref="event", uselist=False) 

78 subscribers = relationship( 

79 "User", backref="subscribed_events", secondary="event_subscriptions", lazy="dynamic", viewonly=True 

80 ) 

81 organizers = relationship( 

82 "User", backref="organized_events", secondary="event_organizers", lazy="dynamic", viewonly=True 

83 ) 

84 creator_user = relationship("User", backref="created_events", foreign_keys="Event.creator_user_id") 

85 owner_user = relationship("User", backref="owned_events", foreign_keys="Event.owner_user_id") 

86 owner_cluster = relationship( 

87 "Cluster", 

88 backref=backref("owned_events", lazy="dynamic"), 

89 uselist=False, 

90 foreign_keys="Event.owner_cluster_id", 

91 ) 

92 

93 __table_args__ = ( 

94 # Only one of owner_user and owner_cluster should be set 

95 CheckConstraint( 

96 "(owner_user_id IS NULL) <> (owner_cluster_id IS NULL)", 

97 name="one_owner", 

98 ), 

99 ) 

100 

101 

102class EventOccurrence(Base): 

103 __tablename__ = "event_occurrences" 

104 

105 id: Mapped[int] = mapped_column( 

106 BigInteger, communities_seq, primary_key=True, server_default=communities_seq.next_value() 

107 ) 

108 event_id: Mapped[int] = mapped_column(ForeignKey("events.id"), index=True) 

109 

110 # the user that created this particular occurrence of a repeating event (same as event.creator_user_id if single event) 

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

112 content: Mapped[str] = mapped_column(String) # CommonMark without images 

113 photo_key: Mapped[str | None] = mapped_column(ForeignKey("uploads.key"), nullable=True) 

114 

115 is_cancelled: Mapped[bool] = mapped_column(Boolean, default=False, server_default=expression.false()) 

116 is_deleted: Mapped[bool] = mapped_column(Boolean, default=False, server_default=expression.false()) 

117 

118 # a null geom is an online-only event 

119 geom: Mapped[Geom | None] = mapped_column(Geometry(geometry_type="POINT", srid=4326), nullable=True) 

120 # physical address, iff geom is not null 

121 address: Mapped[str | None] = mapped_column(String, nullable=True) 

122 # videoconferencing link, etc, must be specified if no geom, otherwise optional 

123 link: Mapped[str | None] = mapped_column(String, nullable=True) 

124 

125 timezone = "Etc/UTC" 

126 

127 # time during which the event takes place; this is a range type (instead of separate start+end times) which 

128 # simplifies database constraints, etc 

129 during: Mapped["DateTimeTZRange"] = mapped_column(TSTZRANGE) 

130 

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

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

133 

134 creator_user = relationship( 

135 "User", backref="created_event_occurrences", foreign_keys="EventOccurrence.creator_user_id" 

136 ) 

137 event = relationship( 

138 "Event", 

139 backref=backref("occurrences", lazy="dynamic"), 

140 remote_side="Event.id", 

141 foreign_keys="EventOccurrence.event_id", 

142 ) 

143 

144 photo = relationship("Upload") 

145 

146 __table_args__ = ( 

147 # Geom and address go together 

148 CheckConstraint( 

149 # geom and address are either both null or neither of them are null 

150 "(geom IS NULL) = (address IS NULL)", 

151 name="geom_iff_address", 

152 ), 

153 # Online-only events need a link, note that online events may also have a link 

154 CheckConstraint( 

155 # exactly oen of geom or link is non-null 

156 "(geom IS NULL) <> (link IS NULL)", 

157 name="link_or_geom", 

158 ), 

159 # Can't have overlapping occurrences in the same Event 

160 ExcludeConstraint(("event_id", "="), ("during", "&&"), name="event_occurrences_event_id_during_excl"), # type: ignore[no-untyped-call] 

161 ) 

162 

163 @property 

164 def coordinates(self) -> tuple[float, float] | None: 

165 # returns (lat, lng) or None 

166 return get_coordinates(self.geom) 

167 

168 @hybrid_property 

169 def start_time(self) -> Any: 

170 return self.during.lower 

171 

172 @start_time.expression 

173 def start_time(cls) -> Any: # noqa: ARG003,D102 

174 return func.lower(cls.during) 

175 

176 @hybrid_property 

177 def end_time(self) -> Any: 

178 return self.during.upper 

179 

180 @end_time.expression 

181 def end_time(cls) -> Any: # noqa: ARG003,D102 

182 return func.upper(cls.during) 

183 

184 

185class EventSubscription(Base): 

186 """ 

187 Users' subscriptions to events 

188 """ 

189 

190 __tablename__ = "event_subscriptions" 

191 __table_args__ = (UniqueConstraint("event_id", "user_id"),) 

192 

193 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

194 

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

196 event_id: Mapped[int] = mapped_column(ForeignKey("events.id"), index=True) 

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

198 

199 user = relationship("User") 

200 event = relationship("Event") 

201 

202 

203class EventOrganizer(Base): 

204 """ 

205 Organizers for events 

206 """ 

207 

208 __tablename__ = "event_organizers" 

209 __table_args__ = (UniqueConstraint("event_id", "user_id"),) 

210 

211 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

212 

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

214 event_id: Mapped[int] = mapped_column(ForeignKey("events.id"), index=True) 

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

216 

217 user = relationship("User") 

218 event = relationship("Event") 

219 

220 

221class AttendeeStatus(enum.Enum): 

222 going = enum.auto() 

223 maybe = enum.auto() 

224 

225 

226class EventOccurrenceAttendee(Base): 

227 """ 

228 Attendees for events 

229 """ 

230 

231 __tablename__ = "event_occurrence_attendees" 

232 __table_args__ = (UniqueConstraint("occurrence_id", "user_id"),) 

233 

234 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

235 

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

237 occurrence_id: Mapped[int] = mapped_column(ForeignKey("event_occurrences.id"), index=True) 

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

239 attendee_status: Mapped[AttendeeStatus] = mapped_column(Enum(AttendeeStatus)) 

240 

241 user = relationship("User") 

242 occurrence = relationship("EventOccurrence", backref=backref("attendances", lazy="dynamic")) 

243 

244 reminder_sent: Mapped[bool] = mapped_column(Boolean, default=False, server_default=expression.false()) 

245 

246 

247class EventCommunityInviteRequest(Base): 

248 """ 

249 Requests to send out invitation notifications/emails to the community for a given event occurrence 

250 """ 

251 

252 __tablename__ = "event_community_invite_requests" 

253 

254 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

255 

256 occurrence_id: Mapped[int] = mapped_column(ForeignKey("event_occurrences.id"), index=True) 

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

258 

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

260 

261 decided: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) 

262 decided_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), nullable=True) 

263 approved: Mapped[bool | None] = mapped_column(Boolean, nullable=True) 

264 

265 occurrence = relationship("EventOccurrence", backref=backref("community_invite_requests", lazy="dynamic")) 

266 user = relationship("User", foreign_keys="EventCommunityInviteRequest.user_id") 

267 

268 __table_args__ = ( 

269 # each user can only request once 

270 UniqueConstraint("occurrence_id", "user_id"), 

271 # each event can only have one notification sent out 

272 Index( 

273 "ix_event_community_invite_requests_unique", 

274 occurrence_id, 

275 unique=True, 

276 postgresql_where=and_(approved.is_not(None), approved == True), 

277 ), 

278 # decided and approved ought to be null simultaneously 

279 CheckConstraint( 

280 "((decided IS NULL) AND (decided_by_user_id IS NULL) AND (approved IS NULL)) OR \ 

281 ((decided IS NOT NULL) AND (decided_by_user_id IS NOT NULL) AND (approved IS NOT NULL))", 

282 name="decided_approved", 

283 ), 

284 )