4
4
from functools import wraps
5
5
from threading import Thread
6
6
from queue import Queue as SyncQueue
7
- from idom .core .types import ComponentType
7
+ from typing import Any , Awaitable , Callable
8
+
9
+ from typing_extensions import ParamSpec
8
10
9
11
import ipywidgets as widgets
10
- from IPython .display import display as ipython_display
12
+ from IPython .display import display as ipython_display , DisplayHandle
11
13
from traitlets import Unicode
12
14
from idom .core .layout import Layout , LayoutEvent , LayoutUpdate
13
15
from idom .core .serve import VdomJsonPatch , render_json_patch
14
-
16
+ from idom . core . types import ComponentType
15
17
16
18
_IMPORT_SOURCE_BASE_URL = ""
17
19
18
20
19
- def set_import_source_base_url (base_url ) :
21
+ def set_import_source_base_url (base_url : str ) -> None :
20
22
"""Fallback URL for import sources, if no Jupyter Server is discovered by the client"""
21
23
global _IMPORT_SOURCE_BASE_URL
22
24
_IMPORT_SOURCE_BASE_URL = base_url
23
25
24
26
25
- def run (constructor ) :
27
+ def run (constructor : Callable [[], ComponentType ]) -> DisplayHandle | None :
26
28
"""Run the given IDOM elemen definition as a Jupyter Widget.
27
29
28
30
This function is meant to be similarly to ``idom.run``.
29
31
"""
30
32
return ipython_display (LayoutWidget (constructor ()))
31
33
32
34
33
- def widgetize (constructor ):
35
+ _P = ParamSpec ("_P" )
36
+
37
+
38
+ def widgetize (constructor : Callable [_P , ComponentType ]) -> Callable [_P , LayoutWidget ]:
34
39
"""A decorator that turns an IDOM element into a Jupyter Widget constructor"""
35
40
36
41
@wraps (constructor )
37
- def wrapper (* args , ** kwargs ) :
42
+ def wrapper (* args : Any , ** kwargs : Any ) -> LayoutWidget :
38
43
return LayoutWidget (constructor (* args , ** kwargs ))
39
44
40
45
return wrapper
@@ -63,15 +68,15 @@ class LayoutWidget(widgets.DOMWidget):
63
68
64
69
_import_source_base_url = Unicode ().tag (sync = True )
65
70
66
- def __init__ (self , component : ComponentType ):
71
+ def __init__ (self , component : ComponentType ) -> None :
67
72
super ().__init__ (_import_source_base_url = _IMPORT_SOURCE_BASE_URL )
68
73
self ._idom_model = {}
69
74
self ._idom_views = set ()
70
75
self ._idom_layout = Layout (component )
71
76
self ._idom_loop = _spawn_threaded_event_loop (self ._idom_layout_render_loop ())
72
77
self .on_msg (lambda _ , * args , ** kwargs : self ._idom_on_msg (* args , ** kwargs ))
73
78
74
- def _idom_on_msg (self , message , buffers ):
79
+ def _idom_on_msg (self , message : dict [ str , Any ], buffers : Any ):
75
80
m_type = message .get ("type" )
76
81
if m_type == "client-ready" :
77
82
v_id = message ["viewID" ]
@@ -89,7 +94,7 @@ def _idom_on_msg(self, message, buffers):
89
94
if v_id in self ._idom_views :
90
95
self ._idom_views .remove (message ["viewID" ])
91
96
92
- async def _idom_layout_render_loop (self ):
97
+ async def _idom_layout_render_loop (self ) -> None :
93
98
async with self ._idom_layout :
94
99
while True :
95
100
diff = await render_json_patch (self ._idom_layout )
@@ -101,8 +106,10 @@ def __repr__(self) -> str:
101
106
return f"LayoutWidget({ self ._idom_layout } )"
102
107
103
108
104
- def _spawn_threaded_event_loop (coro ):
105
- loop_q = SyncQueue ()
109
+ def _spawn_threaded_event_loop (
110
+ coro : Callable [..., Awaitable [Any ]]
111
+ ) -> asyncio .AbstractEventLoop :
112
+ loop_q : SyncQueue [asyncio .AbstractEventLoop ] = SyncQueue ()
106
113
107
114
def run_in_thread () -> None :
108
115
loop = asyncio .new_event_loop ()
0 commit comments