Coverage for src/couchers/context.py: 83%
88 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-12-17 01:15 +0000
1from typing import NoReturn, cast
3import grpc
5from couchers.i18n.i18n import get_raw_translation_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:
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:
88 raise ValueError("Invalid state, logged in but missing user_id")
90 def __verify_interactive(self) -> None:
91 if not self.__is_interactive:
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:
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 get_localized_string(
102 self, component: str, message_id: str, *, substitutions: dict[str, str] | None = None
103 ) -> str:
104 """
105 Get a localized string using the user's language preference.
106 Falls back to the default language if no preference is set.
108 Args:
109 component: Component name (e.g., "errors")
110 message_id: 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 get_raw_translation_string(
117 self.__ui_language_preference, component, message_id, substitutions=substitutions
118 )
120 def abort(self, status_code: grpc.StatusCode, error_message: str) -> NoReturn:
121 """
122 Raises an error that's returned to the user
123 """
124 if not self.__is_interactive:
125 raise NonInteractiveAbortException(status_code, error_message)
126 else:
127 context = cast(grpc.ServicerContext, self._grpc_context)
128 context.abort(status_code, error_message)
130 def abort_with_error_code(
131 self, status_code: grpc.StatusCode, error_message_id: str, *, substitutions: dict[str, str] | None = None
132 ) -> NoReturn:
133 """
134 Raises an error that's returned to the user, but error_message_id should be an entry from translateable errors
135 """
136 if not self.__is_interactive:
137 raise NonInteractiveAbortException(status_code, error_message_id)
138 else:
139 context = cast(grpc.ServicerContext, self._grpc_context)
140 # Get the translated error message using the user's language preference
141 error_message = self.get_localized_string("errors", error_message_id, substitutions=substitutions)
142 context.abort(status_code, error_message)
144 def set_cookies(self, cookies: list[str]) -> None:
145 """
146 Sets a list of HTTP cookies
147 """
148 self.__verify_interactive()
149 self.__cookies += cookies
151 def _send_cookies(self) -> None:
152 data = tuple([("set-cookie", cookie) for cookie in self.__cookies])
153 self._grpc_context.send_initial_metadata(data) # type: ignore[union-attr]
155 @property
156 def headers(self) -> dict[str, str | bytes]:
157 """
158 Gets a list of HTTP headers for the requests
159 """
160 self.__verify_interactive()
161 return self.__headers
163 @property
164 def user_id(self) -> int:
165 """
166 Returns the user ID of the currently logged-in user, if available
167 """
168 self.__verify_logged_in()
169 return cast(int, self._user_id)
171 @property
172 def is_api_key(self) -> bool:
173 """
174 Returns whether the API call was done with an API key or not, if available
175 """
176 self.__verify_logged_in()
177 return cast(bool, self._is_api_key)
179 @property
180 def token(self) -> str:
181 """
182 Returns the token (session cookie/api key) of the current session, if available
183 """
184 self.__verify_interactive()
185 self.__verify_logged_in()
186 return cast(str, self.__token)
188 @property
189 def ui_language_preference(self) -> str | None:
190 return self.__ui_language_preference
193def make_interactive_context(
194 grpc_context: grpc.ServicerContext,
195 user_id: int | None,
196 is_api_key: bool,
197 token: str | None,
198 ui_language_preference: str | None,
199) -> CouchersContext:
200 return CouchersContext(
201 is_interactive=True,
202 grpc_context=grpc_context,
203 user_id=user_id,
204 is_api_key=is_api_key,
205 token=token,
206 ui_language_preference=ui_language_preference,
207 )
210def make_one_off_interactive_user_context(
211 couchers_context: CouchersContext,
212 user_id: int,
213) -> CouchersContext:
214 return CouchersContext(
215 is_interactive=True,
216 grpc_context=couchers_context._grpc_context,
217 user_id=user_id,
218 is_api_key=None,
219 token=None,
220 ui_language_preference=None,
221 )
224def make_media_context(grpc_context: grpc.ServicerContext) -> CouchersContext:
225 return CouchersContext(
226 is_interactive=True,
227 user_id=None,
228 is_api_key=False,
229 grpc_context=grpc_context,
230 token=None,
231 ui_language_preference=None,
232 )
235def make_background_user_context(user_id: int) -> CouchersContext:
236 return CouchersContext(
237 is_interactive=False,
238 user_id=user_id,
239 is_api_key=None,
240 grpc_context=None,
241 token=None,
242 ui_language_preference=None,
243 )