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