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