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
  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    @classmethod
 25    def get_response_types(cls) -> list[type[Response] | None]:
 26        return [LoginProxyRequestResponse]
 27
 28
 29@ioprepped
 30@dataclass
 31class LoginProxyRequestResponse(Response):
 32    """Response to a request for a login proxy."""
 33
 34    # URL to direct the user to for login.
 35    url: Annotated[str, IOAttrs('u')]
 36
 37    # Proxy-Login id for querying results.
 38    proxyid: Annotated[str, IOAttrs('p')]
 39
 40    # Proxy-Login key for querying results.
 41    proxykey: Annotated[str, IOAttrs('k')]
 42
 43
 44@ioprepped
 45@dataclass
 46class LoginProxyStateQueryMessage(Message):
 47    """Soo.. how is that login proxy going?"""
 48
 49    proxyid: Annotated[str, IOAttrs('p')]
 50    proxykey: Annotated[str, IOAttrs('k')]
 51
 52    @classmethod
 53    def get_response_types(cls) -> list[type[Response] | None]:
 54        return [LoginProxyStateQueryResponse]
 55
 56
 57@ioprepped
 58@dataclass
 59class LoginProxyStateQueryResponse(Response):
 60    """Here's the info on that login-proxy you asked about, boss."""
 61
 62    class State(Enum):
 63        """States a login-proxy can be in."""
 64
 65        WAITING = 'waiting'
 66        SUCCESS = 'success'
 67        FAIL = 'fail'
 68
 69    state: Annotated[State, IOAttrs('s')]
 70
 71    # On success, these will be filled out.
 72    credentials: Annotated[str | None, IOAttrs('tk')]
 73
 74
 75@ioprepped
 76@dataclass
 77class LoginProxyCompleteMessage(Message):
 78    """Just so you know, we're done with this proxy."""
 79
 80    proxyid: Annotated[str, IOAttrs('p')]
 81
 82
 83@ioprepped
 84@dataclass
 85class PingMessage(Message):
 86    """Standard ping."""
 87
 88    @classmethod
 89    def get_response_types(cls) -> list[type[Response] | None]:
 90        return [PingResponse]
 91
 92
 93@ioprepped
 94@dataclass
 95class PingResponse(Response):
 96    """pong."""
 97
 98
 99@ioprepped
100@dataclass
101class TestMessage(Message):
102    """Can I get some of that workspace action?"""
103
104    testfoo: Annotated[int, IOAttrs('f')]
105
106    @classmethod
107    def get_response_types(cls) -> list[type[Response] | None]:
108        return [TestResponse]
109
110
111@ioprepped
112@dataclass
113class TestResponse(Response):
114    """Here's that workspace you asked for, boss."""
115
116    testfoo: Annotated[int, IOAttrs('f')]
117
118
119@ioprepped
120@dataclass
121class WorkspaceFetchState:
122    """Common state data for a workspace fetch."""
123
124    manifest: Annotated[DirectoryManifest, IOAttrs('m')]
125    iteration: Annotated[int, IOAttrs('i')] = 0
126    total_deletes: Annotated[int, IOAttrs('tdels')] = 0
127    total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
128    total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None
129
130
131@ioprepped
132@dataclass
133class WorkspaceFetchMessage(Message):
134    """Can I get some of that workspace action?"""
135
136    workspaceid: Annotated[str, IOAttrs('w')]
137    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
138
139    @classmethod
140    def get_response_types(cls) -> list[type[Response] | None]:
141        return [WorkspaceFetchResponse]
142
143
144@ioprepped
145@dataclass
146class WorkspaceFetchResponse(Response):
147    """Here's that workspace you asked for, boss."""
148
149    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
150    deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
151        default_factory=list
152    )
153    downloads_inline: Annotated[
154        dict[str, bytes], IOAttrs('dinl', store_default=False)
155    ] = field(default_factory=dict)
156
157    done: Annotated[bool, IOAttrs('d')] = False
158
159
160@ioprepped
161@dataclass
162class MerchAvailabilityMessage(Message):
163    """Can we show merch link?"""
164
165    @classmethod
166    def get_response_types(cls) -> list[type[Response] | None]:
167        return [MerchAvailabilityResponse]
168
169
170@ioprepped
171@dataclass
172class MerchAvailabilityResponse(Response):
173    """About that merch..."""
174
175    url: Annotated[str | None, IOAttrs('u')]
176
177
178@ioprepped
179@dataclass
180class SignInMessage(Message):
181    """Can I sign in please?"""
182
183    login_type: Annotated[LoginType, IOAttrs('l')]
184    sign_in_token: Annotated[str, IOAttrs('t')]
185
186    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
187    description: Annotated[str, IOAttrs('d', soft_default='-')]
188    apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]
189
190    @classmethod
191    def get_response_types(cls) -> list[type[Response] | None]:
192        return [SignInResponse]
193
194
195@ioprepped
196@dataclass
197class SignInResponse(Response):
198    """Here's that sign-in result you asked for, boss."""
199
200    credentials: Annotated[str | None, IOAttrs('c')]
201
202
203@ioprepped
204@dataclass
205class ManageAccountMessage(Message):
206    """Message asking for a manage-account url."""
207
208    @classmethod
209    def get_response_types(cls) -> list[type[Response] | None]:
210        return [ManageAccountResponse]
211
212
213@ioprepped
214@dataclass
215class ManageAccountResponse(Response):
216    """Here's that sign-in result you asked for, boss."""
217
218    url: Annotated[str | None, IOAttrs('u')]