# Released under the MIT License. See LICENSE for details.
#
"""Generates typed message-send overloads for baplus's CloudSubsystem.
The client's cloud message-send APIs (the callback, blocking, async,
and future forms) are generic under the hood but expose typed
``@overload`` stacks so callers get message/response type safety.
Those stacks are generated into marked sections of
``baplus/_cloud.py`` from the registry here by ``make update``
(drift fails ``update-check``/CI).
To expose a message on an additional send form (or add a new message
type), update its ``_registry`` entry and run ``make update``.
Response types are derived from each message class's
``get_response_types()``, so they can't drift from the wire protocol.
"""
from __future__ import annotations # Docs-generation hack.
import importlib
from enum import Enum
from typing import TYPE_CHECKING
from efrotools.util import replace_section
if TYPE_CHECKING:
from efro.message import Message
# Column limit matching our Python formatting.
_MAX_LINE = 79
def _registry() -> dict[type['Message'], set[SendForm]]:
"""The client cloud-message registry: message type -> send forms.
Order here is emission order (grouped roughly by purpose).
"""
# Import these here to keep this module cheap to import.
import bacommon.classic
import bacommon.cloud
import bacommon.clouddialog
cb = SendForm.CB
syn = SendForm.SYNC
asy = SendForm.ASYNC
fut = SendForm.FUTURE
return {
# Account/sign-in.
bacommon.cloud.LoginProxyRequestMessage: {cb, asy},
bacommon.cloud.LoginProxyStateQueryMessage: {cb, asy},
bacommon.cloud.LoginProxyCompleteMessage: {cb, asy},
bacommon.cloud.SignInMessage: {cb},
bacommon.cloud.ManageAccountMessage: {cb},
bacommon.cloud.AuthRequestMessage: {cb},
bacommon.cloud.TransientAPIKeyRequest: {cb},
# General.
bacommon.cloud.CloudValsRequest: {cb},
bacommon.cloud.PingMessage: {cb},
# (Registered on all forms; doubles as the canonical
# form-exercising message. Note mypy needs at least two
# overloads per form, so never let a form's registered
# message count drop to one.)
bacommon.cloud.TestMessage: {cb, syn, asy, fut},
bacommon.cloud.AnalyticsEventMessage: {cb},
# Shipped from the log-reporter's own bg thread. The future
# form so the reporter can bound its wait during its final
# at-shutdown flush; normal sends just block on the future.
bacommon.cloud.ClientLogReportMessage: {fut},
bacommon.cloud.SecureDataCheckerRequest: {cb},
bacommon.cloud.WorkspaceFetchMessage: {syn},
bacommon.cloud.MerchAvailabilityMessage: {syn},
bacommon.cloud.FulfillDocUIRequest: {syn},
bacommon.cloud.ResolveAssetPackageMessage: {syn},
bacommon.cloud.StoreQueryMessage: {cb},
bacommon.cloud.ChestActionMessage: {cb, asy},
# Classic.
bacommon.classic.SendInfoMessage: {asy},
bacommon.classic.LegacyRequest: {syn, fut},
bacommon.classic.GetClassicPurchasesMessage: {cb},
bacommon.classic.PrivatePartyMessage: {cb},
bacommon.classic.InboxRequestMessage: {cb, asy},
bacommon.classic.ChestInfoMessage: {cb, asy},
bacommon.classic.GlobalProfileCheckMessage: {cb},
bacommon.classic.ScoreSubmitMessage: {cb},
bacommon.classic.GetClassicLeaguePresidentButtonInfoMessage: {cb},
# Cloud-dialogs.
bacommon.clouddialog.ActionMessage: {cb},
}
[docs]
def generate_cloud_module(projroot: str, existing_data: str) -> str:
"""Generate baplus/_cloud.py based on its existing version."""
del projroot # Unused currently.
info = f'# This section generated by {__name__}; do not edit.'
registry = _registry()
out = existing_data
for form, marker, emit in (
(SendForm.CB, 'CB', _emit_cb),
(SendForm.SYNC, 'SYNC', _emit_sync),
(SendForm.ASYNC, 'ASYNC', _emit_async),
(SendForm.FUTURE, 'FUTURE', _emit_future),
):
stubs = [
emit(_public_path(cls), _response_union(cls))
for cls, forms in registry.items()
if form in forms
]
contents = '\n'.join(stubs)
out = replace_section(
out,
f' # __CLOUD_MSG_{marker}_OVERLOADS_BEGIN__\n',
f' # __CLOUD_MSG_{marker}_OVERLOADS_END__\n',
f' {info}\n\n{contents}\n',
keep_markers=True,
)
return out
def _public_path(cls: type) -> str:
"""Return the shortest public dotted path for a class.
Classes are commonly defined in private implementation modules
(``bacommon.clouddialog._clouddialog``) but exposed at package
level; annotations should use the public spelling.
"""
modparts = cls.__module__.split('.')
name = cls.__qualname__
while len(modparts) > 1 and modparts[-1].startswith('_'):
candidate = '.'.join(modparts[:-1])
mod = importlib.import_module(candidate)
if getattr(mod, name, None) is cls:
modparts = modparts[:-1]
else:
break
return '.'.join(modparts + [name])
def _response_union(cls: type['Message']) -> str:
"""Return the response annotation for a message type."""
parts = [
'None' if rtype is None else _public_path(rtype)
for rtype in cls.get_response_types()
]
return ' | '.join(parts)
def _emit_cb(mpath: str, runion: str) -> str:
"""Emit one send_message_cb overload (pre-formatted)."""
# Mirror our standard formatting: inline the callback annotation
# if it fits, else progressively expand (the same shapes the
# formatter would pick, so generated output is format-stable).
inline = f' on_response: Callable[[{runion} | Exception], None],'
if len(inline) <= _MAX_LINE:
mid = f'{inline}\n'
else:
inner = f' [{runion} | Exception], None'
if len(inner) <= _MAX_LINE:
mid = ' on_response: Callable[\n' f'{inner}\n' ' ],\n'
else:
parts = runion.split(' | ') + ['Exception']
plines = f' {parts[0]}\n' + ''.join(
f' | {part}\n' for part in parts[1:]
)
mid = (
' on_response: Callable[\n'
' [\n'
f'{plines}'
' ],\n'
' None,\n'
' ],\n'
)
return (
' @overload\n'
' def send_message_cb(\n'
' self,\n'
f' msg: {mpath},\n'
f'{mid}'
' ) -> None: ...\n'
)
def _emit_ret_form(
defline: str, mpath: str, rtext: str, *, indent_extra: str = ''
) -> str:
"""Emit one return-style overload (sync/async/future forms)."""
del indent_extra # Reserved.
argline = f' self, msg: {mpath}\n'
if len(argline) - 1 > _MAX_LINE:
args = f' self,\n msg: {mpath},\n'
else:
args = argline
retline = f' ) -> {rtext}: ...'
if len(retline) > _MAX_LINE:
raise RuntimeError(
f'Response annotation too long to format: {rtext!r}.'
)
return f' @overload\n{defline}\n{args}{retline}\n'
def _emit_sync(mpath: str, runion: str) -> str:
return _emit_ret_form(' def send_message(', mpath, runion)
def _emit_async(mpath: str, runion: str) -> str:
return _emit_ret_form(' async def send_message_async(', mpath, runion)
def _emit_future(mpath: str, runion: str) -> str:
return _emit_ret_form(
' def send_message_future(',
mpath,
f'concurrent.futures.Future[{runion}]',
)
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum