pythonnative/src/pythonnative/events.py at v0.43.1 · pythonnative/pythonnative · GitHub

Repository navigation

History

228 lines (181 loc) · 8.18 KB

Raw

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

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

"""Tag-based event routing between native views and Python callbacks.

Event props such as ``on_press`` and ``on_change`` stay in a Python

registry. Native views send a tag and event name through the bridge:

- The reconciler strips callable props out of the payload sent to

native handlers and registers them here, keyed by ``(tag, name)``.

- Handlers wire their platform listener **once** at view creation; the

listener calls [`dispatch_event`][pythonnative.events.dispatch_event]

with the view's tag and the event name.

- Re-renders only mutate this Python-side registry; no native call is

made when just a callback identity changes.

The set of event names present on an element is forwarded to handlers

under the [`EVENTS_PROP`][pythonnative.events.EVENTS_PROP] key (a

``frozenset``), so handlers that wire expensive listeners (scroll

delegates, gesture recognizers) can do so conditionally. Dispatching an

event nobody listens to is a cheap dict miss.

"""

import threading

from typing import Any, Callable, Dict, FrozenSet, Optional, Tuple

EVENTS_PROP = "_pn_events"

"""Prop key carrying the ``frozenset`` of event names wired on an element."""

GESTURES_PROP = "gestures"

"""Prop key carrying gesture descriptors (see ``pythonnative.gestures``)."""

# Prop dicts that may carry nested callables, mapped to the event name

# each nested key is hoisted to.

_NESTED_EVENT_PROPS: Dict[str, Dict[str, str]] = {

"refresh_control": {"on_refresh": "on_refresh"},

}

class EventRegistry:

"""Process-wide map of ``(tag, event name) -> Python callback``.

Registration and lookup are protected by a lock. Bridge backends queue

native input to the Python application thread before dispatching it.

"""

def __init__(self) -> None:

self._lock = threading.Lock()

self._scopes: Dict[int, Any] = {}

self._callbacks: Dict[int, Dict[str, Callable[..., Any]]] = {}

def set_events(self, tag: int, events: Dict[str, Callable[..., Any]]) -> None:

"""Replace every registration for ``tag`` with ``events``."""

with self._lock:

if events:

self._callbacks[tag] = dict(events)

else:

self._callbacks.pop(tag, None)

def set_scope(self, tag: int, scope: Any) -> None:

"""Associate callbacks with the component that owns this view."""

self._scopes[tag] = scope

def invoke(self, tag: int, name: str, *args: Any) -> Any:

"""Invoke a sync or async handler with component cancellation."""

from .runtime import invoke

callback = self.get(tag, name)

if callback is not None:

return invoke(callback, *args, scope=self._scopes.get(tag))

return None

def clear(self, tag: int) -> None:

"""Drop every registration for ``tag`` (called on view destroy)."""

with self._lock:

self._callbacks.pop(tag, None)

self._scopes.pop(tag, None)

def get(self, tag: int, name: str) -> Optional[Callable[..., Any]]:

"""Return the callback for ``(tag, name)``, or ``None``."""

with self._lock:

bucket = self._callbacks.get(tag)

if bucket is None:

return None

return bucket.get(name)

def has(self, tag: int, name: str) -> bool:

"""Return whether a callback is registered for ``(tag, name)``."""

return self.get(tag, name) is not None

def dispatch(self, tag: int, name: str, *args: Any) -> bool:

"""Invoke the callback for ``(tag, name)`` with ``args``.

Returns:

``True`` when a callback existed and was invoked (even if

it raised: exceptions never propagate into the platform's

UI thread; in dev mode they are routed to the RedBox via

[`report_error`][pythonnative.diagnostics.report_error],

otherwise the traceback is printed), ``False`` when nothing

is registered.

"""

callback = self.get(tag, name)

if callback is None:

return False

try:

self.invoke(tag, name, *args)

except Exception as exc:

from . import diagnostics

if not diagnostics.report_error(exc, phase=f"event {name!r}"):

import traceback

traceback.print_exc()

return True

def reset(self) -> None:

"""Drop every registration (test helper)."""

with self._lock:

self._callbacks.clear()

self._scopes.clear()

_registry = EventRegistry()

def get_event_registry() -> EventRegistry:

"""Return the process-wide [`EventRegistry`][pythonnative.events.EventRegistry]."""

return _registry

def dispatch_event(tag: int, name: str, *args: Any) -> bool:

"""Dispatch an event from a native view into Python.

This is the single entry point platform handlers call when a

native listener fires.

Args:

tag: The view's reconciler-assigned tag.

name: Event name, the original prop name (``"on_press"``,

``"on_change"``, …) or a gesture channel (``"gesture:0"``).

*args: Positional arguments forwarded to the user callback,

preserving each prop's documented signature.

Returns:

Whether a callback was registered for ``(tag, name)``.

"""

return _registry.dispatch(tag, name, *args)

# ======================================================================

# Prop splitting

# ======================================================================

def extract_events(props: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Callable[..., Any]]]:

"""Split ``props`` into native-safe props and Python event callbacks.

Rules:

- Top-level callables named ``on_*`` become events under their prop

name and are removed from the native payload.

- ``refresh_control`` dicts have their nested ``on_refresh``

hoisted to the ``"on_refresh"`` event; the remaining keys

(``refreshing``, ``tint_color``) stay in the payload.

- ``gestures`` lists of gesture descriptors are serialized to plain

dicts (handlers wire recognizers from them) while their callbacks

are folded into per-gesture ``"gesture:<i>"`` routers.

- The resulting payload carries ``_pn_events`` (a frozenset of the

event names present), so handlers can wire listeners

conditionally and the prop differ can detect listener

addition/removal without comparing closures.

Args:

props: Raw element props (already stripped of reconciler-owned

keys).

Returns:

``(clean_props, events)`` where ``clean_props`` contains no

callables and ``events`` maps event names to callbacks.

"""

clean: Dict[str, Any] = {}

events: Dict[str, Callable[..., Any]] = {}

for key, value in props.items():

if key.startswith("on_") and callable(value):

events[key] = value

continue

nested_spec = _NESTED_EVENT_PROPS.get(key)

if nested_spec is not None and isinstance(value, dict):

remainder: Dict[str, Any] = {}

for nested_key, nested_value in value.items():

event_name = nested_spec.get(nested_key)

if event_name is not None and callable(nested_value):

events[event_name] = nested_value

else:

remainder[nested_key] = nested_value

clean[key] = remainder

continue

if key == GESTURES_PROP and value:

from .gestures import serialize_gestures

specs, gesture_events = serialize_gestures(value)

clean[key] = specs

events.update(gesture_events)

continue

clean[key] = value

if events:

clean[EVENTS_PROP] = frozenset(events)

return clean, events

def event_names(props: Dict[str, Any]) -> FrozenSet[str]:

"""Return the event-name set a handler should consult for ``props``."""

names = props.get(EVENTS_PROP)

if isinstance(names, frozenset):

return names

if isinstance(names, (set, list, tuple)):

return frozenset(names)

return frozenset()

__all__ = [

"EVENTS_PROP",

"GESTURES_PROP",

"EventRegistry",

"get_event_registry",

"dispatch_event",

"extract_events",

"event_names",

]

pythonnative/src/pythonnative/events.py at v0.43.1 · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

events.py

Latest commit

History

Breadcrumbs

events.py

File metadata and controls