Coverage for app/backend/src/couchers/servicers/public.py: 91%
79 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +0000
1import logging
2import threading
4import grpc
5from cachetools import TTLCache, cached
6from google.protobuf import empty_pb2
7from sqlalchemy import null, select
8from sqlalchemy.orm import Session, selectinload
9from sqlalchemy.sql import func, union_all
11from couchers import experimentation, urls
12from couchers.context import CouchersContext, make_logged_out_context
13from couchers.i18n import LocalizationContext
14from couchers.materialized_views import LiteUser
15from couchers.models import (
16 Cluster,
17 Invoice,
18 InvoiceType,
19 Node,
20 ProfilePublicVisibility,
21 Reference,
22 User,
23 Volunteer,
24)
25from couchers.models.uploads import get_avatar_upload
26from couchers.proto import api_pb2, public_pb2, public_pb2_grpc
27from couchers.proto.google.api import httpbody_pb2
28from couchers.resources import get_static_badge_dict
29from couchers.servicers.api import fluency2api, hostingstatus2api, meetupstatus2api, user_model_to_pb
30from couchers.servicers.gis import _statement_to_geojson_response
31from couchers.sql import users_visible
32from couchers.utils import Timestamp_from_datetime, not_none, now
34logger = logging.getLogger(__name__)
37def format_volunteer_link(volunteer: Volunteer, username: str) -> dict[str, str]:
38 """Format volunteer link information into a dict with link_type, link_text, and link_url."""
39 if volunteer.link_type:
40 return dict(
41 link_type=volunteer.link_type,
42 link_text=not_none(volunteer.link_text),
43 link_url=not_none(volunteer.link_url),
44 )
45 else:
46 return dict(
47 link_type="couchers",
48 link_text=f"@{username}",
49 link_url=urls.user_link(username=username),
50 )
53@cached(cache=TTLCache(maxsize=1, ttl=600), key=lambda _: None, lock=threading.Lock())
54def _get_public_users(session: Session) -> httpbody_pb2.HttpBody:
55 # the response is cached and shared between all callers, so it must not depend on who is asking; logged-in users
56 # get the per-viewer map from Gis.GetUsers instead
57 context = make_logged_out_context(localization=LocalizationContext.en_utc())
59 with_geom = (
60 select(User.username, User.geom)
61 .where(users_visible(context, User))
62 .where(User.public_visibility != ProfilePublicVisibility.nothing)
63 .where(User.public_visibility != ProfilePublicVisibility.map_only)
64 )
66 without_geom = (
67 select(null(), User.randomized_geom)
68 .where(users_visible(context, User))
69 .where(User.randomized_geom != None)
70 .where(User.public_visibility == ProfilePublicVisibility.map_only)
71 )
72 return _statement_to_geojson_response(session, union_all(with_geom, without_geom))
75@cached(cache=TTLCache(maxsize=1, ttl=60), key=lambda _: None, lock=threading.Lock())
76def _get_signup_page_info(session: Session) -> public_pb2.GetSignupPageInfoRes:
77 # last user who signed up
78 last_signup, geom = session.execute(
79 select(User.joined, User.geom).where(User.is_visible).order_by(User.id.desc()).limit(1)
80 ).one()
82 communities = (
83 session.execute(
84 select(Cluster.name)
85 .join(Node, Node.id == Cluster.parent_node_id)
86 .where(Cluster.is_official_cluster)
87 .where(func.ST_Contains(Node.geom, geom))
88 .order_by(Cluster.id.asc())
89 )
90 .scalars()
91 .all()
92 )
94 if len(communities) <= 1: 94 ↛ 97line 94 didn't jump to line 97 because the condition on line 94 was always true
95 # either no community or just global community
96 last_location = "The World"
97 elif len(communities) == 3:
98 # probably global, continent, region, so let's just return the region
99 last_location = communities[-1]
100 else:
101 # probably global, continent, region, city
102 last_location = f"{communities[-1]}, {communities[-2]}"
104 user_count = session.execute(select(func.count()).select_from(User).where(User.is_visible)).scalar_one()
106 return public_pb2.GetSignupPageInfoRes(
107 last_signup=Timestamp_from_datetime(last_signup.replace(second=0, microsecond=0)),
108 last_location=last_location,
109 user_count=user_count,
110 )
113@cached(cache=TTLCache(maxsize=1, ttl=60), key=lambda _: None, lock=threading.Lock())
114def _get_donation_stats(session: Session) -> public_pb2.GetDonationStatsRes:
115 """Get year-to-date donation statistics, excluding merch purchases."""
116 start_of_year = now().replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
118 total_donated = session.execute(
119 select(func.coalesce(func.sum(Invoice.amount), 0))
120 .where(Invoice.invoice_type == InvoiceType.on_platform)
121 .where(Invoice.created >= start_of_year)
122 ).scalar_one()
124 # No request user here (public, cached endpoint), so evaluate the drive's goal/offset globally.
125 # The defaults reproduce the historical drive config; the offset excludes large one-off donations.
126 goal = experimentation.get_global_integer_value("donation_goal_usd", 5000)
127 offset = experimentation.get_global_integer_value("donation_offset_usd", 2000)
129 return public_pb2.GetDonationStatsRes(
130 total_donated_ytd=max(int(total_donated - offset), 0),
131 goal=goal,
132 )
135@cached(cache=TTLCache(maxsize=1, ttl=5), key=lambda _: None, lock=threading.Lock())
136def _get_volunteers(session: Session) -> public_pb2.GetVolunteersRes:
137 volunteers = session.execute(
138 select(Volunteer, LiteUser)
139 .join(LiteUser, LiteUser.id == Volunteer.user_id)
140 .where(LiteUser.is_visible)
141 .where(Volunteer.show_on_team_page)
142 .order_by(
143 Volunteer.sort_key.asc().nulls_last(),
144 Volunteer.stopped_volunteering.desc().nulls_first(),
145 Volunteer.started_volunteering.asc(),
146 )
147 ).all()
149 board_members = set(get_static_badge_dict()["board_member"])
151 def format_volunteer(volunteer: Volunteer, lite_user: LiteUser) -> public_pb2.Volunteer:
152 return public_pb2.Volunteer(
153 name=volunteer.display_name or lite_user.name,
154 username=lite_user.username,
155 is_board_member=lite_user.id in board_members,
156 role=volunteer.role,
157 location=volunteer.display_location or lite_user.city,
158 img=urls.media_url(filename=lite_user.avatar_filename, size="thumbnail")
159 if lite_user.avatar_filename
160 else None,
161 **format_volunteer_link(volunteer, lite_user.username),
162 )
164 return public_pb2.GetVolunteersRes(
165 current_volunteers=[
166 format_volunteer(volunteer, lite_user)
167 for volunteer, lite_user in volunteers
168 if volunteer.stopped_volunteering is None
169 ],
170 past_volunteers=[
171 format_volunteer(volunteer, lite_user)
172 for volunteer, lite_user in volunteers
173 if volunteer.stopped_volunteering is not None
174 ],
175 )
178class Public(public_pb2_grpc.PublicServicer):
179 """
180 Public (logged-out) APIs for getting public info
181 """
183 def GetPublicUsers(
184 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
185 ) -> httpbody_pb2.HttpBody:
186 return _get_public_users(session)
188 def GetPublicUser(
189 self, request: public_pb2.GetPublicUserReq, context: CouchersContext, session: Session
190 ) -> public_pb2.GetPublicUserRes:
191 user = session.execute(
192 select(User)
193 .where(users_visible(context, User))
194 .where(User.username == request.user)
195 .where(
196 User.public_visibility.in_(
197 [ProfilePublicVisibility.limited, ProfilePublicVisibility.most, ProfilePublicVisibility.full]
198 )
199 )
200 .options(
201 selectinload(User.badges),
202 selectinload(User.regions_visited),
203 selectinload(User.regions_lived),
204 selectinload(User.language_abilities),
205 )
206 ).scalar_one_or_none()
208 if not user:
209 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
211 if user.public_visibility == ProfilePublicVisibility.full:
212 return public_pb2.GetPublicUserRes(
213 full_user=user_model_to_pb(user, session, make_logged_out_context(localization=context.localization))
214 )
216 num_references = session.execute(
217 select(func.count())
218 .select_from(Reference)
219 .join(User, User.id == Reference.from_user_id)
220 .where(users_visible(context, User))
221 .where(Reference.to_user_id == user.id)
222 ).scalar_one()
224 if user.public_visibility == ProfilePublicVisibility.limited:
225 return public_pb2.GetPublicUserRes(
226 limited_user=public_pb2.LimitedUser(
227 username=user.username,
228 name=user.name,
229 city=user.city,
230 hometown=user.hometown,
231 num_references=num_references,
232 joined=Timestamp_from_datetime(user.display_joined),
233 hosting_status=hostingstatus2api[user.hosting_status],
234 meetup_status=meetupstatus2api[user.meetup_status],
235 badges=[badge.badge_id for badge in user.badges],
236 )
237 )
239 if user.public_visibility == ProfilePublicVisibility.most: 239 ↛ 272line 239 didn't jump to line 272 because the condition on line 239 was always true
240 avatar_upload = get_avatar_upload(session, user)
242 return public_pb2.GetPublicUserRes(
243 most_user=public_pb2.MostUser(
244 username=user.username,
245 name=user.name,
246 city=user.city,
247 hometown=user.hometown,
248 timezone=user.timezone,
249 num_references=num_references,
250 gender=user.gender,
251 pronouns=user.pronouns,
252 age=int(user.age),
253 joined=Timestamp_from_datetime(user.display_joined),
254 last_active=Timestamp_from_datetime(user.display_last_active),
255 hosting_status=hostingstatus2api[user.hosting_status],
256 meetup_status=meetupstatus2api[user.meetup_status],
257 occupation=user.occupation,
258 education=user.education,
259 about_me=user.about_me,
260 things_i_like=user.things_i_like,
261 language_abilities=[
262 api_pb2.LanguageAbility(code=ability.language_code, fluency=fluency2api[ability.fluency])
263 for ability in user.language_abilities
264 ],
265 regions_visited=[region.code for region in user.regions_visited],
266 regions_lived=[region.code for region in user.regions_lived],
267 avatar_url=avatar_upload.full_url if avatar_upload else None,
268 avatar_thumbnail_url=avatar_upload.thumbnail_url if avatar_upload else None,
269 badges=[badge.badge_id for badge in user.badges],
270 )
271 )
272 raise RuntimeError(user.public_visibility)
274 def GetSignupPageInfo(
275 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
276 ) -> public_pb2.GetSignupPageInfoRes:
277 return _get_signup_page_info(session)
279 def GetVolunteers(
280 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
281 ) -> public_pb2.GetVolunteersRes:
282 return _get_volunteers(session)
284 def GetDonationStats(
285 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
286 ) -> public_pb2.GetDonationStatsRes:
287 return _get_donation_stats(session)