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