bacommon.cloud

Functionality related to cloud functionality.

  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')]
@ioprepped
@dataclass
class LoginProxyRequestMessage(efro.message._message.Message):
20@ioprepped
21@dataclass
22class LoginProxyRequestMessage(Message):
23    """Request send to the cloud to ask for a login-proxy."""
24
25    @classmethod
26    def get_response_types(cls) -> list[type[Response] | None]:
27        return [LoginProxyRequestResponse]

Request send to the cloud to ask for a login-proxy.

@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
25    @classmethod
26    def get_response_types(cls) -> list[type[Response] | None]:
27        return [LoginProxyRequestResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class LoginProxyRequestResponse(efro.message._message.Response):
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')]

Response to a request for a login proxy.

LoginProxyRequestResponse( url: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064f7cd0>], proxyid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064f7b50>], proxykey: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064f7bd0>])
url: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1062db810>]
proxyid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x105a4db90>]
proxykey: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064f5750>]
@ioprepped
@dataclass
class LoginProxyStateQueryMessage(efro.message._message.Message):
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    @classmethod
54    def get_response_types(cls) -> list[type[Response] | None]:
55        return [LoginProxyStateQueryResponse]

Soo.. how is that login proxy going?

LoginProxyStateQueryMessage( proxyid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064b3a50>], proxykey: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064b3990>])
proxyid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x106299250>]
proxykey: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x105d37e10>]
@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
53    @classmethod
54    def get_response_types(cls) -> list[type[Response] | None]:
55        return [LoginProxyStateQueryResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class LoginProxyStateQueryResponse(efro.message._message.Response):
58@ioprepped
59@dataclass
60class LoginProxyStateQueryResponse(Response):
61    """Here's the info on that login-proxy you asked about, boss."""
62
63    class State(Enum):
64        """States a login-proxy can be in."""
65
66        WAITING = 'waiting'
67        SUCCESS = 'success'
68        FAIL = 'fail'
69
70    state: Annotated[State, IOAttrs('s')]
71
72    # On success, these will be filled out.
73    credentials: Annotated[str | None, IOAttrs('tk')]

Here's the info on that login-proxy you asked about, boss.

LoginProxyStateQueryResponse( state: typing.Annotated[LoginProxyStateQueryResponse.State, <efro.dataclassio._base.IOAttrs object at 0x1064c3290>], credentials: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x1064c1fd0>])
state: typing.Annotated[LoginProxyStateQueryResponse.State, <efro.dataclassio._base.IOAttrs object at 0x1064c11d0>]
credentials: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x1064c18d0>]
class LoginProxyStateQueryResponse.State(enum.Enum):
63    class State(Enum):
64        """States a login-proxy can be in."""
65
66        WAITING = 'waiting'
67        SUCCESS = 'success'
68        FAIL = 'fail'

States a login-proxy can be in.

WAITING = <State.WAITING: 'waiting'>
SUCCESS = <State.SUCCESS: 'success'>
FAIL = <State.FAIL: 'fail'>
Inherited Members
enum.Enum
name
value
@ioprepped
@dataclass
class LoginProxyCompleteMessage(efro.message._message.Message):
76@ioprepped
77@dataclass
78class LoginProxyCompleteMessage(Message):
79    """Just so you know, we're done with this proxy."""
80
81    proxyid: Annotated[str, IOAttrs('p')]

Just so you know, we're done with this proxy.

LoginProxyCompleteMessage( proxyid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1064b1d10>])
proxyid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x106451b10>]
Inherited Members
efro.message._message.Message
get_response_types
@ioprepped
@dataclass
class PingMessage(efro.message._message.Message):
84@ioprepped
85@dataclass
86class PingMessage(Message):
87    """Standard ping."""
88
89    @classmethod
90    def get_response_types(cls) -> list[type[Response] | None]:
91        return [PingResponse]

Standard ping.

