Coverage for app / backend / src / couchers / models / clusters.py: 98%

1import enum 

2from datetime import datetime 

3from typing import TYPE_CHECKING 

4 

5from geoalchemy2 import Geometry 

6from sqlalchemy import ( 

7 BigInteger, 

8 Boolean, 

9 CheckConstraint, 

10 DateTime, 

11 Enum, 

12 ForeignKey, 

13 Index, 

14 String, 

15 UniqueConstraint, 

16 func, 

17 text, 

18) 

19from sqlalchemy.orm import ( 

20 DynamicMapped, 

21 Mapped, 

22 column_property, 

23 deferred, 

24 mapped_column, 

25 relationship, 

26) 

27from sqlalchemy.sql import expression 

28 

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

30from couchers.utils import get_coordinates 

31 

32if TYPE_CHECKING: 

33 from couchers.models import Discussion, Event, Thread, Upload, User 

34 from couchers.models.public_trips import PublicTrip 

35 

36 

37class NodeType(enum.Enum): 

38 # Ordinal: lower values are broader geographic areas 

39 world = 1 

40 macroregion = 2 

41 region = 3 

42 subregion = 4 

43 locality = 5 

44 sublocality = 6 

45 

46 

47class Node(Base, kw_only=True): 

48 """ 

49 Node, i.e., geographical subdivision of the world 

50 

51 Administered by the official cluster 

52 """ 

53 

54 __tablename__ = "nodes" 

55 

56 id: Mapped[int] = mapped_column( 

57 BigInteger, 

58 communities_seq, 

59 primary_key=True, 

60 server_default=communities_seq.next_value(), 

61 init=False, 

62 ) 

63 

64 node_type: Mapped[NodeType] = mapped_column(Enum(NodeType)) 

65 

66 # name and description come from the official cluster 

67 parent_node_id: Mapped[int | None] = mapped_column(ForeignKey("nodes.id"), default=None, index=True) 

68 geom: Mapped[Geom] = deferred(mapped_column(Geometry(geometry_type="MULTIPOLYGON", srid=4326), nullable=False)) 

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

70 

71 parent_node: Mapped[Node] = relationship(init=False, back_populates="child_nodes", remote_side="Node.id") 

72 child_nodes: Mapped[list[Node]] = relationship(init=False) 

73 child_clusters: Mapped[list[Cluster]] = relationship( 

74 init=False, back_populates="parent_node", overlaps="official_cluster" 

75 ) 

76 official_cluster: Mapped[Cluster] = relationship( 

77 init=False, 

78 primaryjoin="and_(Node.id == Cluster.parent_node_id, Cluster.is_official_cluster)", 

79 foreign_keys="[Cluster.parent_node_id]", 

80 uselist=False, 

81 viewonly=True, 

82 ) 

83 public_trips: Mapped[list[PublicTrip]] = relationship(init=False, back_populates="node") 

84 

85 

86class Cluster(Base, kw_only=True): 

87 """ 

88 Cluster, administered grouping of content 

89 """ 

90 

91 __tablename__ = "clusters" 

92 

93 id: Mapped[int] = mapped_column( 

94 BigInteger, 

95 communities_seq, 

96 primary_key=True, 

97 server_default=communities_seq.next_value(), 

98 init=False, 

99 ) 

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

101 name: Mapped[str] = mapped_column(String) 

102 # short description 

103 description: Mapped[str] = mapped_column(String) 

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

105 

106 is_official_cluster: Mapped[bool] = mapped_column(Boolean, default=False) 

107 

108 discussions_enabled: Mapped[bool] = mapped_column(Boolean, default=True, server_default=expression.true()) 

109 events_enabled: Mapped[bool] = mapped_column(Boolean, default=True, server_default=expression.true()) 

110 

111 slug: Mapped[str] = column_property(func.slugify(name)) 

112 

113 official_cluster_for_node: Mapped[Node] = relationship( 

114 init=False, 

115 primaryjoin="and_(Cluster.parent_node_id == Node.id, Cluster.is_official_cluster)", 

116 back_populates="official_cluster", 

117 uselist=False, 

118 viewonly=True, 

119 ) 

120 

121 parent_node: Mapped[Node] = relationship( 

122 init=False, 

123 back_populates="child_clusters", 

124 remote_side="Node.id", 

125 foreign_keys="Cluster.parent_node_id", 

126 overlaps="official_cluster", 

127 ) 

128 

