Coverage for src/couchers/servicers/public.py: 63%

1import logging 

2 

3import grpc 

4from cachetools import TTLCache, cached 

5from sqlalchemy.sql import func, union_all 

6 

7from couchers import errors, urls 

8from couchers.materialized_views import LiteUser 

9from couchers.models import Cluster, Node, ProfilePublicVisibility, Reference, User, Volunteer 

10from couchers.resources import get_static_badge_dict 

11from couchers.servicers.account import _format_volunteer_link 

12from couchers.servicers.api import fluency2api, hostingstatus2api, meetupstatus2api, user_model_to_pb 

13from couchers.servicers.gis import _statement_to_geojson_response 

14from couchers.sql import couchers_select as select 

15from couchers.utils import Timestamp_from_datetime, make_logged_out_context 

16from proto import api_pb2, public_pb2, public_pb2_grpc 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21@cached(cache=TTLCache(maxsize=1, ttl=600), key=lambda _: None) 

22def _get_public_users(session): 

23 with_geom = ( 

24 select(User.username, User.geom) 

25 .where(User.is_visible) 

26 .where(User.public_visibility != ProfilePublicVisibility.nothing) 

27 .where(User.public_visibility != ProfilePublicVisibility.map_only) 

28 ) 

29 

30 without_geom = ( 

31 select(None, User.randomized_geom) 

32 .where(User.is_visible) 

33 .where(User.randomized_geom != None) 

34 .where(User.public_visibility == ProfilePublicVisibility.map_only) 

35 ) 

36 return _statement_to_geojson_response(session, union_all(with_geom, without_geom)) 

37 

38 

39@cached(cache=TTLCache(maxsize=1, ttl=60), key=lambda _: None) 

40def _get_signup_page_info(session): 

41 # last user who signed up 

42 last_signup, geom = session.execute( 

43 select(User.joined, User.geom).where(User.is_visible).order_by(User.id.desc()).limit(1) 

44 ).one_or_none() 

45 

46 communities = ( 

47 session.execute( 

48 select(Cluster.name) 

49 .join(Node, Node.id == Cluster.parent_node_id) 

50 .where(Cluster.is_official_cluster) 

51 .where(func.ST_Contains(Node.geom, geom)) 

52 .order_by(Cluster.id.asc()) 

53 ) 

54 .scalars() 

55 .all() 

56 ) 

57 

58 if len(communities) <= 1: 

59 # either no community or just global community 

60 last_location = "The World" 

61 elif len(communities) == 3: 

62 # probably global, continent, region, so let's just return the region 

63 last_location = communities[-1] 

64 else: 

65 # probably global, continent, region, city 

66 last_location = f"{communities[-1]}, {communities[-2]}" 

67 

68 user_count = session.execute(select(func.count()).select_from(User).where(User.is_visible)).scalar_one() 

69 

70 return public_pb2.GetSignupPageInfoRes( 

71 last_signup=Timestamp_from_datetime(last_signup.replace(second=0, microsecond=0)), 

72 last_location=last_location, 

73 user_count=user_count, 

74 ) 

75 

76 

77@cached(cache=TTLCache(maxsize=1, ttl=60), key=lambda _: None) 

78def _get_volunteers(session): 

79 volunteers = session.execute( 

80 select(Volunteer, LiteUser) 

81 .join(LiteUser, LiteUser.id == Volunteer.user_id) 

82 .where(LiteUser.is_visible) 

83 .where(Volunteer.show_on_team_page) 

84 .order_by( 

85 Volunteer.sort_key.asc().nulls_last(), 

86 Volunteer.stopped_volunteering.desc().nulls_first(), 

87 Volunteer.started_volunteering.asc(), 

88 ) 

89 ).all() 

90 

91 board_members = set(get_static_badge_dict()["board_member"]) 

92 

93 def format_volunteer(volunteer, lite_user): 

94 if volunteer.link_type: 

95 link_type = volunteer.link_type 

96 link_text = volunteer.link_text 

97 link_url = volunteer.link_url 

98 else: 

99 link_type = "couchers" 

100 link_text = f"@{lite_user.username}" 

101 link_url = urls.user_link(username=lite_user.username) 

102 