@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
89    @classmethod
90    def get_response_types(cls) -> list[type[Response] | None]:
91        return [PingResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class PingResponse(efro.message._message.Response):
94@ioprepped
95@dataclass
96class PingResponse(Response):
97    """pong."""

pong.

@ioprepped
@dataclass
class TestMessage(efro.message._message.Message):
100@ioprepped
101@dataclass
102class TestMessage(Message):
103    """Can I get some of that workspace action?"""
104
105    testfoo: Annotated[int, IOAttrs('f')]
106
107    @classmethod
108    def get_response_types(cls) -> list[type[Response] | None]:
109        return [TestResponse]

Can I get some of that workspace action?

TestMessage( testfoo: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x1064c8750>])
testfoo: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x1062d9950>]
@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
107    @classmethod
108    def get_response_types(cls) -> list[type[Response] | None]:
109        return [TestResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class TestResponse(efro.message._message.Response):
112@ioprepped
113@dataclass
114class TestResponse(Response):
115    """Here's that workspace you asked for, boss."""
116
117    testfoo: Annotated[int, IOAttrs('f')]

Here's that workspace you asked for, boss.

TestResponse( testfoo: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x1064be0d0>])
testfoo: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x104ce1450>]
@ioprepped
@dataclass
class WorkspaceFetchState:
120@ioprepped
121@dataclass
122class WorkspaceFetchState:
123    """Common state data for a workspace fetch."""
124
125    manifest: Annotated[DirectoryManifest, IOAttrs('m')]
126    iteration: Annotated[int, IOAttrs('i')] = 0
127    total_deletes: Annotated[int, IOAttrs('tdels')] = 0
128    total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
129    total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None

Common state data for a workspace fetch.

WorkspaceFetchState( manifest: typing.Annotated[bacommon.transfer.DirectoryManifest, <efro.dataclassio._base.IOAttrs object at 0x1065294d0>], iteration: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x1065290d0>] = 0, total_deletes: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x10652b6d0>] = 0, total_downloads: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x106529050>] = 0, total_up_to_date: typing.Annotated[int | None, <efro.dataclassio._base.IOAttrs object at 0x10652b290>] = None)
manifest: typing.Annotated[bacommon.transfer.DirectoryManifest, <efro.dataclassio._base.IOAttrs object at 0x105ff6610>]
iteration: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x104d04950>] = 0
total_deletes: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x105ff6ed0>] = 0
total_downloads: typing.Annotated[int, <efro.dataclassio._base.IOAttrs object at 0x1064eeb10>] = 0
total_up_to_date: typing.Annotated[int | None, <efro.dataclassio._base.IOAttrs object at 0x1064ef950>] = None
@ioprepped
@dataclass
class WorkspaceFetchMessage(efro.message._message.Message):
132@ioprepped
133@dataclass
134class WorkspaceFetchMessage(Message):
135    """Can I get some of that workspace action?"""
136
137    workspaceid: Annotated[str, IOAttrs('w')]
138    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
139
140    @classmethod
141    def get_response_types(cls) -> list[type[Response] | None]:
142        return [WorkspaceFetchResponse]

Can I get some of that workspace action?

WorkspaceFetchMessage( workspaceid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x106534d90>], state: typing.Annotated[WorkspaceFetchState, <efro.dataclassio._base.IOAttrs object at 0x106534ad0>])
workspaceid: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x105a4ced0>]
state: typing.Annotated[WorkspaceFetchState, <efro.dataclassio._base.IOAttrs object at 0x106454710>]
@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
140    @classmethod
141    def get_response_types(cls) -> list[type[Response] | None]:
142        return [WorkspaceFetchResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class WorkspaceFetchResponse(efro.message._message.Response):
145@ioprepped
146@dataclass
147class WorkspaceFetchResponse(Response):
148    """Here's that workspace you asked for, boss."""
149
150    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
151    deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
152        default_factory=list
153    )
154    downloads_inline: Annotated[
155        dict[str, bytes], IOAttrs('dinl', store_default=False)
156    ] = field(default_factory=dict)
157
158    done: Annotated[bool, IOAttrs('d')] = False

Here's that workspace you asked for, boss.

