Coverage for src/couchers/context.py: 82%
76 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-08-28 14:55 +0000
1import grpc
4class NonInteractiveContextException(Exception):
5 """If this exception is raised it is a programming error"""
8class NotLoggedInContextException(Exception):
9 """If this exception is raised it is a programming error"""
12class NonInteractiveAbortException(grpc.RpcError):
13 """This exception is raised in background processes when they call context.abort()"""
15 def __init__(self, code, details):
16 super().__init__(details)
17 self._code = code
18 self._details = details
20 def code(self):
21 return self._code
23 def details(self):
24 return self._details
26 def __str__(self):
27 return f"RPC aborted in non-interactive context, code: {self._code}, details: {self._details}"
30class CouchersContext:
31 """
32 The CouchersContext is passed to backend APIs and contains context about what context the function is running in,
33 such as information about the user the action is being taken for, etc.
35 This class contains a bunch of stuff, and there are different ways of invoking functionality, so there are different
36 types of contexts:
38 *Interactive, authenticated, authorized*: this is the main one, a user is logged in and calling the APIs manually.
40 *Interactive, authenticated, single-authorized*: this is a bit of an edge cases, sometimes users invoke functions
41 while not properly logged in, but they are still authorized to invoke some APIs. E.g. a "quick link" on an email
42 that contain signed URLs.
44 *Interactive, unauthenticated*: a public API is being called by a user that is not logged in.
46 *Non-interactive, authenticated*: we are calling an API or taking some action on behalf of a user in a background
47 task.
49 This context will complain a lot to make things work as intended.
51 Do not call the constructor directly, use the `make_*_context_` functions below.
53 You can safely call public methods, don't call methods whose names start with underscores unless you know what
54 you're doing!
55 """
57 def __init__(
58 self,
59 *,
60 is_interactive: bool,
61 grpc_context: grpc.ServicerContext | None,
62 user_id: int | None,
63 is_api_key: bool | None,
64 token: str | None,
65 ui_language_preference: str | None,
66 ):
67 """Don't ever construct directly, always use the `make_*_context_` functions!"""
68 self._grpc_context = grpc_context
69 self._user_id = user_id
70 self._is_api_key = is_api_key
71 self.__token = token
72 self.__ui_language_preference = ui_language_preference
73 self.__is_interactive = is_interactive
74 self.__logged_in = self._user_id is not None
75 self.__cookies = []
77 if self.__is_interactive:
78 if not self._grpc_context:
79 raise ValueError("Tried to construct interactive context without grpc context")
80 self.__headers = dict(self._grpc_context.invocation_metadata())
82 if self.__logged_in:
83 if not self._user_id:
84 raise ValueError("Invalid state, logged in but missing user_id")
86 def __verify_interactive(self):
87 if not self.__is_interactive:
88 raise NonInteractiveContextException("Called an interactive context function in non-interactive context")
90 def __verify_logged_in(self):
91 if not self.__logged_in:
92 raise NotLoggedInContextException("Called a logged-in function from logged-out context")
94 def is_logged_in(self):
95 return self.__logged_in
97 def abort(self, status_code: grpc.StatusCode, error_message: str) -> None:
98 """
99 Raises an error that's returned to the user
100 """
101 if not self.__is_interactive:
102 raise NonInteractiveAbortException(status_code, error_message)
103 else:
104 self._grpc_context.abort(status_code, error_message)
106 def set_cookies(self, cookies: list[str]) -> None:
107 """
108 Sets a list of HTTP cookies
109 """
110 self.__verify_interactive()
111 self.__cookies += cookies
113 def _send_cookies(self) -> None:
114 self._grpc_context.send_initial_metadata([("set-cookie", cookie) for cookie in self.__cookies])
116 @property
117 def headers(self):
118 """
119 Gets a list of HTTP headers for the requests
120 """
121 self.__verify_interactive()
122 return self.__headers
124 @property
125 def user_id(self) -> int:
126 """
127 Returns the user ID of the currently logged in user, if available
128 """
129 self.__verify_logged_in()
130 return self._user_id
132 @property
133 def is_api_key(self) -> bool:
134 """
135 Returns whether the API call was done with API key or not, if available
136 """
137 self.__verify_logged_in()
138 return self._is_api_key
140 @property
141 def token(self) -> str:
142 """
143 Returns the token (session cookie/api key) of the current session, if available
144 """
145 self.__verify_interactive()
146 self.__verify_logged_in()
147 return self.__token
149 @property
150 def ui_language_preference(self) -> str | None:
151 return self.__ui_language_preference
154def make_interactive_user_context(grpc_context, user_id, is_api_key, token, ui_language_preference):
155 return CouchersContext(
156 is_interactive=True,
157 grpc_context=grpc_context,
158 user_id=user_id,
159 is_api_key=is_api_key,
160 token=token,
161 ui_language_preference=ui_language_preference,
162 )
165def make_one_off_interactive_user_context(couchers_context, user_id):
166 return CouchersContext(
167 is_interactive=True,
168 grpc_context=couchers_context._grpc_context,
169 user_id=user_id,
170 is_api_key=None,
171 token=None,
172 ui_language_preference=None,
173 )
176def make_media_context(grpc_context):
177 return CouchersContext(
178 is_interactive=True,
179 user_id=None,
180 is_api_key=False,
181 grpc_context=grpc_context,
182 token=None,
183 ui_language_preference=None,
184 )
187def make_background_user_context(user_id):
188 return CouchersContext(
189 is_interactive=False,
190 user_id=user_id,
191 is_api_key=None,
192 grpc_context=None,
193 token=None,
194 ui_language_preference=None,
195 )