Coverage for app/backend/src/couchers/materialized_views.py: 99%
83 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1import logging
2import typing
3from collections.abc import Sequence
4from datetime import datetime, timedelta
5from typing import Any
7from google.protobuf import empty_pb2
8from sqlalchemy import CompoundSelect, Connection, Float, Index, Integer, MetaData, Select, Table, Text, event
9from sqlalchemy.dialects.postgresql import JSON
10from sqlalchemy.orm import Mapped
11from sqlalchemy.sql import (
12 and_,
13 case,
14 cast,
15 func,
16 literal,
17 literal_column,
18 select,
19 union_all,
20)
21from sqlalchemy.sql.functions import percentile_disc
22from sqlalchemy_utils.view import (
23 CreateView,
24 DropView,
25 create_materialized_view,
26 create_table_from_selectable,
27 refresh_materialized_view,
28)
30from couchers.db import session_scope
31from couchers.helpers.completed_profile import has_completed_profile_expression
32from couchers.models import (
33 ActivenessProbe,
34 ActivenessProbeStatus,
35 Base,
36 ClusterRole,
37 ClusterSubscription,
38 Geom,
39 HostRequest,
40 MatViewBase,
41 Message,
42 MessageType,
43 StayProposalKind,
44 StrongVerificationAttempt,
45 Upload,
46 User,
47)
48from couchers.models.uploads import get_avatar_photo_subquery
50logger = logging.getLogger(__name__)
53def create_materialized_view_with_different_ddl(
54 name: str,
55 select_selectable: Select[Any] | CompoundSelect[Any],
56 create_selectable: Select[Any] | CompoundSelect[Any],
57 metadata: MetaData,
58 indexes: Sequence[Index] | None = None,
59 aliases: dict[str, str] | None = None,
60) -> Table:
61 """
62 Copied wholesale from sqlalchemy_utils (3-clause BSD), with a minor tweak in {select,create}_selectable
64 https://github.com/kvesteri/sqlalchemy-utils/blob/baf53cd1a3e779fc127010543fed53cf4a97fe16/sqlalchemy_utils/view.py#L77-L124
65 """
66 table = create_table_from_selectable(
67 name=name, selectable=select_selectable, indexes=indexes, metadata=None, aliases=aliases
68 )
70 event.listen(metadata, "after_create", CreateView(name, create_selectable, materialized=True))
72 @event.listens_for(metadata, "after_create")
73 def create_indexes(target: Any, connection: Connection, **kw: Any) -> None:
74 for idx in table.indexes:
75 idx.create(connection)
77 event.listen(metadata, "before_drop", DropView(name, materialized=True))
78 return typing.cast(Table, table)
81cluster_subscription_counts_selectable = (
82 select(
83 ClusterSubscription.cluster_id.label("cluster_id"),
84 func.count().label("count"),
85 )
86 .select_from(ClusterSubscription)
87 .outerjoin(User, User.id == ClusterSubscription.user_id)
88 .where(User.is_visible)
89 .group_by(ClusterSubscription.cluster_id)
90)
92cluster_subscription_counts = create_materialized_view(
93 "cluster_subscription_counts",
94 cluster_subscription_counts_selectable,
95 Base.metadata,
96 [
97 Index(
98 "uq_cluster_subscription_counts_cluster_id",
99 cluster_subscription_counts_selectable.subquery().c.cluster_id,
100 unique=True,
101 )
102 ],
103)
106class ClusterSubscriptionCount(MatViewBase):
107 __table__ = cluster_subscription_counts
109 cluster_id: Mapped[int]
110 count: Mapped[int]
113cluster_admin_counts_selectable = (
114 select(
115 ClusterSubscription.cluster_id.label("cluster_id"),
116 func.count().label("count"),
117 )
118 .select_from(ClusterSubscription)
119 .outerjoin(User, User.id == ClusterSubscription.user_id)
120 .where(ClusterSubscription.role == ClusterRole.admin)
121 .where(User.is_visible)
122 .group_by(ClusterSubscription.cluster_id)
123)
125cluster_admin_counts = create_materialized_view(
126 "cluster_admin_counts",
127 cluster_admin_counts_selectable,
128 Base.metadata,
129 [
130 Index(
131 "uq_cluster_admin_counts_cluster_id",
132 cluster_admin_counts_selectable.subquery().c.cluster_id,
133 unique=True,
134 )
135 ],
136)
139class ClusterAdminCount(MatViewBase):
140 __table__ = cluster_admin_counts
142 cluster_id: Mapped[int]
143 count: Mapped[int]
146def make_lite_users_selectable(create: bool = False) -> Select[Any]:
147 if create:
148 # because this is rendered as a select when emitting the CREATE VIEW, using User.geom would be rendered as
149 # `ST_AsEWKB(users.geom)` instead of the literal column, the following fixes it
150 geom_column: Any = literal_column("users.geom")
151 else:
152 geom_column = User.geom
154 strong_verification_subquery = (
155 select(User.id, literal(True).label("true"))
156 .select_from(StrongVerificationAttempt)
157 .join(User, User.id == StrongVerificationAttempt.user_id)
158 .where(StrongVerificationAttempt.has_strong_verification(User))
159 .distinct()
160 .subquery(name="sv_subquery")
161 )
163 avatar_photo_subquery = get_avatar_photo_subquery(name="avatar_photo")
165 # Pre-compute GeoJSON Feature for each user (used by GIS.GetUsers)
166 geojson_feature = cast(
167 func.json_build_object(
168 "type",
169 "Feature",
170 "geometry",
171 cast(func.ST_AsGeoJSON(geom_column, 5), JSON),
172 "properties",
173 func.json_build_object("id", User.id, "has_completed_profile", has_completed_profile_expression()),
174 ),
175 Text,
176 )
178 # Be sure to modify the LiteUser type if you add/remove columns!
179 return (
180 select(
181 User.id.label("id"),
182 User.username.label("username"),
183 User.name.label("name"),
184 User.city.label("city"),
185 User.age.label("age"),
186 geom_column.label("geom"),
187 User.geom_radius.label("radius"),
188 User.is_visible.label("is_visible"),
189 User.shadowed_at.label("shadowed_at"),
190 Upload.filename.label("avatar_filename"),
191 has_completed_profile_expression().label("has_completed_profile"),
192 User.has_completed_my_home.label("has_completed_my_home"),
193 func.coalesce(strong_verification_subquery.c.true, False).label("has_strong_verification"),
194 geojson_feature.label("geojson"),
195 )
196 .select_from(User)
197 .outerjoin(
198 avatar_photo_subquery,
199 avatar_photo_subquery.c.gallery_id == User.profile_gallery_id,
200 )
201 .outerjoin(Upload, Upload.key == avatar_photo_subquery.c.upload_key)
202 .outerjoin(strong_verification_subquery, strong_verification_subquery.c.id == User.id)
203 )
206lite_users_selectable_select = make_lite_users_selectable(create=False)
207lite_users_selectable_create = make_lite_users_selectable(create=True)
209lite_users_subquery = lite_users_selectable_create.subquery()
211lite_users = create_materialized_view_with_different_ddl(
212 "lite_users",
213 lite_users_selectable_select,
214 lite_users_selectable_create,
215 Base.metadata,
216 [
217 Index("uq_lite_users_id", lite_users_subquery.c.id, unique=True),
218 Index("uq_lite_users_username", lite_users_subquery.c.username, unique=True),
219 Index(
220 "ix_lite_users_id_visible",
221 lite_users_subquery.c.id,
222 postgresql_using="hash",
223 postgresql_where=lite_users_subquery.c.is_visible,
224 ),
225 Index(
226 "ix_lite_users_username_visible",
227 lite_users_subquery.c.username,
228 postgresql_using="hash",
229 postgresql_where=lite_users_subquery.c.is_visible,
230 ),
231 ],
232)
235class LiteUser(MatViewBase):
236 __table__ = lite_users
238 # A subset enough to make mypy happy. Taken from "make_lite_users_selectable".
239 id: Mapped[int]
240 username: Mapped[str]
241 name: Mapped[str]
242 city: Mapped[str]
243 age: Mapped[int]
244 geom: Mapped[Geom]
245 radius: Mapped[float]
246 is_visible: Mapped[bool]
247 shadowed_at: Mapped[datetime | None]
248 avatar_filename: Mapped[str]
249 has_completed_profile: Mapped[bool]
250 has_completed_my_home: Mapped[bool]
251 has_strong_verification: Mapped[bool]
252 geojson: Mapped[str]
255def make_clustered_users_selectable(create: bool = False) -> CompoundSelect[Any]:
256 # emits something along the lines of
257 # WITH anon_1 AS (
258 # SELECT id,
259 # geom,
260 # ST_ClusterDBSCAN(geom, eps := .15, minpoints := 5) OVER (ORDER BY id) AS cluster_id
261 # FROM users
262 # WHERE geom IS NOT NULL
263 # )
265 cluster_cte = (
266 select(
267 User.id,
268 User.geom,
269 # DBSCAN clustering with epsilon=.15 deg (~17 km), minpoints=5, cluster will be NULL for not in any cluster
270 func.ST_ClusterDBSCAN(User.geom, 0.15, 5).over(order_by=User.id).label("cluster_id"),
271 )
272 .where(User.is_visible)
273 .cte("clustered")
274 )
276 if create:
277 centroid_geom: Any = literal_column("ST_Centroid(ST_Collect(clustered.geom))")
278 cluster_geom: Any = literal_column("clustered.geom")
279 else:
280 centroid_geom = func.ST_Centroid(func.ST_Collect(cluster_cte.c.geom))
281 cluster_geom = cluster_cte.c.geom
283 clustered_users = (
284 select(centroid_geom.label("geom"), func.count().label("count"))
285 .select_from(cluster_cte)
286 .where(cluster_cte.c.cluster_id != None)
287 .group_by(cluster_cte.c.cluster_id)
288 )
290 isolated_users = (
291 select(cluster_geom.label("geom"), literal(1, type_=Integer).label("count"))
292 .select_from(cluster_cte)
293 .where(cluster_cte.c.cluster_id == None)
294 )
296 return union_all(clustered_users, isolated_users)
299clustered_users_selectable_select = make_clustered_users_selectable(create=False)
300clustered_users_selectable_create = make_clustered_users_selectable(create=True)
302clustered_users = create_materialized_view_with_different_ddl(
303 "clustered_users", clustered_users_selectable_select, clustered_users_selectable_create, Base.metadata
304)
307class ClusteredUser(MatViewBase):
308 __table__ = clustered_users
310 geom: Mapped[Geom]
311 count: Mapped[int]
314def float_(stmt: Any) -> Any:
315 return func.coalesce(cast(stmt, Float), 0.0)
318# this subquery gets the time that the request was sent
319t = select(Message.conversation_id, Message.time).where(Message.message_type == MessageType.chat_created).subquery()
320# this subquery gets the time that the user responded to the request
321s = (
322 select(Message.conversation_id, Message.author_id, func.min(Message.time).label("time"))
323 .group_by(Message.conversation_id, Message.author_id)
324 .subquery()
325)
326all_responses = union_all(
327 # host request responses
328 select(
329 HostRequest.recipient_user_id.label("user_id"),
330 (s.c.time - t.c.time).label("response_time"),
331 )
332 .join(t, t.c.conversation_id == HostRequest.conversation_id)
333 .outerjoin(
334 s, and_(s.c.conversation_id == HostRequest.conversation_id, s.c.author_id == HostRequest.recipient_user_id)
335 )
336 .where(HostRequest.proposal_kind == StayProposalKind.request),
337 # activeness probes
338 select(
339 ActivenessProbe.user_id,
340 (
341 # expired probes have a responded time for when they were marked responded
342 case(
343 (
344 ActivenessProbe.response != ActivenessProbeStatus.expired,
345 ActivenessProbe.responded - ActivenessProbe.probe_initiated,
346 ),
347 else_=None,
348 )
349 ).label("response_time"),
350 ),
351).subquery()
353user_response_rates_selectable = select(
354 all_responses.c.user_id.label("user_id"),
355 # number of requests received
356 func.count().label("requests"),
357 # percentage of requests responded to
358 (func.count(all_responses.c.response_time) / func.count()).label("response_rate"),
359 func.avg(all_responses.c.response_time).label("avg_response_time"),
360 # the 33rd percentile response time
361 percentile_disc(0.33)
362 .within_group(func.coalesce(all_responses.c.response_time, timedelta(days=1000)))
363 .label("response_time_33p"),
364 # the 66th percentile response time
365 percentile_disc(0.66)
366 .within_group(func.coalesce(all_responses.c.response_time, timedelta(days=1000)))
367 .label("response_time_66p"),
368).group_by(all_responses.c.user_id)
370user_response_rates = create_materialized_view(
371 "user_response_rates",
372 user_response_rates_selectable,
373 Base.metadata,
374 [Index("uq_user_response_rates_id", user_response_rates_selectable.subquery().c.user_id, unique=True)],
375)
378class UserResponseRate(MatViewBase):
379 __table__ = user_response_rates
381 user_id: Mapped[int]
382 requests: Mapped[int]
383 response_rate: Mapped[float]
384 avg_response_time: Mapped[float]
385 response_time_33p: Mapped[timedelta]
386 response_time_66p: Mapped[timedelta]
389def refresh_materialized_views(payload: empty_pb2.Empty) -> None:
390 logger.info("Refreshing materialized views")
391 with session_scope() as session:
392 refresh_materialized_view(session, "cluster_subscription_counts", concurrently=True)
393 refresh_materialized_view(session, "cluster_admin_counts", concurrently=True)
394 refresh_materialized_view(session, "clustered_users")
395 refresh_materialized_view(session, "user_response_rates", concurrently=True)
398def refresh_materialized_views_rapid(payload: empty_pb2.Empty) -> None:
399 logger.info("Refreshing materialized views (rapid)")
400 with session_scope() as session:
401 refresh_materialized_view(session, "lite_users", concurrently=True)