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