# Released under the MIT License. See LICENSE for details.
#
"""Controller functionality for DocUI."""
from __future__ import annotations # Docs-generation hack.
# This is the primary hand-written doc-ui controller module. The
# cleanly-separable pieces (the bg-thread runner and shared types)
# already live in their own modules; what remains is cohesive controller
# logic, so allow it to run a bit long.
# pylint: disable=too-many-lines
from typing import TYPE_CHECKING, assert_never
from dataclasses import dataclass
from enum import Enum
import weakref
from efro.util import asserttype
from efro.error import CleanError, CommunicationError
from efro.dataclassio import dataclass_to_json, dataclass_from_json
from bacommon.docui import (
DocUIRequestTypeID,
UnknownDocUIRequest,
DocUIResponseTypeID,
UnknownDocUIResponse,
DocUIWebRequest,
DocUIWebResponse,
)
import bauiv1 as bui
from bauiv1 import builtinassets
from bauiv1lib.docui import _bgrunner
from bauiv1lib.docui._types import DocUILocalAction
from bauiv1lib.docui._window import DocUIWindow
if TYPE_CHECKING:
from typing import Callable
import bacommon.docui.v2
from bacommon.docui import DocUIRequest, DocUIResponse
from bacommon.langstr import LangStrSpec
import bacommon.clienteffect as clfx
from bauiv1lib.docui import prep
class _WinState(Enum):
"""Per-window state."""
FETCHING_FRESH_REQUEST = 0
REDISPLAYING_OLD_STATE = 1
REFRESHING = 2
ERRORED = 3
IDLE = 4
@dataclass
class _WinData:
state: _WinState
refresh_timer: bui.AppTimer | None = None
[docs]
class DocUIController:
"""Manages interactions between DocUI clients and servers.
Can include logic to handle all requests locally or can submit them
to be handled by some server or can do some combination thereof.
"""
[docs]
class ErrorType(Enum):
"""Types of errors that can occur in request processing."""
GENERIC = 'generic'
UNDER_CONSTRUCTION = 'under_construction'
COMMUNICATION_ERROR = 'communication'
NEED_UPDATE = 'need_update'
[docs]
def fulfill_request(self, request: DocUIRequest) -> DocUIResponse:
"""Handle request fulfillment.
Expected to be overridden by child classes.
Be aware that this will always be called in a background thread.
This method is expected to always return a response, even in the
case of errors. Use
:meth:`~bauiv1lib.docui.DocUIController.error_response()` to
translate error conditions to responses.
The one exception to this rule (no pun intended) is the
:class:`efro.error.CleanError` exception. This can be raised as a
quick and dirty way to show custom error messages. The code
``raise CleanError('Something broke.')`` will have the same
effect as ``return self.error_response(custom_message='Something
broke.')``.
"""
raise NotImplementedError()
[docs]
def local_action(self, action: DocUILocalAction) -> None:
"""Do something locally on behalf of the doc-ui.
Controller classes can override this to expose named actions
that can be triggered by doc-ui button presses, responses,
etc.
Of course controllers can also perform arbitrary local actions
alongside their normal request fulfillment; this is simply a way
to do so without needing to provide actual ui pages alongside.
Be *very* careful and focused with what you expose here,
especially if your doc-ui pages are coming from untrusted
sources. Generally things like launching or joining games are
good candidates for local actions.
"""
[docs]
def fulfill_request_web(
self, request: DocUIRequest, url: str
) -> DocUIResponse:
"""Fulfill a request by sending it to a webserver."""
import bacommon.docui.v1 as dui1
import bacommon.docui.v2 as dui2
import urllib3.util
if not isinstance(request, (dui1.Request, dui2.Request)):
raise RuntimeError(f'Unsupported docui request: {type(request)}')
# The v1 and v2 method enums share wire values; normalize to v1
# for our http dispatch below.
method = dui1.RequestMethod(request.method.value)
upool = bui.app.net.urllib3pool
# Allow compressed results.
headers = urllib3.util.make_headers(accept_encoding=True)
# Bundle our doc-ui request with some extra stuff that might
# be relevant to a remote server (language we're using, etc.).
webrequest = DocUIWebRequest(
doc_ui_request=request,
locale=bui.app.locale.current_locale,
engine_build_number=bui.app.env.engine_build_number,
)
try:
# Map docui GET requests to http GET and POST to POST.
if method is dui1.RequestMethod.GET:
# For GET we embed the request into a url param.
raw_response = upool.request(
'GET',
url,
fields={
'doc_ui_web_request': dataclass_to_json(webrequest)
},
headers=headers,
)
elif method is dui1.RequestMethod.POST:
# for POST we send the webrequest as json in body.
headers['Content-Type'] = 'application/json'
raw_response = upool.request(
'POST',
url,
headers=headers,
body=dataclass_to_json(webrequest),
)
elif method is dui1.RequestMethod.UNKNOWN:
raise RuntimeError('Unknown request method.')
else:
assert_never(method)
try:
# We use 'lossy' here so response versions or elements
# that we don't know about will come through as
# 'Unknown' types instead of erroring completely.
webresponse = dataclass_from_json(
DocUIWebResponse, raw_response.data.decode(), lossy=True
)
if (
webresponse.error is None
and webresponse.doc_ui_response is None
):
raise RuntimeError(
'Invalid webresponse includes neither error'
' nor doc-ui-response.'
)
except Exception as exc:
bui.netlog.info(
'Error reading docui web-response.', exc_info=True
)
raise RuntimeError('Error reading docui web-response.') from exc
# For now, consider all errors communication errors (should
# result in retry buttons in some cases). Can get more
# specific in the future for cases where retries would not
# help.
if raw_response.status != 200:
# If the response bundled an error, log it.
if webresponse.error is not None:
bui.netlog.info(
'doc-ui http request returned error: %s',
webresponse.error,
)
return self.error_response(
request, self.ErrorType.COMMUNICATION_ERROR
)
except Exception:
# For now, consider all errors communication errors (should
# result in retry buttons in some cases). Can get more
# specific in the future for cases where retries would not
# help.
bui.netlog.info('Error in docui http request.', exc_info=True)
return self.error_response(
request, self.ErrorType.COMMUNICATION_ERROR
)
assert webresponse.doc_ui_response is not None
self._check_server_response(webresponse.doc_ui_response)
return webresponse.doc_ui_response
def fulfill_request_cloud(
self, request: DocUIRequest, domain: str
) -> DocUIResponse:
"""Fulfill a request by sending it to ballistica's cloud.
:meta private:
"""
import bacommon.cloud
bui.uilog.debug(
'Fetching doc-ui request from cloud (domain=%r).', domain
)
try:
plus = bui.app.plus
if plus is None:
raise RuntimeError('Plus not available.')
account = plus.accounts.primary
if account is not None:
with account:
mresponse = plus.cloud.send_message(
bacommon.cloud.FulfillDocUIRequest(
request=request, domain=domain
)
)
else:
mresponse = plus.cloud.send_message(
bacommon.cloud.FulfillDocUIRequest(
request=request, domain=domain
)
)
assert isinstance(mresponse, bacommon.cloud.FulfillDocUIResponse)
self._check_server_response(mresponse.response)
return mresponse.response
except CommunicationError as exc:
# Label comm-errors so we can possibly show retry buttons.
# Expected/transient (bad network, server hiccup), so warn
# rather than dumping a traceback.
bui.uilog.warning(
'Communication error fetching doc-ui (domain=%r): %s',
domain,
exc,
)
return self.error_response(
request, self.ErrorType.COMMUNICATION_ERROR
)
except Exception:
# Unexpected; this is a real bug worth a full traceback.
bui.uilog.exception(
'Unexpected error fetching doc-ui (domain=%r).', domain
)
return self.error_response(request)
@staticmethod
def _check_server_response(response: DocUIResponse) -> None:
"""Run diagnostics on a pristine server-supplied response.
Called at the receipt points (cloud/web fulfillment) — before
controllers splice in any local content — so finalization
checks only see what the server actually sent.
"""
import bacommon.docui.v2 as dui2
from bauiv1lib.docui import _resolve
if isinstance(response, dui2.Response):
_resolve.check_finalization_leaks(response)
[docs]
def error_response(
self,
request: DocUIRequest,
error_type: ErrorType = ErrorType.GENERIC,
custom_message: str | None = None,
) -> DocUIResponse:
"""Build a simple error message page.
A message is included based on ``error_type``. Pass
``custom_message`` to override this.
Messages are language-agnostic (bundled-package strings), so
error pages localize like any other doc-ui content; a
``custom_message`` shows verbatim (untranslated).
"""
import bacommon.docui.v2 as dui2
from bacommon.langstr import LangStrSpecValue
from bauiv1 import _commonassets
uiact = _commonassets.strings.actions
uistat = _commonassets.strings.status
uival = _commonassets.strings.values
error_msg: LangStrSpec
status_code = dui2.ResponseStatus.UNKNOWN_ERROR
if custom_message is not None:
error_msg = LangStrSpecValue(custom_message)
elif error_type is self.ErrorType.GENERIC:
error_msg = uistat.error_occurred.spec
elif error_type is self.ErrorType.NEED_UPDATE:
error_msg = uistat.need_update.spec
elif error_type is self.ErrorType.UNDER_CONSTRUCTION:
error_msg = uistat.under_construction.spec
elif error_type is self.ErrorType.COMMUNICATION_ERROR:
status_code = dui2.ResponseStatus.COMMUNICATION_ERROR
error_msg = uistat.server_error.spec
else:
assert_never(error_type)
debug = False
# Give a retry button for comm-errors on GET requests (POSTs may
# have unintentional side-effects so holding off on those for
# now).
do_retry = (
isinstance(request, dui2.Request)
and request.method is dui2.RequestMethod.GET
and status_code is dui2.ResponseStatus.COMMUNICATION_ERROR
)
return dui2.Response(
status=status_code,
page=dui2.Page(
title=uival.error.spec,
center_vertically=True,
rows=[
dui2.ButtonRow(
buttons=[
dui2.Button(
(uiact.retry if do_retry else uiact.ok).spec,
action=(
dui2.Replace(
asserttype(request, dui2.Request)
)
if do_retry
else dui2.Local(close_window=True)
),
default=True,
style=dui2.ButtonStyle.MEDIUM,
size=(130, 50),
padding_left=200,
padding_right=200,
padding_top=100,
decorations=[
dui2.Text(
error_msg,
position=(0, 80),
size=(480, 50),
highlight=False,
debug=debug,
),
],
debug=debug,
),
],
center_content=True,
debug=debug,
),
],
),
)
[docs]
def create_window(
self,
request: DocUIRequest,
*,
transition: str | None = 'in_right',
origin_widget: bui.Widget | None = None,
auxiliary_style: bool = True,
uiopenstateid: str | None = None,
suppress_win_extra_type_warning: bool = False,
) -> DocUIWindow:
"""Create a new window to handle a request."""
assert bui.in_logic_thread()
# Create a shiny new window.
win = DocUIWindow(
self,
request,
transition=transition,
origin_widget=origin_widget,
auxiliary_style=auxiliary_style,
uiopenstateid=uiopenstateid,
suppress_win_extra_type_warning=suppress_win_extra_type_warning,
)
self._set_win_data(win, _WinData(_WinState.FETCHING_FRESH_REQUEST))
# Lock its ui and kick off a bg task to populate it.
win.lock_ui()
_bgrunner.submit(
bui.CallStrict(
self._process_request_in_bg,
request,
weakwin=weakref.ref(win),
uiscale=bui.app.ui_v1.uiscale,
scroll_width=win.scroll_width,
scroll_height=win.scroll_height,
idprefix=win.main_window_id_prefix,
immediate=False,
)
)
return win
[docs]
def save_window_shared_state(
self, window: DocUIWindow, state: dict
) -> None:
"""Called when a window shared state is being saved."""
del window, state # Unused.
[docs]
def restore_window_shared_state(
self, window: DocUIWindow, state: dict
) -> None:
"""Called when a window shared state is being restored."""
del window, state # Unused.
[docs]
def restore(
self,
win: DocUIWindow,
*,
last_response: DocUIResponse | None,
has_had_response: bool,
) -> DocUIWindow:
"""Restore a window from previous state.
May immediately display old results or may kick off a new
request.
"""
import bacommon.docui.v2 as dui2
assert bui.in_logic_thread()
explicit_response: DocUIResponse | None = None
explicit_error: DocUIController.ErrorType | None = None
if last_response is not None:
# Re-prep our restored response so we have something to show
# immediately. We'll then fetch an updated version in the
# background to get the latest version.
explicit_response = last_response
else:
# We have no previous response to restore. Fetch a new one.
# If the current request is a POST, never auto-refetch. Just
# build an error response.
assert isinstance(win.request, dui2.Request)
if win.request.method is dui2.RequestMethod.POST:
# Do we want a specific error for this? Though this case
# should be rare I think.
explicit_error = self.ErrorType.GENERIC
else:
explicit_error = None
# We're either errored or redisplaying an old state.
if explicit_error is None:
self._set_win_data(win, _WinData(_WinState.REDISPLAYING_OLD_STATE))
else:
self._set_win_data(win, _WinData(_WinState.ERRORED))
# Lock the ui and kick off this update.
win.lock_ui()
_bgrunner.submit(
bui.CallStrict(
self._process_request_in_bg,
win.request,
weakwin=weakref.ref(win),
uiscale=bui.app.ui_v1.uiscale,
scroll_width=win.scroll_width,
scroll_height=win.scroll_height,
idprefix=win.main_window_id_prefix,
# If this window has had a response already, snap things
# in immediately with no transitions.
immediate=has_had_response,
explicit_error=explicit_error,
explicit_response=explicit_response,
)
)
return win
[docs]
def replace(
self,
win: DocUIWindow,
request: DocUIRequest,
*,
origin_widget: bui.Widget | None = None,
is_refresh: bool = False,
) -> None:
"""Kick off a request to replace existing window contents."""
import bacommon.docui.v2 as dui2
assert bui.in_logic_thread()
win.request = request
requesttype = request.get_type_id()
if requesttype is DocUIRequestTypeID.V1:
# This client no longer works in v1.
bui.uilog.error('Got v1 doc-ui request; this is unsupported.')
self._submit_fresh_request(
win,
origin_widget,
is_refresh,
explicit_error=self.ErrorType.GENERIC,
)
elif requesttype is DocUIRequestTypeID.V2:
assert isinstance(win.request, dui2.Request)
self._submit_fresh_request(win, origin_widget, is_refresh)
elif requesttype is DocUIRequestTypeID.UNKNOWN:
# Got a request type we don't know. Show a 'need a newer
# build' error.
assert isinstance(win.request, UnknownDocUIRequest)
self._submit_fresh_request(
win,
origin_widget,
is_refresh,
explicit_error=self.ErrorType.NEED_UPDATE,
)
else:
assert_never(requesttype)
def _submit_fresh_request(
self,
win: DocUIWindow,
origin_widget: bui.Widget | None,
is_refresh: bool,
*,
explicit_error: DocUIController.ErrorType | None = None,
) -> None:
"""Lock the ui and kick off a fresh request's bg processing."""
# Timers (docui timed-actions especially) can still fire after
# the app threadpool is torn down; bow out quietly instead of
# erroring on submit once shutdown has begun.
if bui.app.shutting_down:
return
self._set_win_data(
win,
_WinData(
_WinState.ERRORED
if explicit_error is not None
else (
_WinState.REFRESHING
if is_refresh
else _WinState.FETCHING_FRESH_REQUEST
)
),
)
win.lock_ui(origin_widget)
_bgrunner.submit(
bui.CallStrict(
self._process_request_in_bg,
win.request,
weakwin=weakref.ref(win),
uiscale=bui.app.ui_v1.uiscale,
scroll_width=win.scroll_width,
scroll_height=win.scroll_height,
idprefix=win.main_window_id_prefix,
immediate=True,
explicit_error=explicit_error,
)
)
[docs]
def run_action(
self,
window: DocUIWindow,
widgetid: str | None,
action: bacommon.docui.v2.Action | None,
is_timed: bool = False,
) -> None:
"""Called when a button is pressed in a doc-ui."""
# pylint: disable=too-many-branches
# pylint: disable=cyclic-import
import bacommon.docui.v2 as dui
assert bui.in_logic_thread()
# If locked, been and tell them to try again.
if window.locked:
builtinassets.audio.error.get().play()
from bauiv1 import _commonassets
bui.screenmessage(
_commonassets.strings.status.page_refreshing_try_again,
color=(1, 0, 0),
)
return
widget: bui.Widget | None
if widgetid is not None:
# Find the associated button.
widget = bui.widget_by_id(widgetid)
if widget is None:
bui.uilog.warning(
'DocUI button press widget not found: %s (not expected)',
widgetid,
)
return
else:
widget = None
# Play error beeps on buttons with no actions assigned to let
# the user know nothing is supposed to happen.
if action is None:
builtinassets.audio.error.get().play()
return
action_type = action.get_type_id()
if action_type is dui.ActionTypeID.BROWSE:
assert isinstance(action, dui.Browse)
if is_timed:
# Don't let timers pop up new windows. Untrusted servers
# would have a field-day with this.
bui.uilog.warning(
'Ignoring BROWSE action (disallowed in timed actions).'
)
else:
if action.default_sound:
builtinassets.audio.swish.get().play()
window.main_window_replace(
lambda: self.create_window(
action.request,
origin_widget=widget,
auxiliary_style=False,
suppress_win_extra_type_warning=True,
)
)
elif action_type is dui.ActionTypeID.REPLACE:
assert isinstance(action, dui.Replace)
# Play default click sound only if this is coming from a
# button.
if widget is not None and action.default_sound:
builtinassets.audio.click01.get().play()
# Force a state save so if our UI gets rebuilt with the same
# IDs we'll wind up with the same selection and whatnot.
window.main_window_save_shared_state()
self.replace(window, action.request, origin_widget=widget)
elif action_type is dui.ActionTypeID.LOCAL:
assert isinstance(action, dui.Local)
if action.default_sound:
if action.close_window:
# Always play close-window swish, even if we don't have
# a source button.
builtinassets.audio.swish.get().play()
else:
# Only play click sound if this is coming from a button.
if widget is not None:
builtinassets.audio.click01.get().play()
if action.close_window:
window.main_window_back()
self._run_immediate_effects_and_actions(
client_effects=action.immediate_client_effects,
local_action=action.immediate_local_action,
local_action_args=action.immediate_local_action_args,
widget=widget,
window=window,
is_timed=is_timed,
)
elif action_type is dui.ActionTypeID.UNKNOWN:
assert isinstance(action, dui.UnknownAction)
bui.screenmessage('Unknown action.', color=(1, 0, 0))
builtinassets.audio.error.get().play()
else:
# Make sure we handle all options.
assert_never(action_type)
def _run_immediate_effects_and_actions(
self,
*,
client_effects: list[clfx.Effect],
local_action: str | None,
local_action_args: dict | None,
widget: bui.Widget | None,
window: DocUIWindow,
is_timed: bool,
) -> None:
# We don't allow timed actions to trigger immediate
# client-effects/local-actions. It would be too easy for such
# things to get unintentionally re-triggered when navigating
# back/etc. We only want those to happen due to direct button
# presses or initial (non-refresh) responses, which should keep
# things feeling mostly intentional.
if is_timed:
if client_effects:
bui.uilog.warning(
'Ignoring client-effects (disallowed in timed actions).'
)
if local_action is not None:
bui.uilog.warning(
'Ignoring local-action (disallowed in timed actions).'
)
return
if bui.app.classic is not None and client_effects:
bui.app.classic.run_bs_client_effects(client_effects)
if local_action is not None:
try:
self.local_action(
DocUILocalAction(
name=local_action,
args=(
{}
if local_action_args is None
else local_action_args
),
widget=widget,
window=window,
)
)
except Exception:
bui.uilog.exception(
'Error running local-action %s.',
local_action,
)
def _get_win_data(self, window: DocUIWindow) -> _WinData:
val = getattr(window, '_wcdata')
assert isinstance(val, _WinData)
return val
def _set_win_data(self, window: DocUIWindow, data: _WinData) -> None:
setattr(window, '_wcdata', data)
def _process_request_in_bg(
self,
request: DocUIRequest,
*,
weakwin: weakref.ref[DocUIWindow],
uiscale: bui.UIScale,
scroll_width: float,
scroll_height: float,
idprefix: str,
immediate: bool,
explicit_error: ErrorType | None = None,
explicit_response: DocUIResponse | None = None,
) -> None:
"""Wrangle a request from within a background thread.
This will always return a response, even on error conditions.
"""
# pylint: disable=cyclic-import
import bacommon.docui.v2 as dui2
from bauiv1lib.docui import prep
assert not bui.in_logic_thread()
response: DocUIResponse | None = None
error: DocUIController.ErrorType | None = None
if explicit_error is not None:
error = explicit_error
elif explicit_response is not None:
response = explicit_response
else:
try:
response = self.fulfill_request(request)
except CleanError as exc:
# The one exception case we officially handle. Translate
# this to an error response with a custom message.
response = self.error_response(request, custom_message=str(exc))
except Exception:
# fulfill_request is expected to gracefully return even
# on errors. Make noise if it didn't.
bui.uilog.exception(
'Error in fulfill_request().\n'
'It should always return responses; not throw exceptions.\n'
'Use error_response() when errors occur.',
exc_info=True,
)
error = self.ErrorType.GENERIC
# Validate any response we got.
if response is not None:
assert error is None
responsetype = response.get_type_id()
if responsetype is DocUIResponseTypeID.V1:
# This client no longer works in v1 (servers serve v2
# to any build with v2 support, so this implies either
# a server bug or a v1-only mod controller).
bui.uilog.error('Got v1 doc-ui response; this is unsupported.')
error = self.ErrorType.GENERIC
response = None
elif responsetype is DocUIResponseTypeID.V2:
assert isinstance(response, dui2.Response)
minbuild = response.minimum_engine_build
if (
minbuild is not None
and minbuild > bui.app.env.engine_build_number
):
bui.uilog.debug(
'doc-ui response requires engine build %d but we'
' are %d; showing need-update prompt.',
minbuild,
bui.app.env.engine_build_number,
)
error = self.ErrorType.NEED_UPDATE
response = None
else:
try:
# Resolve referenced packages in our locale and
# de-index deferred effects; the page then preps
# and renders natively.
from bauiv1lib.docui import _resolve
_resolve.resolve_response(response)
except Exception:
bui.uilog.exception(
'Error resolving v2 doc-ui response.'
)
error = self.ErrorType.GENERIC
response = None
elif responsetype is DocUIResponseTypeID.UNKNOWN:
assert isinstance(response, UnknownDocUIResponse)
bui.uilog.debug(
'Got unsupported docui response.', exc_info=True
)
error = self.ErrorType.NEED_UPDATE
response = None
else:
# Make sure we cover all types we're aware of.
assert_never(responsetype)
if error is not None:
response = self.error_response(request, error)
# Currently must be v2 if it made it to here.
assert isinstance(response, dui2.Response)
pageprep = prep.prep_page(
response.page,
packages=list(response.packages),
uiscale=uiscale,
scroll_width=scroll_width,
scroll_height=scroll_height,
immediate=immediate,
idprefix=idprefix,
)
# Go ahead and just push the response along with our weakref
# back to the logic thread for handling. We could quick-out here
# if the window is dead, but wrangling its refs here could
# theoretically lead to it being deallocated here which could be
# problematic.
bui.pushcall(
bui.CallStrict(
self._handle_response_in_ui_thread,
response,
weakwin,
pageprep,
),
from_other_thread=True,
)
def _handle_response_in_ui_thread(
self,
response: DocUIResponse,
weakwin: weakref.ref[DocUIWindow],
pageprep: prep.PagePrep,
) -> None:
import bacommon.docui.v2 as dui2
assert bui.in_logic_thread()
# If our target window died since we made the request, no
# biggie.
win = weakwin()
if win is None:
return
# Currently should only be sending ourself v2 responses here.
assert isinstance(response, dui2.Response)
win.unlock_ui()
win.set_last_response(
response,
response.status == dui2.ResponseStatus.SUCCESS,
)
# Set the UI.
win.instantiate_ui(pageprep)
state = self._get_win_data(win).state
# Run client-effects and local-actions ONLY after fresh requests
# (don't want sounds and other actions firing when we navigate
# back or resize a window).
if state is _WinState.FETCHING_FRESH_REQUEST:
if response.client_effects and bui.app.classic is not None:
bui.app.classic.run_bs_client_effects(response.client_effects)
if response.local_action is not None:
try:
self.local_action(
DocUILocalAction(
name=response.local_action,
args=(
{}
if response.local_action_args is None
else response.local_action_args
),
widget=None,
window=win,
)
)
except Exception:
bui.uilog.exception(
'Error running local-action %s.',
response.local_action,
)
# Possibly take further action depending on state.
if state is _WinState.REDISPLAYING_OLD_STATE:
# Ok; we're done showing old state. For GET we can now kick off
# a refresh to swap in the latest version of the page; for POST
# this is as far as we go (don't want to repeat POST effects).
# (win.request stays the original v1-or-v2 request here.)
from bauiv1lib.docui import _resolve
if _resolve.request_is_get(win.request):
self.replace(win, win.request, is_refresh=True)
else:
self._set_idle_and_schedule_timed_action(response, weakwin)
elif state is _WinState.ERRORED or state is _WinState.IDLE:
pass
elif (
state is _WinState.FETCHING_FRESH_REQUEST
or state is _WinState.REFRESHING
):
self._set_idle_and_schedule_timed_action(response, weakwin)
else:
assert_never(state)
def _set_idle_and_schedule_timed_action(
self, response: DocUIResponse, weakwin: weakref.ref[DocUIWindow]
) -> None:
import bacommon.docui.v2 as dui2
win = weakwin()
assert win is not None
assert self._get_win_data(win).state is not _WinState.IDLE
assert isinstance(response, dui2.Response)
refresh_timer: bui.AppTimer | None = None
if response.timed_action is not None:
# Limit delay to .25 seconds or more to prevent excessive
# churn. Can revisit if there is a strong use case.
refresh_timer = bui.AppTimer(
max(0.250, response.timed_action_delay),
bui.WeakCallStrict(
self._run_timed_action,
weakwin,
response.timed_action,
),
)
self._set_win_data(win, _WinData(_WinState.IDLE, refresh_timer))
def _run_timed_action(
self,
weakwin: weakref.ref[DocUIWindow],
action: bacommon.docui.v2.Action,
) -> None:
# If our target window died since we set this timer, no biggie.
win = weakwin()
if win is None:
return
state = self._get_win_data(win).state
if state is not _WinState.IDLE:
bui.uilog.warning(
'win has non-idle state in _run_timed_action; not expected'
)
return
self.run_action(win, widgetid=None, action=action, is_timed=True)
# 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