Coverage for app/backend/src/couchers/server.py: 93%

1from concurrent import futures 

2from typing import Any 

3 

4import grpc 

5 

6from couchers.config import config 

7from couchers.constants import SERVER_THREADS 

8from couchers.db import _get_base_engine 

9from couchers.interceptors import ( 

10 CouchersMiddlewareInterceptor, 

11 ErrorSanitizationInterceptor, 

12 OTelInterceptor, 

13) 

14from couchers.metrics import grpc_in_flight_gauge, start_worker_resource_sampler 

15from couchers.proto import ( 

16 account_pb2_grpc, 

17 admin_pb2_grpc, 

18 api_pb2_grpc, 

19 auth_pb2_grpc, 

20 blocking_pb2_grpc, 

21 bugs_pb2_grpc, 

22 communities_pb2_grpc, 

23 conversations_pb2_grpc, 

24 discussions_pb2_grpc, 

25 donations_pb2_grpc, 

26 editor_pb2_grpc, 

27 events_pb2_grpc, 

28 galleries_pb2_grpc, 

29 gis_pb2_grpc, 

30 groups_pb2_grpc, 

31 iris_pb2_grpc, 

32 jail_pb2_grpc, 

33 media_pb2_grpc, 

34 moderation_pb2_grpc, 

35 notifications_pb2_grpc, 

36 pages_pb2_grpc, 

37 postal_verification_pb2_grpc, 

38 public_pb2_grpc, 

39 public_trips_pb2_grpc, 

40 references_pb2_grpc, 

41 reporting_pb2_grpc, 

42 requests_pb2_grpc, 

43 resources_pb2_grpc, 

44 search_pb2_grpc, 

45 stripe_pb2_grpc, 

46 threads_pb2_grpc, 

47) 

48from couchers.servicers.account import Account, Iris 

49from couchers.servicers.admin import Admin 

50from couchers.servicers.api import API 

51from couchers.servicers.auth import Auth 

52from couchers.servicers.blocking import Blocking 

53from couchers.servicers.bugs import Bugs 

54from couchers.servicers.communities import Communities 

55from couchers.servicers.conversations import Conversations 

56from couchers.servicers.discussions import Discussions 

57from couchers.servicers.donations import Donations, Stripe 

58from couchers.servicers.editor import Editor 

59from couchers.servicers.events import Events 

60from couchers.servicers.galleries import Galleries 

61from couchers.servicers.gis import GIS 

62from couchers.servicers.groups import Groups 

63from couchers.servicers.jail import Jail 

64from couchers.servicers.media import Media, get_media_auth_interceptor 

65from couchers.servicers.moderation import Moderation 

66from couchers.servicers.notifications import Notifications 

67from couchers.servicers.pages import Pages 

68from couchers.servicers.postal_verification import PostalVerification 

69from couchers.servicers.public import Public 

70from couchers.servicers.public_trips import PublicTrips 

71from couchers.servicers.references import References 

72from couchers.servicers.reporting import Reporting 

73from couchers.servicers.requests import Requests 

74from couchers.servicers.resources import Resources 

75from couchers.servicers.search import Search 

76from couchers.servicers.threads import Threads 

77 

78 

79class _InstrumentedThreadPoolExecutor(futures.ThreadPoolExecutor): 

80 # gRPC submits one task per RPC 

81 def submit(self, fn: Any, /, *args: Any, **kwargs: Any) -> futures.Future[Any]: 

82 grpc_in_flight_gauge.inc() 

83 future = super().submit(fn, *args, **kwargs) 

84 future.add_done_callback(lambda _: grpc_in_flight_gauge.dec()) 

85 return future 

86 

87 

88def create_main_server(port: int, start_resource_sampler: bool = False) -> grpc.Server: 

89 executor = _InstrumentedThreadPoolExecutor(SERVER_THREADS) 

90 server = grpc.server( 

91 executor, 

92 interceptors=[ 

93 ErrorSanitizationInterceptor(), 

94 OTelInterceptor(), 

95 CouchersMiddlewareInterceptor(), 

96 ], 

97 ) 

98 if start_resource_sampler: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 start_worker_resource_sampler(executor, _get_base_engine()) 

100 server.add_insecure_port(f"[::]:{port}") 

101 

102 account_pb2_grpc.add_AccountServicer_to_server(Account(), server) 

103 admin_pb2_grpc.add_AdminServicer_to_server(Admin(), server) 

104 api_pb2_grpc.add_APIServicer_to_server(API(), server) 

105 moderation_pb2_grpc.add_ModerationServicer_to_server(Moderation(), server) 

106 auth_pb2_grpc.add_AuthServicer_to_server(Auth(), server) 

107 blocking_pb2_grpc.add_BlockingServicer_to_server(Blocking(), server) 

108 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), server) 

109 communities_pb2_grpc.add_CommunitiesServicer_to_server(Communities(), server) 

110 conversations_pb2_grpc.add_ConversationsServicer_to_server(Conversations(), server) 

111 discussions_pb2_grpc.add_DiscussionsServicer_to_server(Discussions(), server) 

112 donations_pb2_grpc.add_DonationsServicer_to_server(Donations(), server) 

113 editor_pb2_grpc.add_EditorServicer_to_server(Editor(), server) 

114 events_pb2_grpc.add_EventsServicer_to_server(Events(), server) 

115 galleries_pb2_grpc.add_GalleriesServicer_to_server(Galleries(), server) 

116 gis_pb2_grpc.add_GISServicer_to_server(GIS(), server) 

117 groups_pb2_grpc.add_GroupsServicer_to_server(Groups(), server) 

118 iris_pb2_grpc.add_IrisServicer_to_server(Iris(), server) 

119 jail_pb2_grpc.add_JailServicer_to_server(Jail(), server) 

120 notifications_pb2_grpc.add_NotificationsServicer_to_server(Notifications(), server) 

121 pages_pb2_grpc.add_PagesServicer_to_server(Pages(), server) 

122 postal_verification_pb2_grpc.add_PostalVerificationServicer_to_server(PostalVerification(), server) 

123 public_pb2_grpc.add_PublicServicer_to_server(Public(), server) 

124 public_trips_pb2_grpc.add_PublicTripsServicer_to_server(PublicTrips(), server) 

125 references_pb2_grpc.add_ReferencesServicer_to_server(References(), server) 

126 reporting_pb2_grpc.add_ReportingServicer_to_server(Reporting(), server) 

127 requests_pb2_grpc.add_RequestsServicer_to_server(Requests(), server) 

128 resources_pb2_grpc.add_ResourcesServicer_to_server(Resources(), server) 

129 search_pb2_grpc.add_SearchServicer_to_server(Search(), server) 

130 stripe_pb2_grpc.add_StripeServicer_to_server(Stripe(), server) 

131 threads_pb2_grpc.add_ThreadsServicer_to_server(Threads(), server) 

132 return server 

133 

134 

135def create_media_server(port: int, threads: int = 8) -> grpc.Server: 

136 media_server = grpc.server( 

137 futures.ThreadPoolExecutor(threads), 

138 interceptors=[ 

139 get_media_auth_interceptor(config.MEDIA_SERVER_BEARER_TOKEN), 

140 ], 

141 ) 

142 media_server.add_insecure_port(f"[::]:{port}") 

143 media_pb2_grpc.add_MediaServicer_to_server(Media(), media_server) 

144 return media_server