-
-
Notifications
You must be signed in to change notification settings - Fork 18
/
layout_handler.py
139 lines (115 loc) · 5.26 KB
/
layout_handler.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
# mypy: disable-error-code=attr-defined
import asyncio
class ReactPyLayoutHandler:
"""Encapsulate the entire layout handler with a class to prevent overlapping
variable names between user code.
This code is designed to be run directly by PyScript, and is not intended to be run
in a normal Python environment.
"""
def __init__(self, uuid):
self.uuid = uuid
@staticmethod
def update_model(update, root_model):
"""Apply an update ReactPy's internal DOM model."""
from jsonpointer import set_pointer
if update["path"]:
set_pointer(root_model, update["path"], update["model"])
else:
root_model.update(update["model"])
def render_html(self, layout, model):
"""Submit ReactPy's internal DOM model into the HTML DOM."""
from pyscript.js_modules import morphdom
import js
# Create a new container to render the layout into
container = js.document.getElementById(f"pyscript-{self.uuid}")
temp_root_container = container.cloneNode(False)
self.build_element_tree(layout, temp_root_container, model)
# Use morphdom to update the DOM
morphdom.default(container, temp_root_container)
# Remove the cloned container to prevent memory leaks
temp_root_container.remove()
def build_element_tree(self, layout, parent, model):
"""Recursively build an element tree, starting from the root component."""
import js
if isinstance(model, str):
parent.appendChild(js.document.createTextNode(model))
elif isinstance(model, dict):
if not model["tagName"]:
for child in model.get("children", []):
self.build_element_tree(layout, parent, child)
return
tag = model["tagName"]
attributes = model.get("attributes", {})
children = model.get("children", [])
element = js.document.createElement(tag)
for key, value in attributes.items():
if key == "style":
for style_key, style_value in value.items():
setattr(element.style, style_key, style_value)
elif key == "className":
element.className = value
else:
element.setAttribute(key, value)
for event_name, event_handler_model in model.get(
"eventHandlers", {}
).items():
self.create_event_handler(
layout, element, event_name, event_handler_model
)
for child in children:
self.build_element_tree(layout, element, child)
parent.appendChild(element)
else:
raise ValueError(f"Unknown model type: {type(model)}")
@staticmethod
def create_event_handler(layout, element, event_name, event_handler_model):
"""Create an event handler for an element. This function is used as an
adapter between ReactPy and browser events."""
from pyodide.ffi.wrappers import add_event_listener
target = event_handler_model["target"]
def event_handler(*args):
asyncio.create_task(
layout.deliver({"type": "layout-event", "target": target, "data": args})
)
event_name = event_name.lstrip("on_").lower().replace("_", "")
add_event_listener(element, event_name, event_handler)
@staticmethod
def delete_old_workspaces():
"""To prevent memory leaks, we must delete all user generated Python code when
it is no longer in use (removed from the page). To do this, we compare what
UUIDs exist on the DOM, versus what UUIDs exist within the PyScript global
interpreter."""
import js
dom_workspaces = js.document.querySelectorAll(".pyscript")
dom_uuids = {element.dataset.uuid for element in dom_workspaces}
python_uuids = {
value.split("_")[-1]
for value in globals()
if value.startswith("user_workspace_")
}
# Delete any workspaces that are not being used
for uuid in python_uuids - dom_uuids:
task_name = f"task_{uuid}"
if task_name in globals():
task: asyncio.Task = globals()[task_name]
task.cancel()
del globals()[task_name]
else:
print(f"Warning: Could not auto delete PyScript task {task_name}")
workspace_name = f"user_workspace_{uuid}"
if workspace_name in globals():
del globals()[workspace_name]
else:
print(
f"Warning: Could not auto delete PyScript workspace {workspace_name}"
)
async def run(self, workspace_function):
"""Run the layout handler. This function is main executor for all user generated code."""
from reactpy.core.layout import Layout
self.delete_old_workspaces()
root_model: dict = {}
async with Layout(workspace_function()) as root_layout:
while True:
update = await root_layout.render()
self.update_model(update, root_model)
self.render_html(root_layout, root_model)