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

1import enum 

2from datetime import datetime 

3 

4from geoalchemy2 import Geometry 

5from sqlalchemy import ( 

6 BigInteger, 

7 Boolean, 

8 CheckConstraint, 

9 DateTime, 

10 Enum, 

11 ForeignKey, 

12 Index, 

13 String, 

14 UniqueConstraint, 

15 func, 

16 text, 

17) 

18from sqlalchemy.orm import Mapped, backref, column_property, deferred, mapped_column, relationship 

19from sqlalchemy.sql import expression 

20 

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

22from couchers.utils import get_coordinates 

23 

24 

25class Node(Base): 

26 """ 

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

28 

29 Administered by the official cluster 

30 """ 

31 

32 __tablename__ = "nodes" 

33 

34 id: Mapped[int] = mapped_column( 

35 BigInteger, 

36 communities_seq, 

37 primary_key=True, 

38 server_default=communities_seq.next_value(), 

39 ) 

40 

41 # name and description come from the official cluster 

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

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

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

45 

46 parent_node = relationship("Node", backref="child_nodes", remote_side="Node.id") 

47 

48 

49class Cluster(Base): 

50 """ 

51 Cluster, administered grouping of content 

52 """ 

53 

54 __tablename__ = "clusters" 

55 

56 id: Mapped[int] = mapped_column( 

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

58 ) 

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

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

61 # short description 

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

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

64 

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

66 

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

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

69 

70 slug = column_property(func.slugify(name)) 

71 

72 official_cluster_for_node = relationship( 

73 "Node", 

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

75 backref=backref("official_cluster", uselist=False), 

76 uselist=False, 

77 viewonly=True, 

78 ) 

79 

80 parent_node = relationship( 

81 "Node", backref="child_clusters", remote_side="Node.id", foreign_keys="Cluster.parent_node_id" 

82 ) 

83 

84 nodes = relationship("Cluster", backref="clusters", secondary="node_cluster_associations", viewonly=True) 

85 # all pages 

86 pages = relationship( 

87 "Page", backref="clusters", secondary="cluster_page_associations", lazy="dynamic", viewonly=True 

88 ) 

89 events = relationship("Event", backref="clusters", secondary="cluster_event_associations", viewonly=True) 

90 discussions = relationship( 

91 "Discussion", backref="clusters", secondary="cluster_discussion_associations", viewonly=True 

92 ) 

93 

94 # includes also admins 

95 members = relationship( 

96 "User", 

97 lazy="dynamic", 

98 backref="cluster_memberships", 

99 secondary="cluster_subscriptions", 

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

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

102 viewonly=True, 

103 ) 

104 

105 admins = relationship( 

106 "User", 

107 lazy="dynamic", 

108 backref="cluster_adminships", 

109 secondary="cluster_subscriptions", 

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

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

112 viewonly=True, 

113 ) 

114 

115 main_page = relationship( 

116 "Page", 

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

118 viewonly=True, 

119 uselist=False, 

120 ) 

121 

122 @property 

123 def is_leaf(self) -> bool: 

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

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

126 

127 __table_args__ = ( 

128 # Each node can have at most one official cluster 

129 Index( 

130 "ix_clusters_owner_parent_node_id_is_official_cluster", 

131 parent_node_id, 

132 is_official_cluster, 

133 unique=True, 

134 postgresql_where=is_official_cluster, 

135 ), 

136 # trigram index on unaccented name 

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

138 Index( 

139 "idx_clusters_name_unaccented_trgm", 

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

141 postgresql_using="gin", 

142 ), 

143 ) 

144 

145 

146class NodeClusterAssociation(Base): 

147 """ 

148 NodeClusterAssociation, grouping of nodes 

149 """ 

150 

151 __tablename__ = "node_cluster_associations" 

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

153 

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

155 

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

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

158 

159 node = relationship("Node", backref="node_cluster_associations") 

160 cluster = relationship("Cluster", backref="node_cluster_associations") 

161 

162 

163class ClusterRole(enum.Enum): 

164 member = enum.auto() 

165 admin = enum.auto() 

166 

167 

168class ClusterSubscription(Base): 

169 """ 

170 ClusterSubscription of a user 

171 """ 

172 

173 __tablename__ = "cluster_subscriptions" 

174 

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

176 

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

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

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

180 

181 user = relationship("User", backref="cluster_subscriptions") 

182 cluster = relationship("Cluster", backref="cluster_subscriptions") 

183 

184 __table_args__ = ( 

185 UniqueConstraint("user_id", "cluster_id"), 

186 Index( 

187 "ix_cluster_subscriptions_members", 

188 cluster_id, 

189 user_id, 

190 ), 

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

192 Index( 

193 "ix_cluster_subscriptions_admins", 

194 user_id, 

195 cluster_id, 

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

197 ), 

198 ) 

199 

200 

201class ClusterPageAssociation(Base): 

202 """ 

203 pages related to clusters 

204 """ 

205 

206 __tablename__ = "cluster_page_associations" 

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

208 

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

210 

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

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

213 

214 page = relationship("Page", backref="cluster_page_associations") 

215 cluster = relationship("Cluster", backref="cluster_page_associations") 

216 

217 

218class PageType(enum.Enum): 

219 main_page = enum.auto() 

220 place = enum.auto() 

221 guide = enum.auto() 

222 

223 

224class Page(Base): 

225 """ 

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

227 """ 

228 

229 __tablename__ = "pages" 

230 

231 id: Mapped[int] = mapped_column( 

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

233 ) 

234 

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

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

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

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

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

240 

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

242 

243 parent_node = relationship("Node", backref="child_pages", remote_side="Node.id", foreign_keys="Page.parent_node_id") 

244 

245 thread = relationship("Thread", backref="page", uselist=False) 

246 creator_user = relationship("User", backref="created_pages", foreign_keys="Page.creator_user_id") 

247 owner_user = relationship("User", backref="owned_pages", foreign_keys="Page.owner_user_id") 

248 owner_cluster = relationship( 

249 "Cluster", backref=backref("owned_pages", lazy="dynamic"), uselist=False, foreign_keys="Page.owner_cluster_id" 

250 ) 

251 

252 editors = relationship("User", secondary="page_versions", viewonly=True) 

253 

254 __table_args__ = ( 

255 # Only one of owner_user and owner_cluster should be set 

256 CheckConstraint( 

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

258 name="one_owner", 

259 ), 

260 # Only clusters can own main pages 

261 CheckConstraint( 

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

263 name="main_page_owned_by_cluster", 

264 ), 

265 # Each cluster can have at most one main page 

266 Index( 

267 "ix_pages_owner_cluster_id_type", 

268 owner_cluster_id, 

269 type, 

270 unique=True, 

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

272 ), 

273 ) 

274 

275 def __repr__(self) -> str: 

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

277 

278 

279class PageVersion(Base): 

280 """ 

281 version of page content 

282 """ 

283 

284 __tablename__ = "page_versions" 

285 

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

287 

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

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

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

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

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

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

294 # the human-readable address 

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

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

297 

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

299 

300 page = relationship("Page", backref="versions", order_by="PageVersion.id") 

301 editor_user = relationship("User", backref="edited_pages") 

302 photo = relationship("Upload") 

303 

304 __table_args__ = ( 

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

306 CheckConstraint( 

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

308 name="geom_iff_address", 

309 ), 

310 ) 

311 

312 @property 

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

314 # returns (lat, lng) or None 

315 return get_coordinates(self.geom) 

316 

317 def __repr__(self) -> str: 

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