|
@@ -1,3 +1,4 @@
|
|
|
+from dataclasses import dataclass
|
|
|
from pathlib import Path
|
|
|
from typing import Any, Callable, Dict, List, Tuple
|
|
|
|
|
@@ -6,14 +7,38 @@ from typing_extensions import Self
|
|
|
from .. import background_tasks, globals
|
|
|
from ..dependencies import register_vue_component
|
|
|
from ..element import Element
|
|
|
-from ..helpers import is_coroutine
|
|
|
+from ..helpers import KWONLY_SLOTS, is_coroutine
|
|
|
|
|
|
register_vue_component(name='refreshable', path=Path(__file__).parent.joinpath('refreshable.js'))
|
|
|
|
|
|
|
|
|
+@dataclass(**KWONLY_SLOTS)
|
|
|
+class RefreshableTarget:
|
|
|
+ container: Element
|
|
|
+ instance: Any
|
|
|
+ args: List[Any]
|
|
|
+ kwargs: Dict[str, Any]
|
|
|
+
|
|
|
+ def run(self, func: Callable[..., Any]) -> None:
|
|
|
+ if is_coroutine(func):
|
|
|
+ async def wait_for_result() -> None:
|
|
|
+ with self.container:
|
|
|
+ if self.instance is None:
|
|
|
+ await func(*self.args, **self.kwargs)
|
|
|
+ else:
|
|
|
+ await func(self.instance, *self.args, **self.kwargs)
|
|
|
+ return wait_for_result()
|
|
|
+ else:
|
|
|
+ with self.container:
|
|
|
+ if self.instance is None:
|
|
|
+ func(*self.args, **self.kwargs)
|
|
|
+ else:
|
|
|
+ func(self.instance, *self.args, **self.kwargs)
|
|
|
+
|
|
|
+
|
|
|
class refreshable:
|
|
|
|
|
|
- def __init__(self, func: Callable) -> None:
|
|
|
+ def __init__(self, func: Callable[..., Any]) -> None:
|
|
|
"""Refreshable UI functions
|
|
|
|
|
|
The `@ui.refreshable` decorator allows you to create functions that have a `refresh` method.
|
|
@@ -21,24 +46,27 @@ class refreshable:
|
|
|
"""
|
|
|
self.func = func
|
|
|
self.instance = None
|
|
|
- self.containers: List[Tuple[Element, List[Any], Dict[str, Any]]] = []
|
|
|
+ self.targets: List[RefreshableTarget] = []
|
|
|
|
|
|
def __get__(self, instance, _) -> Self:
|
|
|
self.instance = instance
|
|
|
return self
|
|
|
|
|
|
- def __call__(self, *args, **kwargs) -> None:
|
|
|
+ def __call__(self, *args: Any, **kwargs: Any) -> None:
|
|
|
self.prune()
|
|
|
- with Element('refreshable') as container:
|
|
|
- container.use_component('refreshable')
|
|
|
- self.containers.append((container, args, kwargs))
|
|
|
- return self.func(*args, **kwargs) if self.instance is None else self.func(self.instance, *args, **kwargs)
|
|
|
+ container = Element('refreshable')
|
|
|
+ container.use_component('refreshable')
|
|
|
+ target = RefreshableTarget(container=container, instance=self.instance, args=args, kwargs=kwargs)
|
|
|
+ self.targets.append(target)
|
|
|
+ return target.run(self.func)
|
|
|
|
|
|
def refresh(self) -> None:
|
|
|
self.prune()
|
|
|
- for container, args, kwargs in self.containers:
|
|
|
- container.clear()
|
|
|
- result = self._run_in_container(container, *args, **kwargs)
|
|
|
+ for target in self.targets:
|
|
|
+ if target.instance != self.instance:
|
|
|
+ continue
|
|
|
+ target.container.clear()
|
|
|
+ result = target.run(self.func)
|
|
|
if is_coroutine(self.func):
|
|
|
if globals.loop and globals.loop.is_running():
|
|
|
background_tasks.create(result)
|
|
@@ -46,24 +74,4 @@ class refreshable:
|
|
|
globals.app.on_startup(result)
|
|
|
|
|
|
def prune(self) -> None:
|
|
|
- self.containers = [
|
|
|
- (container, args, kwargs)
|
|
|
- for container, args, kwargs in self.containers
|
|
|
- if container.client.id in globals.clients
|
|
|
- ]
|
|
|
-
|
|
|
- def _run_in_container(self, container: Element, *args, **kwargs) -> None:
|
|
|
- if is_coroutine(self.func):
|
|
|
- async def wait_for_result() -> None:
|
|
|
- with container:
|
|
|
- if self.instance is None:
|
|
|
- await self.func(*args, **kwargs)
|
|
|
- else:
|
|
|
- await self.func(self.instance, *args, **kwargs)
|
|
|
- return wait_for_result()
|
|
|
- else:
|
|
|
- with container:
|
|
|
- if self.instance is None:
|
|
|
- self.func(*args, **kwargs)
|
|
|
- else:
|
|
|
- self.func(self.instance, *args, **kwargs)
|
|
|
+ self.targets = [target for target in self.targets if target.container.client.id in globals.clients]
|