bacommon.cloud

  1# Released under the MIT License. See LICENSE for details.
  2#
  3"""Functionality related to cloud functionality."""
  4
  5from __future__ import annotations
  6from dataclasses import dataclass, field
  7from typing import TYPE_CHECKING, Annotated, override
  8from enum import Enum
  9
 10from efro.message import Message, Response
 11from efro.dataclassio import ioprepped, IOAttrs
 12from bacommon.transfer import DirectoryManifest
 13from bacommon.login import LoginType
 14
 15if TYPE_CHECKING:
 16    pass
 17
 18
 19@ioprepped
 20@dataclass
 21class LoginProxyRequestMessage(Message):
 22    """Request send to the cloud to ask for a login-proxy."""
 23
 24    @override
 25    @classmethod
 26    def get_response_types(cls) -> list[type[Response] | None]:
 27        return [LoginProxyRequestResponse]
 28
 29
 30@ioprepped
 31@dataclass
 32class LoginProxyRequestResponse(Response):
 33    """Response to a request for a login proxy."""
 34
 35    # URL to direct the user to for login.
 36    url: Annotated[str, IOAttrs('u')]
 37
 38    # Proxy-Login id for querying results.
 39    proxyid: Annotated[str, IOAttrs('p')]
 40
 41    # Proxy-Login key for querying results.
 42    proxykey: Annotated[str, IOAttrs('k')]
 43
 44
 45@ioprepped
 46@dataclass
 47class LoginProxyStateQueryMessage(Message):
 48    """Soo.. how is that login proxy going?"""
 49
 50    proxyid: Annotated[str, IOAttrs('p')]
 51    proxykey: Annotated[str, IOAttrs('k')]
 52
 53    @override
 54    @classmethod
 55    def get_response_types(cls) -> list[type[Response] | None]:
 56        return [LoginProxyStateQueryResponse]
 57
 58
 59@ioprepped
 60@dataclass
 61class LoginProxyStateQueryResponse(Response):
 62    """Here's the info on that login-proxy you asked about, boss."""
 63
 64    class State(Enum):
 65        """States a login-proxy can be in."""
 66
 67        WAITING = 'waiting'
 68        SUCCESS = 'success'
 69        FAIL = 'fail'
 70
 71    state: Annotated[State, IOAttrs('s')]
 72
 73    # On success, these will be filled out.
 74    credentials: Annotated[str | None, IOAttrs('tk')]
 75
 76
 77@ioprepped
 78@dataclass
 79class LoginProxyCompleteMessage(Message):
 80    """Just so you know, we're done with this proxy."""
 81
 82    proxyid: Annotated[str, IOAttrs('p')]
 83
 84
 85@ioprepped
 86@dataclass
 87class PingMessage(Message):
 88    """Standard ping."""
 89
 90    @override
 91    @classmethod
 92    def get_response_types(cls) -> list[type[Response] | None]:
 93        return [PingResponse]
 94
 95
 96@ioprepped
 97@dataclass
 98class PingResponse(Response):
 99    """pong."""
100
101
102@ioprepped
103@dataclass
104class TestMessage(Message):
105    """Can I get some of that workspace action?"""
106
107    testfoo: Annotated[int, IOAttrs('f')]
108
109    @override
110    @classmethod
111    def get_response_types(cls) -> list[type[Response] | None]:
112        return [TestResponse]
113
114
115@ioprepped
116@dataclass
117class TestResponse(Response):
118    """Here's that workspace you asked for, boss."""
119
120    testfoo: Annotated[int, IOAttrs('f')]
121
122
123@ioprepped
124@dataclass
125class PromoCodeMessage(Message):
126    """User is entering a promo code"""
127
128    code: Annotated[str, IOAttrs('c')]
129
130    @override
131    @classmethod
132    def get_response_types(cls) -> list[type[Response] | None]:
133        return [PromoCodeResponse]
134
135
136@ioprepped
137@dataclass
138class PromoCodeResponse(Response):
139    """Applied that promo code for ya, boss."""
140
141    valid: Annotated[bool, IOAttrs('v')]
142    message: Annotated[str | None, IOAttrs('m', store_default=False)] = None
143
144
145@ioprepped
146@dataclass
147class WorkspaceFetchState:
148    """Common state data for a workspace fetch."""
149
150    manifest: Annotated[DirectoryManifest, IOAttrs('m')]
151    iteration: Annotated[int, IOAttrs('i')] = 0
152    total_deletes: Annotated[int, IOAttrs('tdels')] = 0
153    total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
154    total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None
155
156
157@ioprepped
158@dataclass
159class WorkspaceFetchMessage(Message):
160    """Can I get some of that workspace action?"""
161
162    workspaceid: Annotated[str, IOAttrs('w')]
163    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
164
165    @override
166    @classmethod
167    def get_response_types(cls) -> list[type[Response] | None]:
168        return [WorkspaceFetchResponse]
169
170
171@ioprepped
172@dataclass
173class WorkspaceFetchResponse(Response):
174    """Here's that workspace you asked for, boss."""
175
176    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
177    deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
178        default_factory=list
179    )
180    downloads_inline: Annotated[
181        dict[str, bytes], IOAttrs('dinl', store_default=False)
182    ] = field(default_factory=dict)
183
184    done: Annotated[bool, IOAttrs('d')] = False
185
186
187@ioprepped
188@dataclass
189class MerchAvailabilityMessage(Message):
190    """Can we show merch link?"""
191
192    @override
193    @classmethod
194    def get_response_types(cls) -> list[type[Response] | None]:
195        return [MerchAvailabilityResponse]
196
197
198@ioprepped
199@dataclass
200class MerchAvailabilityResponse(Response):
201    """About that merch..."""
202
203    url: Annotated[str | None, IOAttrs('u')]
204
205
206@ioprepped
207@dataclass
208class SignInMessage(Message):
209    """Can I sign in please?"""
210
211    login_type: Annotated[LoginType, IOAttrs('l')]
212    sign_in_token: Annotated[str, IOAttrs('t')]
213
214    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
215    description: Annotated[str, IOAttrs('d', soft_default='-')]
216    apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]
217
218    @override
219    @classmethod
220    def get_response_types(cls) -> list[type[Response] | None]:
221        return [SignInResponse]
222
223
224@ioprepped
225@dataclass
226class SignInResponse(Response):
227    """Here's that sign-in result you asked for, boss."""
228
229    credentials: Annotated[str | None, IOAttrs('c')]
230
231
232@ioprepped
233@dataclass
234class ManageAccountMessage(Message):
235    """Message asking for a manage-account url."""
236
237    @override
238    @classmethod
239    def get_response_types(cls) -> list[type[Response] | None]:
240        return [ManageAccountResponse]
241
242
243@ioprepped
244@dataclass
245class ManageAccountResponse(Response):
246    """Here's that sign-in result you asked for, boss."""
247
248    url: Annotated[str | None, IOAttrs('u')]