pythonnative/src/pythonnative/scheduler.py at refs/heads/main · pythonnative/pythonnative · GitHub

Repository navigation

History

178 lines (145 loc) · 6.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

"""Update scheduling primitives shared by hooks and the reconciler.

Two pieces live here:

- [`batch_updates`][pythonnative.scheduler.batch_updates], a context manager that

coalesces the render triggers fired by several state setters into

one, tracked per execution context (so it composes with ``async``

code).

- [`TransitionQueue`][pythonnative.scheduler.TransitionQueue], the

per-reconciler queue behind

[`use_transition`][pythonnative.use_transition]: renders marked as

transitions are deferred to a later turn of the framework loop so

urgent updates (typing, presses) stay responsive.

Nothing here is global mutable state beyond context variables, so

several reconcilers (screens, list rows, tests) never interfere.

"""

from __future__ import annotations

from contextlib import contextmanager

from contextvars import ContextVar

from typing import Callable, Generator, List, Optional

from . import diagnostics

__all__ = ["TransitionQueue", "batch_updates", "in_transition", "run_in_transition", "schedule_trigger"]

# Depth of nested ``batch_updates`` blocks plus the triggers deferred by

# the outermost block, per execution context.

_batch_depth: ContextVar[int] = ContextVar("pn_batch_depth", default=0)

_batch_pending: ContextVar[Optional[List[Callable[[], None]]]] = ContextVar("pn_batch_pending", default=None)

# Whether state updates in the current execution context are marked as

# transitions (see ``use_transition``).

_transition_var: ContextVar[bool] = ContextVar("pn_transition", default=False)

def schedule_trigger(trigger: Callable[[], None]) -> None:

"""Run ``trigger`` now, or defer it to the end of the enclosing ``batch_updates`` block."""

pending = _batch_pending.get()

if _batch_depth.get() > 0 and pending is not None:

if trigger not in pending:

pending.append(trigger)

else:

trigger()

@contextmanager

def batch_updates() -> Generator[None, None, None]:

"""Coalesce state updates that span ``await`` boundaries into a single re-render.

Setters already batch automatically: every call made within one

callback, effect, or task step lands in the same render pass, which

the reconciler schedules on the next turn of the application loop.

Use this block when a sequence of updates is interrupted by an

``await`` (a fetch between two setters, say) and should still

commit as one pass: triggers fired inside the ``with`` block are

deferred until it exits.

Example:

```python

import pythonnative as pn

async def load():

with pn.batch_updates():

set_loading(False)

set_items(await api.list_items())

```

"""

depth = _batch_depth.get()

depth_token = _batch_depth.set(depth + 1)

pending_token = None

if depth == 0:

pending_token = _batch_pending.set([])

try:

yield

finally:

_batch_depth.reset(depth_token)

if pending_token is not None:

triggers = _batch_pending.get() or []

_batch_pending.reset(pending_token)

for trigger in triggers:

trigger()

def in_transition() -> bool:

"""Whether state updates in the current context are transitions."""

return _transition_var.get()

def run_in_transition(fn: Callable[[], None]) -> None:

"""Run ``fn`` with its state updates marked as transitions."""

token = _transition_var.set(True)

try:

fn()

finally:

_transition_var.reset(token)

class TransitionQueue:

"""Deferred render triggers and completion callbacks for one reconciler.

Triggers added via [`defer`][pythonnative.scheduler.TransitionQueue.defer]

run together on a later loop turn; callbacks added via

[`on_complete`][pythonnative.scheduler.TransitionQueue.on_complete]

run right after them (this is how ``use_transition`` flips its

``is_pending`` flag back off once the deferred render committed).

"""

__slots__ = ("_triggers", "_callbacks", "_scheduled")

def __init__(self) -> None:

self._triggers: List[Callable[[], None]] = []

self._callbacks: List[Callable[[], None]] = []

self._scheduled = False

def defer(self, trigger: Callable[[], None]) -> None:

"""Queue ``trigger`` for the next flush."""

if trigger not in self._triggers:

self._triggers.append(trigger)

self._ensure_scheduled()

def on_complete(self, callback: Callable[[], None]) -> None:

"""Queue ``callback`` to run after the next flush's triggers."""

self._callbacks.append(callback)

self._ensure_scheduled()

@property

def pending(self) -> bool:

"""Whether a flush is scheduled or work is queued."""

return self._scheduled or bool(self._triggers) or bool(self._callbacks)

def flush(self) -> None:

"""Run every deferred trigger, then the completion callbacks."""

self._scheduled = False

triggers = self._triggers

callbacks = self._callbacks

self._triggers = []

self._callbacks = []

token = _transition_var.set(False)

errors: list[Exception] = []

try:

# One failing trigger must not drop the remaining deferred updates.

with batch_updates():

for fn in triggers:

try:

fn()

except Exception as exc:

errors.append(exc)

for callback in callbacks:

try:

callback()

except Exception as exc:

errors.append(exc)

finally:

_transition_var.reset(token)

for error in errors:

if not diagnostics.report_error(error, phase="transition"):

raise error

def clear(self) -> None:

"""Drop queued work (used on unmount)."""

self._triggers = []

self._callbacks = []

self._scheduled = False

def _ensure_scheduled(self) -> None:

if self._scheduled:

return

self._scheduled = True

from .runtime import get_loop

get_loop().call_soon(self._flush_if_scheduled)

def _flush_if_scheduled(self) -> None:

if self._scheduled:

self.flush()

pythonnative/src/pythonnative/scheduler.py at refs/heads/main · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

scheduler.py

Latest commit

History

Breadcrumbs

scheduler.py

File metadata and controls