103 return public_pb2.Volunteer( 

104 name=volunteer.display_name or lite_user.name, 

105 username=lite_user.username, 

106 is_board_member=lite_user.id in board_members, 

107 role=volunteer.role, 

108 location=volunteer.display_location or lite_user.city, 

109 img=urls.media_url(filename=lite_user.avatar_filename, size="thumbnail") 

110 if lite_user.avatar_filename 

111 else None, 

112 **_format_volunteer_link(volunteer, lite_user.username), 

113 ) 

114 

115 return public_pb2.GetVolunteersRes( 

116 current_volunteers=[ 

117 format_volunteer(volunteer, lite_user) 

118 for volunteer, lite_user in volunteers 

119 if volunteer.stopped_volunteering is None 

120 ], 

121 past_volunteers=[ 

122 format_volunteer(volunteer, lite_user) 

123 for volunteer, lite_user in volunteers 

124 if volunteer.stopped_volunteering is not None 

125 ], 

126 ) 

127 

128 

129class Public(public_pb2_grpc.PublicServicer): 

130 """ 

131 Public (logged out) APIs for getting public info 

132 """ 

133 

134 def GetPublicUsers(self, request, context, session): 

135 return _get_public_users(session) 

136 

137 def GetPublicUser(self, request, context, session): 

138 user = session.execute( 

139 select(User) 

140 .where(User.is_visible) 

141 .where(User.username == request.user) 

142 .where( 

143 User.public_visibility.in_( 

144 [ProfilePublicVisibility.limited, ProfilePublicVisibility.most, ProfilePublicVisibility.full] 

145 ) 

146 ) 

147 ).scalar_one_or_none() 

148 

149 if not user: 

150 context.abort(grpc.StatusCode.NOT_FOUND, errors.USER_NOT_FOUND) 

151 

152 if user.public_visibility == ProfilePublicVisibility.full: 

153 return public_pb2.GetPublicUserRes(full_user=user_model_to_pb(user, session, make_logged_out_context())) 

154 

155 num_references = session.execute( 

156 select(func.count()) 

157 .select_from(Reference) 

158 .join(User, User.id == Reference.from_user_id) 

159 .where(User.is_visible) 

160 .where(Reference.to_user_id == user.id) 

161 ).scalar_one() 

162 

163 if user.public_visibility == ProfilePublicVisibility.limited: 

164 return public_pb2.GetPublicUserRes( 

165 limited_user=public_pb2.LimitedUser( 

166 username=user.username, 

167 name=user.name, 

168 city=user.city, 

169 hometown=user.hometown, 

170 num_references=num_references, 

171 joined=Timestamp_from_datetime(user.display_joined), 

172 hosting_status=hostingstatus2api[user.hosting_status], 

173 meetup_status=meetupstatus2api[user.meetup_status], 

174 badges=[badge.badge_id for badge in user.badges], 

175 ) 

176 ) 

177 

178 if user.public_visibility == ProfilePublicVisibility.most: 

179 return public_pb2.GetPublicUserRes( 

180 most_user=public_pb2.MostUser( 

181 username=user.username, 

182 name=user.name, 

183 city=user.city, 

184 hometown=user.hometown, 

185 timezone=user.timezone, 

186 num_references=num_references, 

187 gender=user.gender, 

188 pronouns=user.pronouns, 

189 age=user.age, 

190 joined=Timestamp_from_datetime(user.display_joined), 

191 last_active=Timestamp_from_datetime(user.display_last_active), 

192 hosting_status=hostingstatus2api[user.hosting_status], 

193 meetup_status=meetupstatus2api[user.meetup_status], 

194 occupation=user.occupation, 

195 education=user.education, 

196 about_me=user.about_me, 

197 things_i_like=user.things_i_like, 

198 language_abilities=[ 

199 api_pb2.LanguageAbility(code=ability.language_code, fluency=fluency2api[ability.fluency]) 

200 for ability in user.language_abilities 

201 ], 

202 regions_visited=[region.code for region in user.regions_visited], 

203 regions_lived=[region.code for region in user.regions_lived], 

204 avatar_url=user.avatar.full_url if user.avatar else None, 

205 avatar_thumbnail_url=user.avatar.thumbnail_url if user.avatar else None, 

206 badges=[badge.badge_id for badge in user.badges], 

207 ) 

208 ) 

209 

210 def GetSignupPageInfo(self, request, context, session): 

211 return _get_signup_page_info(session) 

212 

213 def GetVolunteers(self, request, context, session): 

214 return _get_volunteers(session)