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.metrics import grpc_in_flight_gauge, start_worker_resource_sampler 

10from couchers.middleware.interceptors import ( 

11 CouchersMiddlewareInterceptor, 

12 ErrorSanitizationInterceptor, 

13 OTelInterceptor, 

14) 

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 screens_pb2_grpc, 

45 search_pb2_grpc, 

46 stripe_pb2_grpc, 

47 threads_pb2_grpc, 

48) 

49from couchers.servicers.account import Account, Iris 

50from couchers.servicers.admin import Admin 

51from couchers.servicers.api import API 

52from couchers.servicers.auth import Auth 

53from couchers.servicers.blocking import Blocking 

54from couchers.servicers.bugs import Bugs 

55from couchers.servicers.communities import Communities 

56from couchers.servicers.conversations import Conversations 

57from couchers.servicers.discussions import Discussions 

58from couchers.servicers.donations import Donations, Stripe 

59from couchers.servicers.editor import Editor 

60from couchers.servicers.events import Events 

61from couchers.servicers.galleries import Galleries 

62from couchers.servicers.gis import GIS 

63from couchers.servicers.groups import Groups 

64from couchers.servicers.jail import Jail 

65from couchers.servicers.media import Media, get_media_auth_interceptor 

66from couchers.servicers.moderation import Moderation 

67from couchers.servicers.notifications import Notifications 

68from couchers.servicers.pages import Pages 

69from couchers.servicers.postal_verification import PostalVerification 

70from couchers.servicers.public import Public 

71from couchers.servicers.public_trips import PublicTrips 

72from couchers.servicers.references import References 

73from couchers.servicers.reporting import Reporting 

74from couchers.servicers.requests import Requests 

75from couchers.servicers.resources import Resources 

76from couchers.servicers.screens import Screens 

77from couchers.servicers.search import Search 

78from couchers.servicers.threads import Threads 

79 

80 

81class _InstrumentedThreadPoolExecutor(futures.ThreadPoolExecutor): 

82 # gRPC submits one task per RPC 

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

84 grpc_in_flight_gauge.inc() 

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

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

87 return future 

88 

89 

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

91 executor = _InstrumentedThreadPoolExecutor(SERVER_THREADS) 

92 server = grpc.server( 

93 executor, 

94 interceptors=[ 

95 ErrorSanitizationInterceptor(), 

96 OTelInterceptor(), 

97 CouchersMiddlewareInterceptor(), 

98 ], 

99 ) 

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

101 start_worker_resource_sampler(executor, _get_base_engine()) 

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

103 

104 account_pb2_grpc.add_AccountServicer_to_server(Account(), server) 

105 admin_pb2_grpc.add_AdminServicer_to_server(Admin(), server) 

106 api_pb2_grpc.add_APIServicer_to_server(API(), server) 

107 moderation_pb2_grpc.add_ModerationServicer_to_server(Moderation(), server) 

108 auth_pb2_grpc.add_AuthServicer_to_server(Auth(), server) 

109 blocking_pb2_grpc.add_BlockingServicer_to_server(Blocking(), server) 

110 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), server) 

111 communities_pb2_grpc.add_CommunitiesServicer_to_server(Communities(), server) 

112 conversations_pb2_grpc.add_ConversationsServicer_to_server(Conversations(), server) 

113 discussions_pb2_grpc.add_DiscussionsServicer_to_server(Discussions(), server) 

114 donations_pb2_grpc.add_DonationsServicer_to_server(Donations(), server) 

115 editor_pb2_grpc.add_EditorServicer_to_server(Editor(), server) 

116 events_pb2_grpc.add_EventsServicer_to_server(Events(), server) 

117 galleries_pb2_grpc.add_GalleriesServicer_to_server(Galleries(), server) 

118 gis_pb2_grpc.add_GISServicer_to_server(GIS(), server) 

119 groups_pb2_grpc.add_GroupsServicer_to_server(Groups(), server) 

120 iris_pb2_grpc.add_IrisServicer_to_server(Iris(), server) 

121 jail_pb2_grpc.add_JailServicer_to_server(Jail(), server) 

122 notifications_pb2_grpc.add_NotificationsServicer_to_server(Notifications(), server) 

123 pages_pb2_grpc.add_PagesServicer_to_server(Pages(), server) 

124 postal_verification_pb2_grpc.add_PostalVerificationServicer_to_server(PostalVerification(), server) 

125 public_pb2_grpc.add_PublicServicer_to_server(Public(), server) 

126 public_trips_pb2_grpc.add_PublicTripsServicer_to_server(PublicTrips(), server) 

127 references_pb2_grpc.add_ReferencesServicer_to_server(References(), server) 

128 reporting_pb2_grpc.add_ReportingServicer_to_server(Reporting(), server) 

129 requests_pb2_grpc.add_RequestsServicer_to_server(Requests(), server) 

130 resources_pb2_grpc.add_ResourcesServicer_to_server(Resources(), server) 

131 screens_pb2_grpc.add_ScreensServicer_to_server(Screens(), server) 

132 search_pb2_grpc.add_SearchServicer_to_server(Search(), server) 

133 stripe_pb2_grpc.add_StripeServicer_to_server(Stripe(), server) 

134 threads_pb2_grpc.add_ThreadsServicer_to_server(Threads(), server) 

135 return server 

136 

137 

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

139 media_server = grpc.server( 

140 futures.ThreadPoolExecutor(threads), 

141 interceptors=[ 

142 get_media_auth_interceptor(config.MEDIA_SERVER_BEARER_TOKEN), 

143 ], 

144 ) 

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

146 media_pb2_grpc.add_MediaServicer_to_server(Media(), media_server) 

147 return media_server