129 nodes: Mapped[list[Cluster]] = relationship( 

130 init=False, backref="clusters", secondary="node_cluster_associations", viewonly=True 

131 ) 

132 # all pages 

133 pages: DynamicMapped[Page] = relationship( 

134 init=False, backref="clusters", secondary="cluster_page_associations", lazy="dynamic", viewonly=True 

135 ) 

136 events: Mapped[Event] = relationship( 

137 init=False, backref="clusters", secondary="cluster_event_associations", viewonly=True 

138 ) 

139 discussions: Mapped[Discussion] = relationship( 

140 init=False, backref="clusters", secondary="cluster_discussion_associations", viewonly=True 

141 ) 

142 

143 # includes also admins 

144 members: DynamicMapped[User] = relationship( 

145 init=False, 

146 lazy="dynamic", 

147 backref="cluster_memberships", 

148 secondary="cluster_subscriptions", 

149 primaryjoin="Cluster.id == ClusterSubscription.cluster_id", 

150 secondaryjoin="User.id == ClusterSubscription.user_id", 

151 viewonly=True, 

152 ) 

153 

154 admins: DynamicMapped[User] = relationship( 

155 init=False, 

156 lazy="dynamic", 

157 backref="cluster_adminships", 

158 secondary="cluster_subscriptions", 

159 primaryjoin="Cluster.id == ClusterSubscription.cluster_id", 

160 secondaryjoin="and_(User.id == ClusterSubscription.user_id, ClusterSubscription.role == 'admin')", 

161 viewonly=True, 

162 ) 

163 

164 cluster_subscriptions: Mapped[list[ClusterSubscription]] = relationship(init=False) 

165 owned_pages: DynamicMapped[Page] = relationship(init=False, lazy="dynamic") 

166 owned_discussions: DynamicMapped[Discussion] = relationship(init=False, lazy="dynamic") 

167 

168 main_page: Mapped[Page] = relationship( 

169 init=False, 

170 primaryjoin="and_(Cluster.id == Page.owner_cluster_id, Page.type == 'main_page')", 

171 viewonly=True, 

172 uselist=False, 

173 ) 

174 

175 @property 

176 def is_leaf(self) -> bool: 

177 """Whether the cluster is a leaf node in the cluster hierarchy.""" 

178 return len(self.parent_node.child_nodes) == 0 

179 

180 __table_args__ = ( 

181 # Each node can have at most one official cluster 

182 Index( 

183 "ix_clusters_owner_parent_node_id_is_official_cluster", 

184 parent_node_id, 

185 is_official_cluster, 

186 unique=True, 

187 postgresql_where=is_official_cluster, 

188 ), 

189 # trigram index on unaccented name 

190 # note that the function `unaccent` is not immutable so cannot be used in an index, that's why we wrap it 

191 Index( 

192 "idx_clusters_name_unaccented_trgm", 

193 text("immutable_unaccent(name) gin_trgm_ops"), 

194 postgresql_using="gin", 

195 ), 

196 ) 

197 

198 

199class NodeClusterAssociation(Base, kw_only=True): 

200 """ 

201 NodeClusterAssociation, grouping of nodes 

202 """ 

203 

204 __tablename__ = "node_cluster_associations" 

205 __table_args__ = (UniqueConstraint("node_id", "cluster_id"),) 

206 

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

208 

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

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

211 

212 node: Mapped[Node] = relationship(init=False, backref="node_cluster_associations") 

213 cluster: Mapped[Cluster] = relationship(init=False, backref="node_cluster_associations") 

214 

215 

216class ClusterRole(enum.Enum): 

217 member = enum.auto() 

218 admin = enum.auto() 

219 

220 

221class ClusterSubscription(Base, kw_only=True): 

222 """ 

223 ClusterSubscription of a user 

224 """ 

225 

226 __tablename__ = "cluster_subscriptions" 

227 

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

229 

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

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

232 role: Mapped[ClusterRole] = mapped_column(Enum(ClusterRole)) 

233 

234 user: Mapped[User] = relationship(init=False, backref="cluster_subscriptions") 

235 cluster: Mapped[Cluster] = relationship(init=False, back_populates="cluster_subscriptions") 

236 

237 __table_args__ = ( 

238 UniqueConstraint("user_id", "cluster_id"), 

239 Index( 

240 "ix_cluster_subscriptions_members", 

241 cluster_id, 

242 user_id, 

243 ), 

244 # For fast lookup of nodes this user is an admin of 

245 Index( 

246 "ix_cluster_subscriptions_admins", 

247 user_id, 

248 cluster_id, 

249 postgresql_where=(role == ClusterRole.admin), 

250 ), 

251 ) 

