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 

35 

36class Node(Base, kw_only=True): 

37 """ 

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

39 

40 Administered by the official cluster 

41 """ 

42 

43 __tablename__ = "nodes" 

44 

45 id: Mapped[int] = mapped_column( 

46 BigInteger, 

47 communities_seq, 

48 primary_key=True, 

49 server_default=communities_seq.next_value(), 

50 init=False, 

51 ) 

52 

53 # name and description come from the official cluster 

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

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

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

57 

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

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

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

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

62 ) 

63 official_cluster: Mapped[Cluster] = relationship( 

64 init=False, 

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

66 foreign_keys="[Cluster.parent_node_id]", 

67 uselist=False, 

68 viewonly=True, 

69 ) 

70 

71 

72class Cluster(Base, kw_only=True): 

73 """ 

74 Cluster, administered grouping of content 

75 """ 

76 

77 __tablename__ = "clusters" 

78 

79 id: Mapped[int] = mapped_column( 

80 BigInteger, 

81 communities_seq, 

82 primary_key=True, 

83 server_default=communities_seq.next_value(), 

84 init=False, 

85 ) 

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

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

88 # short description 

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

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

91 

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

93 

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

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

96 

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

98 

99 official_cluster_for_node: Mapped[Node] = relationship( 

100 init=False, 

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

102 back_populates="official_cluster", 

103 uselist=False, 

104 viewonly=True, 

105 ) 

106 

107 parent_node: Mapped[Node] = relationship( 

108 init=False, 

109 back_populates="child_clusters", 

110 remote_side="Node.id", 

111 foreign_keys="Cluster.parent_node_id", 

112 overlaps="official_cluster", 

113 ) 

114 

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

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

117 ) 

118 # all pages 

119 pages: DynamicMapped[Page] = relationship( 

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

121 ) 

122 events: Mapped[Event] = relationship( 

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

124 ) 

125 discussions: Mapped[Discussion] = relationship( 

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

127 ) 

128 

129 # includes also admins 

130 members: DynamicMapped[User] = relationship( 

131 init=False, 

132 lazy="dynamic", 

133 backref="cluster_memberships", 

134 secondary="cluster_subscriptions", 

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

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

137 viewonly=True, 

138 ) 

139 

140 admins: DynamicMapped[User] = relationship( 

141 init=False, 

142 lazy="dynamic", 

143 backref="cluster_adminships", 

144 secondary="cluster_subscriptions", 

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

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

147 viewonly=True, 

148 ) 

149 

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

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

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

153 

154 main_page: Mapped[Page] = relationship( 

155 init=False, 

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

157 viewonly=True, 

158 uselist=False, 

159 ) 

160 

161 @property 

162 def is_leaf(self) -> bool: 

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

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

165 

166 __table_args__ = ( 

167 # Each node can have at most one official cluster 

168 Index( 

169 "ix_clusters_owner_parent_node_id_is_official_cluster", 

170 parent_node_id, 

171 is_official_cluster, 

172 unique=True, 

173 postgresql_where=is_official_cluster, 

174 ), 

175 # trigram index on unaccented name 

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

177 Index( 

178 "idx_clusters_name_unaccented_trgm", 

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

180 postgresql_using="gin", 

181 ), 

182 ) 

183 

184 

185class NodeClusterAssociation(Base, kw_only=True): 

186 """ 

187 NodeClusterAssociation, grouping of nodes 

188 """ 

189 

190 __tablename__ = "node_cluster_associations" 

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

192 

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

194 

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

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

197 

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

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

200 

201 

202class ClusterRole(enum.Enum): 

203 member = enum.auto() 

204 admin = enum.auto() 

205 

206 

207class ClusterSubscription(Base, kw_only=True): 

208 """ 

209 ClusterSubscription of a user 

210 """ 

211 

212 __tablename__ = "cluster_subscriptions" 

213 

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

215 

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

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

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

219 

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

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

222 

223 __table_args__ = ( 

224 UniqueConstraint("user_id", "cluster_id"), 

225 Index( 

226 "ix_cluster_subscriptions_members", 

227 cluster_id, 

228 user_id, 

229 ), 

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

231 Index( 

232 "ix_cluster_subscriptions_admins", 

233 user_id, 

234 cluster_id, 

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

236 ), 

237 ) 

238 

239 

240class ClusterPageAssociation(Base, kw_only=True): 

241 """ 

242 pages related to clusters 

243 """ 

244 

245 __tablename__ = "cluster_page_associations" 

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

247 

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

249 

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

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

252 

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

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

255 

256 

257class PageType(enum.Enum): 

258 main_page = enum.auto() 

259 place = enum.auto() 

260 guide = enum.auto() 

261 

262 

263class Page(Base, kw_only=True): 

264 """ 

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

266 """ 

267 

268 __tablename__ = "pages" 

269 

270 id: Mapped[int] = mapped_column( 

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

272 ) 

273 

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

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

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

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

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

279 

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

281 

282 parent_node: Mapped[Node] = relationship( 

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

284 ) 

285 

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

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

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

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

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

291 ) 

292 

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

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

295 

296 __table_args__ = ( 

297 # Only one of owner_user and owner_cluster should be set 

298 CheckConstraint( 

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

300 name="one_owner", 

301 ), 

302 # Only clusters can own main pages 

303 CheckConstraint( 

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

305 name="main_page_owned_by_cluster", 

306 ), 

307 # Each cluster can have at most one main page 

308 Index( 

309 "ix_pages_owner_cluster_id_type", 

310 owner_cluster_id, 

311 type, 

312 unique=True, 

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

314 ), 

315 ) 

316 

317 def __repr__(self) -> str: 

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

319 

320 

321class PageVersion(Base, kw_only=True): 

322 """ 

323 version of page content 

324 """ 

325 

326 __tablename__ = "page_versions" 

327 

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

329 

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

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

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

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

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

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

336 # the human-readable address 

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

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

339 

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

341 

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

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

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

345 

346 __table_args__ = ( 

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

348 CheckConstraint( 

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

350 name="geom_iff_address", 

351 ), 

352 ) 

353 

354 @property 

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

356 # returns (lat, lng) or None 

357 return get_coordinates(self.geom) 

358 

359 def __repr__(self) -> str: 

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