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, 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 19class WebLocation(Enum): 20 """Set of places we can be directed on ballistica.net.""" 21 22 ACCOUNT_EDITOR = 'e' 23 ACCOUNT_DELETE_SECTION = 'd' 24 25 26@ioprepped 27@dataclass 28class LoginProxyRequestMessage(Message): 29 """Request send to the cloud to ask for a login-proxy.""" 30 31 @override 32 @classmethod 33 def get_response_types(cls) -> list[type[Response] | None]: 34 return [LoginProxyRequestResponse] 35 36 37@ioprepped 38@dataclass 39class LoginProxyRequestResponse(Response): 40 """Response to a request for a login proxy.""" 41 42 # URL to direct the user to for sign in. 43 url: Annotated[str, IOAttrs('u')] 44 45 # URL to use for overlay-web-browser sign ins. 46 url_overlay: Annotated[str, IOAttrs('uo')] 47 48 # Proxy-Login id for querying results. 49 proxyid: Annotated[str, IOAttrs('p')] 50 51 # Proxy-Login key for querying results. 52 proxykey: Annotated[str, IOAttrs('k')] 53 54 55@ioprepped 56@dataclass 57class LoginProxyStateQueryMessage(Message): 58 """Soo.. how is that login proxy going?""" 59 60 proxyid: Annotated[str, IOAttrs('p')] 61 proxykey: Annotated[str, IOAttrs('k')] 62 63 @override 64 @classmethod 65 def get_response_types(cls) -> list[type[Response] | None]: 66 return [LoginProxyStateQueryResponse] 67 68 69@ioprepped 70@dataclass 71class LoginProxyStateQueryResponse(Response): 72 """Here's the info on that login-proxy you asked about, boss.""" 73 74 class State(Enum): 75 """States a login-proxy can be in.""" 76 77 WAITING = 'waiting' 78 SUCCESS = 'success' 79 FAIL = 'fail' 80 81 state: Annotated[State, IOAttrs('s')] 82 83 # On success, these will be filled out. 84 credentials: Annotated[str | None, IOAttrs('tk')] 85 86 87@ioprepped 88@dataclass 89class LoginProxyCompleteMessage(Message): 90 """Just so you know, we're done with this proxy.""" 91 92 proxyid: Annotated[str, IOAttrs('p')] 93 94 95@ioprepped 96@dataclass 97class PingMessage(Message): 98 """Standard ping.""" 99 100 @override 101 @classmethod 102 def get_response_types(cls) -> list[type[Response] | None]: 103 return [PingResponse] 104 105 106@ioprepped 107@dataclass 108class PingResponse(Response): 109 """pong.""" 110 111 112@ioprepped 113@dataclass 114class TestMessage(Message): 115 """Can I get some of that workspace action?""" 116 117 testfoo: Annotated[int, IOAttrs('f')] 118 119 @override 120 @classmethod 121 def get_response_types(cls) -> list[type[Response] | None]: 122 return [TestResponse] 123 124 125@ioprepped 126@dataclass 127class TestResponse(Response): 128 """Here's that workspace you asked for, boss.""" 129 130 testfoo: Annotated[int, IOAttrs('f')] 131 132 133@ioprepped 134@dataclass 135class SendInfoMessage(Message): 136 """User is using the send-info function""" 137 138 description: Annotated[str, IOAttrs('c')] 139 140 @override 141 @classmethod 142 def get_response_types(cls) -> list[type[Response] | None]: 143 return [SendInfoResponse] 144 145 146@ioprepped 147@dataclass 148class SendInfoResponse(Response): 149 """Response to sending into the server.""" 150 151 handled: Annotated[bool, IOAttrs('v')] 152 message: Annotated[str | None, IOAttrs('m', store_default=False)] = None 153 legacy_code: Annotated[str | None, IOAttrs('l', store_default=False)] = None 154 155 156@ioprepped 157@dataclass 158class WorkspaceFetchState: 159 """Common state data for a workspace fetch.""" 160 161 manifest: Annotated[DirectoryManifest, IOAttrs('m')] 162 iteration: Annotated[int, IOAttrs('i')] = 0 163 total_deletes: Annotated[int, IOAttrs('tdels')] = 0 164 total_downloads: Annotated[int, IOAttrs('tdlds')] = 0 165 total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None 166 167 168@ioprepped 169@dataclass 170class WorkspaceFetchMessage(Message): 171 """Can I get some of that workspace action?""" 172 173 workspaceid: Annotated[str, IOAttrs('w')] 174 state: Annotated[WorkspaceFetchState, IOAttrs('s')] 175 176 @override 177 @classmethod 178 def get_response_types(cls) -> list[type[Response] | None]: 179 return [WorkspaceFetchResponse] 180 181 182@ioprepped 183@dataclass 184class WorkspaceFetchResponse(Response): 185 """Here's that workspace you asked for, boss.""" 186 187 state: Annotated[WorkspaceFetchState, IOAttrs('s')] 188 deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field( 189 default_factory=list 190 ) 191 downloads_inline: Annotated[ 192 dict[str, bytes], IOAttrs('dinl', store_default=False) 193 ] = field(default_factory=dict) 194 195 done: Annotated[bool, IOAttrs('d')] = False 196 197 198@ioprepped 199@dataclass 200class MerchAvailabilityMessage(Message): 201 """Can we show merch link?""" 202 203 @override 204 @classmethod 205 def get_response_types(cls) -> list[type[Response] | None]: 206 return [MerchAvailabilityResponse] 207 208 209@ioprepped 210@dataclass 211class MerchAvailabilityResponse(Response): 212 """About that merch...""" 213 214 url: Annotated[str | None, IOAttrs('u')] 215 216 217@ioprepped 218@dataclass 219class SignInMessage(Message): 220 """Can I sign in please?""" 221 222 login_type: Annotated[LoginType, IOAttrs('l')] 223 sign_in_token: Annotated[str, IOAttrs('t')] 224 225 # For debugging. Can remove soft_default once build 20988+ is ubiquitous. 226 description: Annotated[str, IOAttrs('d', soft_default='-')] 227 apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)] 228 229 @override 230 @classmethod 231 def get_response_types(cls) -> list[type[Response] | None]: 232 return [SignInResponse] 233 234 235@ioprepped 236@dataclass 237class SignInResponse(Response): 238 """Here's that sign-in result you asked for, boss.""" 239 240 credentials: Annotated[str | None, IOAttrs('c')] 241 242 243@ioprepped 244@dataclass 245class ManageAccountMessage(Message): 246 """Message asking for a manage-account url.""" 247 248 weblocation: Annotated[WebLocation, IOAttrs('l')] = ( 249 WebLocation.ACCOUNT_EDITOR 250 ) 251 252 @override 253 @classmethod 254 def get_response_types(cls) -> list[type[Response] | None]: 255 return [ManageAccountResponse] 256 257 258@ioprepped 259@dataclass 260class ManageAccountResponse(Response): 261 """Here's that sign-in result you asked for, boss.""" 262 263 url: Annotated[str | None, IOAttrs('u')] 264 265 266@ioprepped 267@dataclass 268class StoreQueryMessage(Message): 269 """Message asking about purchasable stuff and store related state.""" 270 271 @override 272 @classmethod 273 def get_response_types(cls) -> list[type[Response] | None]: 274 return [StoreQueryResponse] 275 276 277@ioprepped 278@dataclass 279class StoreQueryResponse(Response): 280 """Here's that store info you asked for, boss.""" 281 282 class Result(Enum): 283 """Our overall result.""" 284 285 SUCCESS = 's' 286 ERROR = 'e' 287 288 @dataclass 289 class Purchase: 290 """Info about a purchasable thing.""" 291 292 purchaseid: Annotated[str, IOAttrs('id')] 293 294 # Overall result; all data is undefined if not SUCCESS. 295 result: Annotated[Result, IOAttrs('r')] 296 297 tokens: Annotated[int, IOAttrs('t')] 298 gold_pass: Annotated[bool, IOAttrs('g')] 299 300 available_purchases: Annotated[list[Purchase], IOAttrs('p')] 301 token_info_url: Annotated[str, IOAttrs('tiu')] 302 303 304@ioprepped 305@dataclass 306class BSPrivatePartyMessage(Message): 307 """Message asking about info we need for private-party UI.""" 308 309 need_datacode: Annotated[bool, IOAttrs('d')] 310 311 @override 312 @classmethod 313 def get_response_types(cls) -> list[type[Response] | None]: 314 return [BSPrivatePartyResponse] 315 316 317@ioprepped 318@dataclass 319class BSPrivatePartyResponse(Response): 320 """Here's that private party UI info you asked for, boss.""" 321 322 success: Annotated[bool, IOAttrs('s')] 323 tokens: Annotated[int, IOAttrs('t')] 324 gold_pass: Annotated[bool, IOAttrs('g')] 325 datacode: Annotated[str | None, IOAttrs('d')]
20class WebLocation(Enum): 21 """Set of places we can be directed on ballistica.net.""" 22 23 ACCOUNT_EDITOR = 'e' 24 ACCOUNT_DELETE_SECTION = 'd'
Set of places we can be directed on ballistica.net.
Inherited Members
- enum.Enum
- name
- value
27@ioprepped 28@dataclass 29class LoginProxyRequestMessage(Message): 30 """Request send to the cloud to ask for a login-proxy.""" 31 32 @override 33 @classmethod 34 def get_response_types(cls) -> list[type[Response] | None]: 35 return [LoginProxyRequestResponse]
Request send to the cloud to ask for a login-proxy.
38@ioprepped 39@dataclass 40class LoginProxyRequestResponse(Response): 41 """Response to a request for a login proxy.""" 42 43 # URL to direct the user to for sign in. 44 url: Annotated[str, IOAttrs('u')] 45 46 # URL to use for overlay-web-browser sign ins. 47 url_overlay: Annotated[str, IOAttrs('uo')] 48 49 # Proxy-Login id for querying results. 50 proxyid: Annotated[str, IOAttrs('p')] 51 52 # Proxy-Login key for querying results. 53 proxykey: Annotated[str, IOAttrs('k')]
Response to a request for a login proxy.
56@ioprepped 57@dataclass 58class LoginProxyStateQueryMessage(Message): 59 """Soo.. how is that login proxy going?""" 60 61 proxyid: Annotated[str, IOAttrs('p')] 62 proxykey: Annotated[str, IOAttrs('k')] 63 64 @override 65 @classmethod 66 def get_response_types(cls) -> list[type[Response] | None]: 67 return [LoginProxyStateQueryResponse]
Soo.. how is that login proxy going?
70@ioprepped 71@dataclass 72class LoginProxyStateQueryResponse(Response): 73 """Here's the info on that login-proxy you asked about, boss.""" 74 75 class State(Enum): 76 """States a login-proxy can be in.""" 77 78 WAITING = 'waiting' 79 SUCCESS = 'success' 80 FAIL = 'fail' 81 82 state: Annotated[State, IOAttrs('s')] 83 84 # On success, these will be filled out. 85 credentials: Annotated[str | None, IOAttrs('tk')]
Here's the info on that login-proxy you asked about, boss.
75 class State(Enum): 76 """States a login-proxy can be in.""" 77 78 WAITING = 'waiting' 79 SUCCESS = 'success' 80 FAIL = 'fail'
States a login-proxy can be in.
Inherited Members
- enum.Enum
- name
- value
88@ioprepped 89@dataclass 90class LoginProxyCompleteMessage(Message): 91 """Just so you know, we're done with this proxy.""" 92 93 proxyid: Annotated[str, IOAttrs('p')]
Just so you know, we're done with this proxy.
Inherited Members
- efro.message._message.Message
- get_response_types
96@ioprepped 97@dataclass 98class PingMessage(Message): 99 """Standard ping.""" 100 101 @override 102 @classmethod 103 def get_response_types(cls) -> list[type[Response] | None]: 104 return [PingResponse]
Standard ping.
pong.
113@ioprepped 114@dataclass 115class TestMessage(Message): 116 """Can I get some of that workspace action?""" 117 118 testfoo: Annotated[int, IOAttrs('f')] 119 120 @override 121 @classmethod 122 def get_response_types(cls) -> list[type[Response] | None]: 123 return [TestResponse]
Can I get some of that workspace action?
126@ioprepped 127@dataclass 128class TestResponse(Response): 129 """Here's that workspace you asked for, boss.""" 130 131 testfoo: Annotated[int, IOAttrs('f')]
Here's that workspace you asked for, boss.
134@ioprepped 135@dataclass 136class SendInfoMessage(Message): 137 """User is using the send-info function""" 138 139 description: Annotated[str, IOAttrs('c')] 140 141 @override 142 @classmethod 143 def get_response_types(cls) -> list[type[Response] | None]: 144 return [SendInfoResponse]
User is using the send-info function
147@ioprepped 148@dataclass 149class SendInfoResponse(Response): 150 """Response to sending into the server.""" 151 152 handled: Annotated[bool, IOAttrs('v')] 153 message: Annotated[str | None, IOAttrs('m', store_default=False)] = None 154 legacy_code: Annotated[str | None, IOAttrs('l', store_default=False)] = None
Response to sending into the server.
157@ioprepped 158@dataclass 159class WorkspaceFetchState: 160 """Common state data for a workspace fetch.""" 161 162 manifest: Annotated[DirectoryManifest, IOAttrs('m')] 163 iteration: Annotated[int, IOAttrs('i')] = 0 164 total_deletes: Annotated[int, IOAttrs('tdels')] = 0 165 total_downloads: Annotated[int, IOAttrs('tdlds')] = 0 166 total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None
Common state data for a workspace fetch.
169@ioprepped 170@dataclass 171class WorkspaceFetchMessage(Message): 172 """Can I get some of that workspace action?""" 173 174 workspaceid: Annotated[str, IOAttrs('w')] 175 state: Annotated[WorkspaceFetchState, IOAttrs('s')] 176 177 @override 178 @classmethod 179 def get_response_types(cls) -> list[type[Response] | None]: 180 return [WorkspaceFetchResponse]
Can I get some of that workspace action?
183@ioprepped 184@dataclass 185class WorkspaceFetchResponse(Response): 186 """Here's that workspace you asked for, boss.""" 187 188 state: Annotated[WorkspaceFetchState, IOAttrs('s')] 189 deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field( 190 default_factory=list 191 ) 192 downloads_inline: Annotated[ 193 dict[str, bytes], IOAttrs('dinl', store_default=False) 194 ] = field(default_factory=dict) 195 196 done: Annotated[bool, IOAttrs('d')] = False
Here's that workspace you asked for, boss.
199@ioprepped 200@dataclass 201class MerchAvailabilityMessage(Message): 202 """Can we show merch link?""" 203 204 @override 205 @classmethod 206 def get_response_types(cls) -> list[type[Response] | None]: 207 return [MerchAvailabilityResponse]
Can we show merch link?
210@ioprepped 211@dataclass 212class MerchAvailabilityResponse(Response): 213 """About that merch...""" 214 215 url: Annotated[str | None, IOAttrs('u')]
About that merch...
218@ioprepped 219@dataclass 220class SignInMessage(Message): 221 """Can I sign in please?""" 222 223 login_type: Annotated[LoginType, IOAttrs('l')] 224 sign_in_token: Annotated[str, IOAttrs('t')] 225 226 # For debugging. Can remove soft_default once build 20988+ is ubiquitous. 227 description: Annotated[str, IOAttrs('d', soft_default='-')] 228 apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)] 229 230 @override 231 @classmethod 232 def get_response_types(cls) -> list[type[Response] | None]: 233 return [SignInResponse]
Can I sign in please?
236@ioprepped 237@dataclass 238class SignInResponse(Response): 239 """Here's that sign-in result you asked for, boss.""" 240 241 credentials: Annotated[str | None, IOAttrs('c')]
Here's that sign-in result you asked for, boss.
244@ioprepped 245@dataclass 246class ManageAccountMessage(Message): 247 """Message asking for a manage-account url.""" 248 249 weblocation: Annotated[WebLocation, IOAttrs('l')] = ( 250 WebLocation.ACCOUNT_EDITOR 251 ) 252 253 @override 254 @classmethod 255 def get_response_types(cls) -> list[type[Response] | None]: 256 return [ManageAccountResponse]
Message asking for a manage-account url.
259@ioprepped 260@dataclass 261class ManageAccountResponse(Response): 262 """Here's that sign-in result you asked for, boss.""" 263 264 url: Annotated[str | None, IOAttrs('u')]
Here's that sign-in result you asked for, boss.
267@ioprepped 268@dataclass 269class StoreQueryMessage(Message): 270 """Message asking about purchasable stuff and store related state.""" 271 272 @override 273 @classmethod 274 def get_response_types(cls) -> list[type[Response] | None]: 275 return [StoreQueryResponse]
Message asking about purchasable stuff and store related state.
278@ioprepped 279@dataclass 280class StoreQueryResponse(Response): 281 """Here's that store info you asked for, boss.""" 282 283 class Result(Enum): 284 """Our overall result.""" 285 286 SUCCESS = 's' 287 ERROR = 'e' 288 289 @dataclass 290 class Purchase: 291 """Info about a purchasable thing.""" 292 293 purchaseid: Annotated[str, IOAttrs('id')] 294 295 # Overall result; all data is undefined if not SUCCESS. 296 result: Annotated[Result, IOAttrs('r')] 297 298 tokens: Annotated[int, IOAttrs('t')] 299 gold_pass: Annotated[bool, IOAttrs('g')] 300 301 available_purchases: Annotated[list[Purchase], IOAttrs('p')] 302 token_info_url: Annotated[str, IOAttrs('tiu')]
Here's that store info you asked for, boss.
Our overall result.
Inherited Members
- enum.Enum
- name
- value
289 @dataclass 290 class Purchase: 291 """Info about a purchasable thing.""" 292 293 purchaseid: Annotated[str, IOAttrs('id')]
Info about a purchasable thing.
305@ioprepped 306@dataclass 307class BSPrivatePartyMessage(Message): 308 """Message asking about info we need for private-party UI.""" 309 310 need_datacode: Annotated[bool, IOAttrs('d')] 311 312 @override 313 @classmethod 314 def get_response_types(cls) -> list[type[Response] | None]: 315 return [BSPrivatePartyResponse]
Message asking about info we need for private-party UI.
318@ioprepped 319@dataclass 320class BSPrivatePartyResponse(Response): 321 """Here's that private party UI info you asked for, boss.""" 322 323 success: Annotated[bool, IOAttrs('s')] 324 tokens: Annotated[int, IOAttrs('t')] 325 gold_pass: Annotated[bool, IOAttrs('g')] 326 datacode: Annotated[str | None, IOAttrs('d')]
Here's that private party UI info you asked for, boss.