WorkspaceFetchResponse( state: typing.Annotated[WorkspaceFetchState, <efro.dataclassio._base.IOAttrs object at 0x1065416d0>], deletes: typing.Annotated[list[str], <efro.dataclassio._base.IOAttrs object at 0x106541050>] = <factory>, downloads_inline: typing.Annotated[dict[str, bytes], <efro.dataclassio._base.IOAttrs object at 0x106540b10>] = <factory>, done: typing.Annotated[bool, <efro.dataclassio._base.IOAttrs object at 0x106542950>] = False)
state: typing.Annotated[WorkspaceFetchState, <efro.dataclassio._base.IOAttrs object at 0x106420c90>]
deletes: typing.Annotated[list[str], <efro.dataclassio._base.IOAttrs object at 0x105bd2010>]
downloads_inline: typing.Annotated[dict[str, bytes], <efro.dataclassio._base.IOAttrs object at 0x106529e90>]
done: typing.Annotated[bool, <efro.dataclassio._base.IOAttrs object at 0x106528090>] = False
@ioprepped
@dataclass
class MerchAvailabilityMessage(efro.message._message.Message):
161@ioprepped
162@dataclass
163class MerchAvailabilityMessage(Message):
164    """Can we show merch link?"""
165
166    @classmethod
167    def get_response_types(cls) -> list[type[Response] | None]:
168        return [MerchAvailabilityResponse]

Can we show merch link?

@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
166    @classmethod
167    def get_response_types(cls) -> list[type[Response] | None]:
168        return [MerchAvailabilityResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class MerchAvailabilityResponse(efro.message._message.Response):
171@ioprepped
172@dataclass
173class MerchAvailabilityResponse(Response):
174    """About that merch..."""
175
176    url: Annotated[str | None, IOAttrs('u')]

About that merch...

MerchAvailabilityResponse( url: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x10654b850>])
url: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x1064c1810>]
@ioprepped
@dataclass
class SignInMessage(efro.message._message.Message):
179@ioprepped
180@dataclass
181class SignInMessage(Message):
182    """Can I sign in please?"""
183
184    login_type: Annotated[LoginType, IOAttrs('l')]
185    sign_in_token: Annotated[str, IOAttrs('t')]
186
187    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
188    description: Annotated[str, IOAttrs('d', soft_default='-')]
189    apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]
190
191    @classmethod
192    def get_response_types(cls) -> list[type[Response] | None]:
193        return [SignInResponse]

Can I sign in please?

SignInMessage( login_type: typing.Annotated[bacommon.login.LoginType, <efro.dataclassio._base.IOAttrs object at 0x106558590>], sign_in_token: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x106559850>], description: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x106559210>], apptime: typing.Annotated[float, <efro.dataclassio._base.IOAttrs object at 0x1065599d0>])
login_type: typing.Annotated[bacommon.login.LoginType, <efro.dataclassio._base.IOAttrs object at 0x1064f4750>]
sign_in_token: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x1062bed10>]
description: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x105a96b50>]
apptime: typing.Annotated[float, <efro.dataclassio._base.IOAttrs object at 0x10652de50>]
@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
191    @classmethod
192    def get_response_types(cls) -> list[type[Response] | None]:
193        return [SignInResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class SignInResponse(efro.message._message.Response):
196@ioprepped
197@dataclass
198class SignInResponse(Response):
199    """Here's that sign-in result you asked for, boss."""
200
201    credentials: Annotated[str | None, IOAttrs('c')]

Here's that sign-in result you asked for, boss.

SignInResponse( credentials: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x1065380d0>])
credentials: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x1055137d0>]
@ioprepped
@dataclass
class ManageAccountMessage(efro.message._message.Message):
204@ioprepped
205@dataclass
206class ManageAccountMessage(Message):
207    """Message asking for a manage-account url."""
208
209    @classmethod
210    def get_response_types(cls) -> list[type[Response] | None]:
211        return [ManageAccountResponse]

Message asking for a manage-account url.

@classmethod
def get_response_types(cls) -> list[type[efro.message._message.Response] | None]:
209    @classmethod
210    def get_response_types(cls) -> list[type[Response] | None]:
211        return [ManageAccountResponse]

Return all Response types this Message can return when sent.

The default implementation specifies a None return type.

@ioprepped
@dataclass
class ManageAccountResponse(efro.message._message.Response):
214@ioprepped
215@dataclass
216class ManageAccountResponse(Response):
217    """Here's that sign-in result you asked for, boss."""
218
219    url: Annotated[str | None, IOAttrs('u')]

Here's that sign-in result you asked for, boss.

ManageAccountResponse( url: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x10653b210>])
url: typing.Annotated[str | None, <efro.dataclassio._base.IOAttrs object at 0x1056eaa90>]