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')]
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.
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.
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.
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?
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.
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.
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.
Inherited Members
- enum.Enum
- name
- value
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.
Inherited Members
- efro.message._message.Message
- get_response_types
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.
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.
pong.
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?
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.
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.
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.
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?
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.
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.
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?
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.
171@ioprepped 172@dataclass 173class MerchAvailabilityResponse(Response): 174 """About that merch...""" 175 176 url: Annotated[str | None, IOAttrs('u')]
About that merch...
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?
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.
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.
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.
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.
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.