Coverage for app/backend/src/couchers/context.py: 86%
122 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-21 09:29 +0000
1from typing import TYPE_CHECKING, NoReturn, cast
3import grpc
5from couchers import experimentation
6from couchers.i18n import LocalizationContext
7from couchers.i18n.locales import get_translation_component
9if TYPE_CHECKING:
10 from growthbook import GrowthBook
13class NonInteractiveContextException(Exception):
14 """If this exception is raised, it is a programming error"""
17class NotLoggedInContextException(Exception):
18 """If this exception is raised, it is a programming error"""
21class NonInteractiveAbortException(grpc.RpcError):
22 """This exception is raised in background processes when they call context.abort()"""
24 def __init__(self, code: grpc.StatusCode, details: str) -> None:
25 super().__init__(details)
26 self._code = code
27 self._details = details
29 def code(self) -> grpc.StatusCode:
30 return self._code
32 def details(self) -> str:
33 return self._details
35 def __str__(self) -> str:
36 return f"RPC aborted in non-interactive context, code: {self._code}, details: {self._details}"
39class CouchersContext:
40 """
41 The CouchersContext is passed to backend APIs and contains context about what context the function is running in,
42 such as information about the user the action is being taken for, etc.
44 This class contains a bunch of stuff, and there are different ways of invoking functionality, so there are different
45 types of contexts:
47 *Interactive, authenticated, authorized*: this is the main one, a user is logged in and calling the APIs manually.
49 *Interactive, authenticated, single-authorized*: this is a bit of an edge cases, sometimes users invoke functions
50 while not properly logged in, but they are still authorized to invoke some APIs. E.g. a "quick link" on an email
51 that contain signed URLs.
53 *Interactive, unauthenticated*: a public API is being called by a user that is not logged in.
55 *Non-interactive, authenticated*: we are calling an API or taking some action on behalf of a user in a background
56 task.
58 This context will complain a lot to make things work as intended.
60 Do not call the constructor directly, use the `make_*_context_` functions below.
62 You can safely call public methods, don't call methods whose names start with underscores unless you know what
63 you're doing!
64 """
66 def __init__(
67 self,
68 *,
69 is_interactive: bool,
70 grpc_context: grpc.ServicerContext | None,
71 user_id: int | None,
72 is_api_key: bool | None,
73 token: str | None,
74 localization: LocalizationContext,
75 sofa: str | None = None,
76 serialize_shadowed: bool,
77 ):
78 """Don't ever construct directly, always use the `make_*_context_` functions!"""
79 self._grpc_context = grpc_context
80 self._user_id = user_id
81 self._is_api_key = is_api_key
82 self.__token = token
83 self.__localization = localization
84 self._sofa = sofa
85 self.__serialize_shadowed = serialize_shadowed
86 self.__is_interactive = is_interactive
87 self.__logged_in = self._user_id is not None
88 self.__cookies: list[str] = []
89 self.__response_headers: list[tuple[str, str]] = []
90 self._growthbook: GrowthBook | None = None
92 if self.__is_interactive:
93 if not self._grpc_context: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise ValueError("Tried to construct interactive context without grpc context")
95 self.__headers = dict(self._grpc_context.invocation_metadata())
97 if self.__logged_in:
98 if not self._user_id: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 raise ValueError("Invalid state, logged in but missing user_id")
101 def __verify_interactive(self) -> None:
102 if not self.__is_interactive: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 raise NonInteractiveContextException("Called an interactive context function in non-interactive context")
105 def __verify_logged_in(self) -> None:
106 if not self.__logged_in: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 raise NotLoggedInContextException("Called a logged-in function from logged-out context")
109 def is_logged_in(self) -> bool:
110 return self.__logged_in
112 def is_logged_out(self) -> bool:
113 return not self.__logged_in
115 def abort(self, status_code: grpc.StatusCode, error_message: str) -> NoReturn:
116 """
117 Raises an error that's returned to the user
118 """
119 if not self.__is_interactive: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 raise NonInteractiveAbortException(status_code, error_message)
121 else:
122 context = cast(grpc.ServicerContext, self._grpc_context)
123 context.abort(status_code, error_message)
125 def abort_with_error_code(
126 self,
127 status_code: grpc.StatusCode,
128 error_message_id: str,
129 *,
130 substitutions: dict[str, str | int] | None = None,
131 ) -> NoReturn:
132 """
133 Raises an error that's returned to the user, but error_message_id should be an entry from translateable errors
135 error_message_id may be namespaced with a translation component, like i18next, e.g. "admin:object_not_found"
136 looks up "object_not_found" in the "admin" component (where admin/editor errors live). Without a prefix the
137 "main" component is used.
138 """
139 if not self.__is_interactive: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 raise NonInteractiveAbortException(status_code, error_message_id)
141 else:
142 context = cast(grpc.ServicerContext, self._grpc_context)
143 component, _, error_name = error_message_id.rpartition(":")
144 # Get the translated error message using the user's language preference
145 error_message = self.localization.localize_string(
146 f"errors.{error_name}",
147 i18next=get_translation_component(component or "main"),
148 substitutions=substitutions,
149 )
150 context.abort(status_code, error_message)
152 def set_cookies(self, cookies: list[str]) -> None:
153 """
154 Sets a list of HTTP cookies
155 """
156 self.__verify_interactive()
157 self.__cookies += cookies
159 def set_response_headers(self, headers: list[tuple[str, str]]) -> None:
160 """
161 Sets extra HTTP response headers (forwarded by Envoy as gRPC initial metadata)
162 """
163 self.__verify_interactive()
164 self.__response_headers += headers
166 def _send_cookies(self) -> None:
167 data = tuple([("set-cookie", cookie) for cookie in self.__cookies]) + tuple(self.__response_headers)
168 self._grpc_context.send_initial_metadata(data) # type: ignore[union-attr]
170 @property
171 def headers(self) -> dict[str, str | bytes]:
172 """
173 Gets a list of HTTP headers for the requests
174 """
175 self.__verify_interactive()
176 return self.__headers
178 def get_header(self, name: str) -> str | None:
179 self.__verify_interactive()
180 return cast(str | None, self.__headers.get(name))
182 @property
183 def user_id(self) -> int:
184 """
185 Returns the user ID of the currently logged-in user, if available
186 """
187 self.__verify_logged_in()
188 return cast(int, self._user_id)
190 @property
191 def is_api_key(self) -> bool:
192 """
193 Returns whether the API call was done with an API key or not, if available
194 """
195 self.__verify_logged_in()
196 return cast(bool, self._is_api_key)
198 @property
199 def token(self) -> str:
200 """
201 Returns the token (session cookie/api key) of the current session, if available
202 """
203 self.__verify_interactive()
204 self.__verify_logged_in()
205 return cast(str, self.__token)
207 @property
208 def localization(self) -> LocalizationContext:
209 return self.__localization
211 @property
212 def serialize_shadowed(self) -> bool:
213 return self.__serialize_shadowed
215 # Feature-flag evaluation methods mirror the OpenFeature evaluation API, evaluating for this
216 # context's user. The gating lives in experimentation; we just pass our cached per-request
217 # evaluator. The in-code default is honored even for flags not yet set up in GrowthBook.
218 def get_boolean_value(self, flag_key: str, default: bool) -> bool:
219 return experimentation._feature_value(flag_key, default, self._get_growthbook)
221 def get_string_value(self, flag_key: str, default: str) -> str:
222 return experimentation._feature_value(flag_key, default, self._get_growthbook)
224 def get_integer_value(self, flag_key: str, default: int) -> int:
225 return experimentation._feature_value(flag_key, default, self._get_growthbook)
227 def get_float_value(self, flag_key: str, default: float) -> float:
228 return experimentation._feature_value(flag_key, default, self._get_growthbook)
230 def get_object_value[T](self, flag_key: str, default: T) -> T:
231 return experimentation._feature_value(flag_key, default, self._get_growthbook)
233 def _get_growthbook(self) -> GrowthBook:
234 if self._growthbook is None:
235 # _user_id is None when logged out: evaluate anonymously, falling through to defaults.
236 self._growthbook = experimentation._create_evaluator(self._user_id)
237 return self._growthbook
240def make_interactive_context(
241 grpc_context: grpc.ServicerContext,
242 user_id: int | None,
243 is_api_key: bool,
244 token: str | None,
245 localization: LocalizationContext,
246 sofa: str | None = None,
247) -> CouchersContext:
248 return CouchersContext(
249 is_interactive=True,
250 grpc_context=grpc_context,
251 user_id=user_id,
252 is_api_key=is_api_key,
253 token=token,
254 localization=localization,
255 sofa=sofa,
256 serialize_shadowed=False,
257 )
260def make_one_off_interactive_user_context(couchers_context: CouchersContext, user_id: int) -> CouchersContext:
261 return CouchersContext(
262 is_interactive=True,
263 grpc_context=couchers_context._grpc_context,
264 user_id=user_id,
265 is_api_key=None,
266 token=None,
267 localization=couchers_context.localization,
268 serialize_shadowed=False,
269 )
272def make_media_context(grpc_context: grpc.ServicerContext) -> CouchersContext:
273 return CouchersContext(
274 is_interactive=True,
275 user_id=None,
276 is_api_key=False,
277 grpc_context=grpc_context,
278 token=None,
279 localization=LocalizationContext.en_utc(),
280 serialize_shadowed=False,
281 )
284def make_background_user_context(user_id: int, localization: LocalizationContext | None = None) -> CouchersContext:
285 return CouchersContext(
286 is_interactive=False,
287 user_id=user_id,
288 is_api_key=None,
289 grpc_context=None,
290 token=None,
291 localization=localization or LocalizationContext.en_utc(),
292 serialize_shadowed=False,
293 )
296def make_notification_user_context(user_id: int, localization: LocalizationContext | None = None) -> CouchersContext:
297 return CouchersContext(
298 is_interactive=False,
299 user_id=user_id,
300 is_api_key=None,
301 grpc_context=None,
302 token=None,
303 localization=localization or LocalizationContext.en_utc(),
304 serialize_shadowed=True,
305 )
308def make_logged_out_context(localization: LocalizationContext) -> CouchersContext:
309 return CouchersContext(
310 user_id=None,
311 is_interactive=False,
312 is_api_key=None,
313 grpc_context=None,
314 token=None,
315 localization=localization,
316 serialize_shadowed=False,
317 )