Coverage for app / backend / src / couchers / servicers / public.py: 91%
75 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-03 06:18 +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
9from sqlalchemy.sql import func, union_all
11from couchers import urls
12from couchers.constants import DONATION_GOAL_USD, DONATION_OFFSET_USD
13from couchers.context import CouchersContext, make_logged_out_context
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.utils import Timestamp_from_datetime, not_none, now
33logger = logging.getLogger(__name__)
36def format_volunteer_link(volunteer: Volunteer, username: str) -> dict[str, str]:
37 """Format volunteer link information into a dict with link_type, link_text, and link_url."""
38 if volunteer.link_type:
39 return dict(
40 link_type=volunteer.link_type,
41 link_text=not_none(volunteer.link_text),
42 link_url=not_none(volunteer.link_url),
43 )
44 else:
45 return dict(
46 link_type="couchers",
47 link_text=f"@{username}",
48 link_url=urls.user_link(username=username),
49 )
52@cached(cache=TTLCache(maxsize=1, ttl=600), key=lambda _: None, lock=threading.Lock())
53def _get_public_users(session: Session) -> httpbody_pb2.HttpBody:
54 with_geom = (
55 select(User.username, User.geom)
56 .where(User.is_visible)
57 .where(User.public_visibility != ProfilePublicVisibility.nothing)
58 .where(User.public_visibility != ProfilePublicVisibility.map_only)
59 )
61 without_geom = (
62 select(null(), User.randomized_geom)
63 .where(User.is_visible)
64 .where(User.randomized_geom != None)
65 .where(User.public_visibility == ProfilePublicVisibility.map_only)
66 )
67 return _statement_to_geojson_response(session, union_all(with_geom, without_geom))
70@cached(cache=TTLCache(maxsize=1, ttl=60), key=lambda _: None, lock=threading.Lock())
71def _get_signup_page_info(session: Session) -> public_pb2.GetSignupPageInfoRes:
72 # last user who signed up
73 last_signup, geom = session.execute(
74 select(User.joined, User.geom).where(User.is_visible).order_by(User.id.desc()).limit(1)
75 ).one()
77 communities = (
78 session.execute(
79 select(Cluster.name)
80 .join(Node, Node.id == Cluster.parent_node_id)
81 .where(Cluster.is_official_cluster)
82 .where(func.ST_Contains(Node.geom, geom))
83 .order_by(Cluster.id.asc())
84 )
85 .scalars()
86 .all()
87 )
89 if len(communities) <= 1: 89 ↛ 92line 89 didn't jump to line 92 because the condition on line 89 was always true
90 # either no community or just global community
91 last_location = "The World"
92 elif len(communities) == 3:
93 # probably global, continent, region, so let's just return the region
94 last_location = communities[-1]
95 else:
96 # probably global, continent, region, city
97 last_location = f"{communities[-1]}, {communities[-2]}"
99 user_count = session.execute(select(func.count()).select_from(User).where(User.is_visible)).scalar_one()
101 return public_pb2.GetSignupPageInfoRes(
102 last_signup=Timestamp_from_datetime(last_signup.replace(second=0, microsecond=0)),
103 last_location=last_location,
104 user_count=user_count,
105 )
108@cached(cache=TTLCache(maxsize=1, ttl=60), key=lambda _: None, lock=threading.Lock())
109def _get_donation_stats(session: Session) -> public_pb2.GetDonationStatsRes:
110 """Get year-to-date donation statistics, excluding merch purchases."""
111 start_of_year = now().replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
113 total_donated = session.execute(
114 select(func.coalesce(func.sum(Invoice.amount), 0))
115 .where(Invoice.invoice_type == InvoiceType.on_platform)
116 .where(Invoice.created >= start_of_year)
117 ).scalar_one()
119 return public_pb2.GetDonationStatsRes(
120 total_donated_ytd=max(int(total_donated - DONATION_OFFSET_USD), 0),
121 goal=DONATION_GOAL_USD,
122 )
125@cached(cache=TTLCache(maxsize=1, ttl=5), key=lambda _: None, lock=threading.Lock())
126def _get_volunteers(session: Session) -> public_pb2.GetVolunteersRes:
127 volunteers = session.execute(
128 select(Volunteer, LiteUser)
129 .join(LiteUser, LiteUser.id == Volunteer.user_id)
130 .where(LiteUser.is_visible)
131 .where(Volunteer.show_on_team_page)
132 .order_by(
133 Volunteer.sort_key.asc().nulls_last(),
134 Volunteer.stopped_volunteering.desc().nulls_first(),
135 Volunteer.started_volunteering.asc(),
136 )
137 ).all()
139 board_members = set(get_static_badge_dict()["board_member"])
141 def format_volunteer(volunteer: Volunteer, lite_user: LiteUser) -> public_pb2.Volunteer:
142 return public_pb2.Volunteer(
143 name=volunteer.display_name or lite_user.name,
144 username=lite_user.username,
145 is_board_member=lite_user.id in board_members,
146 role=volunteer.role,
147 location=volunteer.display_location or lite_user.city,
148 img=urls.media_url(filename=lite_user.avatar_filename, size="thumbnail")
149 if lite_user.avatar_filename
150 else None,
151 **format_volunteer_link(volunteer, lite_user.username),
152 )
154 return public_pb2.GetVolunteersRes(
155 current_volunteers=[
156 format_volunteer(volunteer, lite_user)
157 for volunteer, lite_user in volunteers
158 if volunteer.stopped_volunteering is None
159 ],
160 past_volunteers=[
161 format_volunteer(volunteer, lite_user)
162 for volunteer, lite_user in volunteers
163 if volunteer.stopped_volunteering is not None
164 ],
165 )
168class Public(public_pb2_grpc.PublicServicer):
169 """
170 Public (logged-out) APIs for getting public info
171 """
173 def GetPublicUsers(
174 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
175 ) -> httpbody_pb2.HttpBody:
176 return _get_public_users(session)
178 def GetPublicUser(
179 self, request: public_pb2.GetPublicUserReq, context: CouchersContext, session: Session
180 ) -> public_pb2.GetPublicUserRes:
181 user = session.execute(
182 select(User)
183 .where(User.is_visible)
184 .where(User.username == request.user)
185 .where(
186 User.public_visibility.in_(
187 [ProfilePublicVisibility.limited, ProfilePublicVisibility.most, ProfilePublicVisibility.full]
188 )
189 )
190 ).scalar_one_or_none()
192 if not user:
193 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
195 if user.public_visibility == ProfilePublicVisibility.full:
196 return public_pb2.GetPublicUserRes(full_user=user_model_to_pb(user, session, make_logged_out_context()))
198 num_references = session.execute(
199 select(func.count())
200 .select_from(Reference)
201 .join(User, User.id == Reference.from_user_id)
202 .where(User.is_visible)
203 .where(Reference.to_user_id == user.id)
204 ).scalar_one()
206 if user.public_visibility == ProfilePublicVisibility.limited:
207 return public_pb2.GetPublicUserRes(
208 limited_user=public_pb2.LimitedUser(
209 username=user.username,
210 name=user.name,
211 city=user.city,
212 hometown=user.hometown,
213 num_references=num_references,
214 joined=Timestamp_from_datetime(user.display_joined),
215 hosting_status=hostingstatus2api[user.hosting_status],
216 meetup_status=meetupstatus2api[user.meetup_status],
217 badges=[badge.badge_id for badge in user.badges],
218 )
219 )
221 if user.public_visibility == ProfilePublicVisibility.most: 221 ↛ 254line 221 didn't jump to line 254 because the condition on line 221 was always true
222 avatar_upload = get_avatar_upload(session, user)
224 return public_pb2.GetPublicUserRes(
225 most_user=public_pb2.MostUser(
226 username=user.username,
227 name=user.name,
228 city=user.city,
229 hometown=user.hometown,
230 timezone=user.timezone,
231 num_references=num_references,
232 gender=user.gender,
233 pronouns=user.pronouns,
234 age=int(user.age),
235 joined=Timestamp_from_datetime(user.display_joined),
236 last_active=Timestamp_from_datetime(user.display_last_active),
237 hosting_status=hostingstatus2api[user.hosting_status],
238 meetup_status=meetupstatus2api[user.meetup_status],
239 occupation=user.occupation,
240 education=user.education,
241 about_me=user.about_me,
242 things_i_like=user.things_i_like,
243 language_abilities=[
244 api_pb2.LanguageAbility(code=ability.language_code, fluency=fluency2api[ability.fluency])
245 for ability in user.language_abilities
246 ],
247 regions_visited=[region.code for region in user.regions_visited],
248 regions_lived=[region.code for region in user.regions_lived],
249 avatar_url=avatar_upload.full_url if avatar_upload else None,
250 avatar_thumbnail_url=avatar_upload.thumbnail_url if avatar_upload else None,
251 badges=[badge.badge_id for badge in user.badges],
252 )
253 )
254 raise RuntimeError(user.public_visibility)
256 def GetSignupPageInfo(
257 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
258 ) -> public_pb2.GetSignupPageInfoRes:
259 return _get_signup_page_info(session)
261 def GetVolunteers(
262 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
263 ) -> public_pb2.GetVolunteersRes:
264 return _get_volunteers(session)
266 def GetDonationStats(
267 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
268 ) -> public_pb2.GetDonationStatsRes:
269 return _get_donation_stats(session)