Coverage for app / backend / src / couchers / context.py: 81%
91 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-19 14:14 +0000
1from typing import NoReturn, cast
3import grpc
5from couchers.i18n import LocalizationContext
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 localization: LocalizationContext,
70 sofa: str | None = None,
71 ):
72 """Don't ever construct directly, always use the `make_*_context_` functions!"""
73 self._grpc_context = grpc_context
74 self._user_id = user_id
75 self._is_api_key = is_api_key
76 self.__token = token
77 self.__localization = localization
78 self._sofa = sofa
79 self.__is_interactive = is_interactive
80 self.__logged_in = self._user_id is not None
81 self.__cookies: list[str] = []
83 if self.__is_interactive:
84 if not self._grpc_context: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 raise ValueError("Tried to construct interactive context without grpc context")
86 self.__headers = dict(self._grpc_context.invocation_metadata())
88 if self.__logged_in:
89 if not self._user_id: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 raise ValueError("Invalid state, logged in but missing user_id")
92 def __verify_interactive(self) -> None:
93 if not self.__is_interactive: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise NonInteractiveContextException("Called an interactive context function in non-interactive context")
96 def __verify_logged_in(self) -> None:
97 if not self.__logged_in: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 raise NotLoggedInContextException("Called a logged-in function from logged-out context")
100 def is_logged_in(self) -> bool:
101 return self.__logged_in
103 def is_logged_out(self) -> bool:
104 return not self.__logged_in
106 def abort(self, status_code: grpc.StatusCode, error_message: str) -> NoReturn:
107 """
108 Raises an error that's returned to the user
109 """
110 if not self.__is_interactive: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true
111 raise NonInteractiveAbortException(status_code, error_message)
112 else:
113 context = cast(grpc.ServicerContext, self._grpc_context)
114 context.abort(status_code, error_message)
116 def abort_with_error_code(
117 self, status_code: grpc.StatusCode, error_message_id: str, *, substitutions: dict[str, str | int] | None = None
118 ) -> NoReturn:
119 """
120 Raises an error that's returned to the user, but error_message_id should be an entry from translateable errors
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_id)
124 else:
125 context = cast(grpc.ServicerContext, self._grpc_context)
126 # Get the translated error message using the user's language preference
127 error_message = self.localization.localize_string(f"errors.{error_message_id}", substitutions=substitutions)
128 context.abort(status_code, error_message)
130 def set_cookies(self, cookies: list[str]) -> None:
131 """
132 Sets a list of HTTP cookies
133 """
134 self.__verify_interactive()
135 self.__cookies += cookies
137 def _send_cookies(self) -> None:
138 data = tuple([("set-cookie", cookie) for cookie in self.__cookies])
139 self._grpc_context.send_initial_metadata(data) # type: ignore[union-attr]
141 @property
142 def headers(self) -> dict[str, str | bytes]:
143 """
144 Gets a list of HTTP headers for the requests
145 """
146 self.__verify_interactive()
147 return self.__headers
149 @property
150 def user_id(self) -> int:
151 """
152 Returns the user ID of the currently logged-in user, if available
153 """
154 self.__verify_logged_in()
155 return cast(int, self._user_id)
157 @property
158 def is_api_key(self) -> bool:
159 """
160 Returns whether the API call was done with an API key or not, if available
161 """
162 self.__verify_logged_in()
163 return cast(bool, self._is_api_key)
165 @property
166 def token(self) -> str:
167 """
168 Returns the token (session cookie/api key) of the current session, if available
169 """
170 self.__verify_interactive()
171 self.__verify_logged_in()
172 return cast(str, self.__token)
174 @property
175 def localization(self) -> LocalizationContext:
176 return self.__localization
179def make_interactive_context(
180 grpc_context: grpc.ServicerContext,
181 user_id: int | None,
182 is_api_key: bool,
183 token: str | None,
184 localization: LocalizationContext,
185 sofa: str | None = None,
186) -> CouchersContext:
187 return CouchersContext(
188 is_interactive=True,
189 grpc_context=grpc_context,
190 user_id=user_id,
191 is_api_key=is_api_key,
192 token=token,
193 localization=localization,
194 sofa=sofa,
195 )
198def make_one_off_interactive_user_context(couchers_context: CouchersContext, user_id: int) -> CouchersContext:
199 return CouchersContext(
200 is_interactive=True,
201 grpc_context=couchers_context._grpc_context,
202 user_id=user_id,
203 is_api_key=None,
204 token=None,
205 localization=couchers_context.localization,
206 )
209def make_media_context(grpc_context: grpc.ServicerContext) -> CouchersContext:
210 return CouchersContext(
211 is_interactive=True,
212 user_id=None,
213 is_api_key=False,
214 grpc_context=grpc_context,
215 token=None,
216 localization=LocalizationContext.en_utc(),
217 )
220def make_background_user_context(user_id: int, localization: LocalizationContext | None = None) -> CouchersContext:
221 return CouchersContext(
222 is_interactive=False,
223 user_id=user_id,
224 is_api_key=None,
225 grpc_context=None,
226 token=None,
227 localization=localization or LocalizationContext.en_utc(),
228 )
231def make_logged_out_context(localization: LocalizationContext) -> CouchersContext:
232 return CouchersContext(
233 user_id=None,
234 is_interactive=False,
235 is_api_key=None,
236 grpc_context=None,
237 token=None,
238 localization=localization,
239 )