252 

253 

254class ClusterPageAssociation(Base, kw_only=True): 

255 """ 

256 pages related to clusters 

257 """ 

258 

259 __tablename__ = "cluster_page_associations" 

260 __table_args__ = (UniqueConstraint("page_id", "cluster_id"),) 

261 

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

263 

264 page_id: Mapped[int] = mapped_column(ForeignKey("pages.id"), index=True) 

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

266 

267 page: Mapped[Page] = relationship(init=False, backref="cluster_page_associations") 

268 cluster: Mapped[Cluster] = relationship(init=False, backref="cluster_page_associations") 

269 

270 

271class PageType(enum.Enum): 

272 main_page = enum.auto() 

273 place = enum.auto() 

274 guide = enum.auto() 

275 

276 

277class Page(Base, kw_only=True): 

278 """ 

279 similar to a wiki page about a community, POI or guide 

280 """ 

281 

282 __tablename__ = "pages" 

283 

284 id: Mapped[int] = mapped_column( 

285 BigInteger, communities_seq, primary_key=True, server_default=communities_seq.next_value(), init=False 

286 ) 

287 

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

289 type: Mapped[PageType] = mapped_column(Enum(PageType)) 

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

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

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

293 

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

295 

296 parent_node: Mapped[Node] = relationship( 

297 init=False, backref="child_pages", remote_side="Node.id", foreign_keys="Page.parent_node_id" 

298 ) 

299 

300 thread: Mapped[Thread] = relationship(init=False, backref="page", uselist=False) 

301 creator_user: Mapped[User] = relationship(init=False, backref="created_pages", foreign_keys="Page.creator_user_id") 

302 owner_user: Mapped[User | None] = relationship(init=False, backref="owned_pages", foreign_keys="Page.owner_user_id") 

303 owner_cluster: Mapped[Cluster | None] = relationship( 

304 init=False, back_populates="owned_pages", uselist=False, foreign_keys="Page.owner_cluster_id" 

305 ) 

306 

307 editors: Mapped[list[User]] = relationship(init=False, secondary="page_versions", viewonly=True) 

308 versions: Mapped[list[PageVersion]] = relationship(init=False, back_populates="page", order_by="PageVersion.id") 

309 

310 __table_args__ = ( 

311 # Only one of owner_user and owner_cluster should be set 

312 CheckConstraint( 

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

314 name="one_owner", 

315 ), 

316 # Only clusters can own main pages 

317 CheckConstraint( 

318 "NOT (owner_cluster_id IS NULL AND type = 'main_page')", 

319 name="main_page_owned_by_cluster", 

320 ), 

321 # Each cluster can have at most one main page 

322 Index( 

323 "ix_pages_owner_cluster_id_type", 

324 owner_cluster_id, 

325 type, 

326 unique=True, 

327 postgresql_where=(type == PageType.main_page), 

328 ), 

329 ) 

330 

331 def __repr__(self) -> str: 

332 return f"Page({self.id=})" 

333 

334 

335class PageVersion(Base, kw_only=True): 

336 """ 

337 version of page content 

338 """ 

339 

340 __tablename__ = "page_versions" 

341 

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

343 

344 page_id: Mapped[int] = mapped_column(ForeignKey("pages.id"), index=True) 

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

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

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

348 photo_key: Mapped[str | None] = mapped_column(ForeignKey("uploads.key"), default=None) 

349 geom: Mapped[Geom | None] = mapped_column(Geometry(geometry_type="POINT", srid=4326), default=None) 

350 # the human-readable address 

351 address: Mapped[str | None] = mapped_column(String, default=None) 

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

353 

354 slug: Mapped[str] = column_property(func.slugify(title)) 

355 

356 page: Mapped[Page] = relationship(init=False, back_populates="versions") 

357 editor_user: Mapped[User] = relationship(init=False, backref="edited_pages") 

358 photo: Mapped[Upload] = relationship(init=False) 

359 

360 __table_args__ = ( 

361 # Geom and address must either both be null or both be set 

362 CheckConstraint( 

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

364 name="geom_iff_address", 

365 ), 

366 ) 

367 

368 @property 

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

370 # returns (lat, lng) or None 

371 return get_coordinates(self.geom) 

372 

373 def __repr__(self) -> str: 

374 return f"PageVersion({self.id=}, {self.page_id=})"