Coverage for app / backend / src / couchers / context.py: 81%
92 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-03 06:18 +0000
1from typing import NoReturn, cast
3import grpc
5from couchers.i18n.localize import localize_string
8class NonInteractiveContextException(Exception):
9 """If this exception is raised, it is a programming error"""
12class NotLoggedInContextException(Exception):
13 """If this exception is raised, it is a programming error"""
16class NonInteractiveAbortException(grpc.RpcError):
17 """This exception is raised in background processes when they call context.abort()"""
19 def __init__(self, code: grpc.StatusCode, details: str) -> None:
20 super().__init__(details)
21 self._code = code
22 self._details = details
24 def code(self) -> grpc.StatusCode:
25 return self._code
27 def details(self) -> str:
28 return self._details
30 def __str__(self) -> str:
31 return f"RPC aborted in non-interactive context, code: {self._code}, details: {self._details}"
34class CouchersContext:
35 """
36 The CouchersContext is passed to backend APIs and contains context about what context the function is running in,
37 such as information about the user the action is being taken for, etc.
39 This class contains a bunch of stuff, and there are different ways of invoking functionality, so there are different
40 types of contexts:
42 *Interactive, authenticated, authorized*: this is the main one, a user is logged in and calling the APIs manually.
44 *Interactive, authenticated, single-authorized*: this is a bit of an edge cases, sometimes users invoke functions
45 while not properly logged in, but they are still authorized to invoke some APIs. E.g. a "quick link" on an email
46 that contain signed URLs.
48 *Interactive, unauthenticated*: a public API is being called by a user that is not logged in.
50 *Non-interactive, authenticated*: we are calling an API or taking some action on behalf of a user in a background
51 task.
53 This context will complain a lot to make things work as intended.
55 Do not call the constructor directly, use the `make_*_context_` functions below.
57 You can safely call public methods, don't call methods whose names start with underscores unless you know what
58 you're doing!
59 """
61 def __init__(
62 self,
63 *,
64 is_interactive: bool,
65 grpc_context: grpc.ServicerContext | None,
66 user_id: int | None,
67 is_api_key: bool | None,
68 token: str | None,
69 ui_language_preference: str | None,
70 ):
71 """Don't ever construct directly, always use the `make_*_context_` functions!"""
72 self._grpc_context = grpc_context
73 self._user_id = user_id
74 self._is_api_key = is_api_key
75 self.__token = token
76 self.__ui_language_preference = ui_language_preference
77 self.__is_interactive = is_interactive
78 self.__logged_in = self._user_id is not None
79 self.__cookies: list[str] = []
81 if self.__is_interactive:
82 if not self._grpc_context: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise ValueError("Tried to construct interactive context without grpc context")
84 self.__headers = dict(self._grpc_context.invocation_metadata())
86 if self.__logged_in:
87 if not self._user_id: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true
88 raise ValueError("Invalid state, logged in but missing user_id")
90 def __verify_interactive(self) -> None:
91 if not self.__is_interactive: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 raise NonInteractiveContextException("Called an interactive context function in non-interactive context")
94 def __verify_logged_in(self) -> None:
95 if not self.__logged_in: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true
96 raise NotLoggedInContextException("Called a logged-in function from logged-out context")
98 def is_logged_in(self) -> bool:
99 return self.__logged_in
101 def is_logged_out(self) -> bool:
102 return not self.__logged_in
104 def get_localized_string(self, key: str, *, substitutions: dict[str, str | int] | None = None) -> str:
105 """
106 Get a localized string using the user's language preference.
107 Falls back to the default language if no preference is set.
109 Args:
110 key: The key for the specific string
111 substitutions: Dictionary of variable substitutions for the string (optional)
113 Returns:
114 The translated string with substitutions applied
115 """
116 return localize_string(self.__ui_language_preference, key, substitutions=substitutions)
118 def abort(self, status_code: grpc.StatusCode, error_message: str) -> NoReturn:
119 """
120 Raises an error that's returned to the user
121 """
122 if not self.__is_interactive: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 raise NonInteractiveAbortException(status_code, error_message)
124 else:
125 context = cast(grpc.ServicerContext, self._grpc_context)
126 context.abort(status_code, error_message)
128 def abort_with_error_code(
129 self, status_code: grpc.StatusCode, error_message_id: str, *, substitutions: dict[str, str | int] | None = None
130 ) -> NoReturn:
131 """
132 Raises an error that's returned to the user, but error_message_id should be an entry from translateable errors
133 """
134 if not self.__is_interactive: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 raise NonInteractiveAbortException(status_code, error_message_id)
136 else:
137 context = cast(grpc.ServicerContext, self._grpc_context)
138 # Get the translated error message using the user's language preference
139 error_message = self.get_localized_string(f"errors.{error_message_id}", substitutions=substitutions)
140 context.abort(status_code, error_message)
142 def set_cookies(self, cookies: list[str]) -> None:
143 """
144 Sets a list of HTTP cookies
145 """
146 self.__verify_interactive()
147 self.__cookies += cookies
149 def _send_cookies(self) -> None:
150 data = tuple([("set-cookie", cookie) for cookie in self.__cookies])
151 self._grpc_context.send_initial_metadata(data) # type: ignore[union-attr]
153 @property
154 def headers(self) -> dict[str, str | bytes]:
155 """
156 Gets a list of HTTP headers for the requests
157 """
158 self.__verify_interactive()
159 return self.__headers
161 @property
162 def user_id(self) -> int:
163 """
164 Returns the user ID of the currently logged-in user, if available
165 """
166 self.__verify_logged_in()
167 return cast(int, self._user_id)
169 @property
170 def is_api_key(self) -> bool:
171 """
172 Returns whether the API call was done with an API key or not, if available
173 """
174 self.__verify_logged_in()
175 return cast(bool, self._is_api_key)
177 @property
178 def token(self) -> str:
179 """
180 Returns the token (session cookie/api key) of the current session, if available
181 """
182 self.__verify_interactive()
183 self.__verify_logged_in()
184 return cast(str, self.__token)
186 @property
187 def ui_language_preference(self) -> str | None:
188 return self.__ui_language_preference
191def make_interactive_context(
192 grpc_context: grpc.ServicerContext,
193 user_id: int | None,
194 is_api_key: bool,
195 token: str | None,
196 ui_language_preference: str | None,
197) -> CouchersContext:
198 return CouchersContext(
199 is_interactive=True,
200 grpc_context=grpc_context,
201 user_id=user_id,
202 is_api_key=is_api_key,
203 token=token,
204 ui_language_preference=ui_language_preference,
205 )
208def make_one_off_interactive_user_context(
209 couchers_context: CouchersContext,
210 user_id: int,
211) -> CouchersContext:
212 return CouchersContext(
213 is_interactive=True,
214 grpc_context=couchers_context._grpc_context,
215 user_id=user_id,
216 is_api_key=None,
217 token=None,
218 ui_language_preference=None,
219 )
222def make_media_context(grpc_context: grpc.ServicerContext) -> CouchersContext:
223 return CouchersContext(
224 is_interactive=True,
225 user_id=None,
226 is_api_key=False,
227 grpc_context=grpc_context,
228 token=None,
229 ui_language_preference=None,
230 )
233def make_background_user_context(user_id: int) -> CouchersContext:
234 return CouchersContext(
235 is_interactive=False,
236 user_id=user_id,
237 is_api_key=None,
238 grpc_context=None,
239 token=None,
240 ui_language_preference=None,
241 )
244def make_logged_out_context() -> CouchersContext:
245 return CouchersContext(
246 user_id=None,
247 is_interactive=False,
248 is_api_key=None,
249 grpc_context=None,
250 token=None,
251 ui_language_preference=None,
252 )