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

Repository navigation

History

1684 lines (1329 loc) · 58 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

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

"""Hook primitives for function components.

Provides React-like hooks for managing state, effects, memoization, and

context within components decorated with

[`component`][pythonnative.component.component]. Hooks must be called at the top

level of a component (not inside conditionals or loops) so they map to

the same slot across renders. In dev mode the framework verifies this

and raises [`HookOrderError`][pythonnative.diagnostics.HookOrderError]

on a violation instead of silently cross-wiring state.

Two effect phases exist, mirroring React:

- [`use_layout_effect`][pythonnative.use_layout_effect] callbacks run

synchronously inside the commit, after native mutations and the

layout pass have been applied. They can measure committed frames and

issue imperative view commands before the user sees the new frame.

- [`use_effect`][pythonnative.use_effect] callbacks (passive effects)

run after the layout effects, at the end of the same commit. An

effect may be an ``async def``; it runs as a task on the framework

loop and is cancelled when its dependencies change or the component

unmounts.

The current hook state travels in a :mod:`contextvars` context rather

than a plain global, so ``async def`` component bodies keep their hook

identity across ``await`` boundaries even when several coroutine

renders interleave on the event loop.

Hooks talk to the reconciler through the small

[`RenderOwner`][pythonnative.hooks.RenderOwner] protocol (mark a

component dirty, request a render, defer a transition, register a back

handler). That is the whole contract between the two modules.

Example:

```python

import pythonnative as pn

@pn.component

def Counter(initial: int = 0):

count, set_count = pn.use_state(initial)

return pn.Column(

pn.Text(f"Count: {count}"),

pn.Button("+", on_press=lambda: set_count(count + 1)),

)

```

"""

from __future__ import annotations

import asyncio

import inspect

from contextvars import ContextVar, Token

from dataclasses import dataclass, field, replace

from typing import (

Any,

Awaitable,

Callable,

Dict,

Generic,

List,

Optional,

Protocol,

Tuple,

TypeVar,

Union,

overload,

)

from . import diagnostics

from .element import Element, Node

from .platform_metrics import SafeAreaInsets, WindowDimensions

from .runtime import TaskScope, _scope, call_on_application_thread

from .scheduler import TransitionQueue, in_transition, run_in_transition, schedule_trigger

from .suspense import Resource

T = TypeVar("T")

StateSetter = Callable[[Union[T, Callable[[T], T]]], None]

"""Setter returned by [`use_state`][pythonnative.use_state]: accepts a value or ``current -> new``."""

_SENTINEL = object()

# Immutable environments are inherited by asyncio tasks. Never mutate the

# mapping or its tuples after publication to an execution context.

_providers: ContextVar[Dict[int, Tuple[Any, ...]]] = ContextVar("pn_providers", default={})

def provider_environment() -> Dict[int, Any]:

"""Return the current immutable provider environment for render identity."""

return {key: stack[-1] for key, stack in _providers.get().items()}

# The component whose body is currently executing. A ContextVar (not a

# global or thread-local) so coroutine component bodies resume with the

# right hook state after every ``await``, no matter how renders

# interleave on the loop.

_hook_context: ContextVar[Optional["HookState"]] = ContextVar("pn_hook_state", default=None)

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

# Reconciler contract

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

class RenderOwner(Protocol):

"""What a hook needs from the object that renders its component.

The reconciler implements this; tests may substitute a stub.

"""

transitions: TransitionQueue

def mark_dirty(self, vnode: Any) -> None:

"""Queue ``vnode``'s component for a local re-render."""

def request_render(self) -> None:

"""Ask the host to flush dirty components (may be deferred)."""

def register_back_handler(self, handler: Callable[[], bool]) -> Callable[[], None]:

"""Register a system back-press handler; returns an unregister callable."""

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

# Ref

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

class Ref(Generic[T]):

"""Mutable container returned by [`use_ref`][pythonnative.use_ref].

A ``Ref`` holds one value on its ``current`` attribute. Mutating

``current`` never triggers a re-render, which makes refs the right

place for timers, last-seen values, and imperative handles.

When a ``Ref`` is passed to a built-in element via the ``ref=``

prop, the reconciler populates ``current`` with the underlying

native view (``UIView`` on iOS, ``android.view.View`` on Android,

a DOM element in the browser preview) after commit, and clears it back to

``None`` on unmount. Composite components (e.g.

[`FlatList`][pythonnative.FlatList]) instead publish a typed

controller object on ``current`` via

[`use_imperative_handle`][pythonnative.use_imperative_handle].

Attributes:

current: The referenced value. ``None`` until populated.

"""

__slots__ = ("current", "_pn_tag", "_pn_frame")

def __init__(self, initial: Optional[T] = None) -> None:

self.current: Optional[T] = initial

# Internal: the native view tag, populated by the reconciler

# when the ref is attached to a built-in element.

self._pn_tag: Optional[int] = None

# Internal: the last committed frame ``(x, y, w, h)``, mirrored

# by the layout pass so Python code can read measured geometry

# without a native round-trip.

self._pn_frame: Optional[Tuple[float, float, float, float]] = None

def __setattr__(self, name: str, value: Any) -> None:

from .reconciler.journal import record_attribute

record_attribute(self, name)

object.__setattr__(self, name, value)

def __repr__(self) -> str:

return f"Ref({self.current!r})"

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

# Hook state container

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

class HookState:

"""Per-instance storage for one component's hooks.

Each component instance owns one ``HookState``. Hooks are matched

to slots by call order, so they must always be called in the same

order across renders. Effects scheduled during render are deferred

(layout effects into ``_pending_layout_effects``, passive effects

into ``_pending_effects``) and flushed by the reconciler in two

phases after native mutations commit.

Attributes:

states: One entry per ``use_state`` / ``use_reducer`` call.

effects: One ``(deps, cleanup)`` tuple per ``use_effect`` call.

layout_effects: One ``(deps, cleanup)`` tuple per

``use_layout_effect`` call.

memos: One ``(deps, value)`` tuple per ``use_memo`` / ``use_callback``.

refs: One [`Ref`][pythonnative.Ref] per ``use_ref`` call.

owner: The [`RenderOwner`][pythonnative.hooks.RenderOwner]

(reconciler) this component is mounted in, or ``None``.

vnode: The reconciler's node for this component, or ``None``.

"""

__slots__ = (

"states",

"_setters",

"_state_queues",

"_reducers",

"_dispatchers",

"effects",

"layout_effects",

"memos",

"refs",

"resources",

"state_index",

"effect_index",

"layout_effect_index",

"memo_index",

"ref_index",

"resource_index",

"context_deps",

"owner",

"vnode",

"_pending_effects",

"_pending_layout_effects",

"_pending_effects_mark",

"_pending_layout_effects_mark",

"_dirty",

"_hook_log",

"_hook_signature",

"_component_name",

"_async_task",

"_async_inputs",

"task_scope",

)

def __init__(self) -> None:

self.states: List[Any] = []

self._setters: Dict[int, Callable[[Any], None]] = {}

self._state_queues: Dict[int, Tuple[Any, List[Any]]] = {}

self._reducers: Dict[int, Callable] = {}

self._dispatchers: Dict[int, Callable] = {}

self.effects: List[Tuple[Any, Any]] = []

self.layout_effects: List[Tuple[Any, Any]] = []

self.memos: List[Tuple[Any, Any]] = []

self.refs: List[Ref] = []

# One ``(deps, Resource)`` per ``use_resource`` call.

self.resources: List[Tuple[Any, Resource]] = []

self.state_index: int = 0

self.effect_index: int = 0

self.layout_effect_index: int = 0

self.memo_index: int = 0

self.ref_index: int = 0

self.resource_index: int = 0

# Contexts read during the last completed render, keyed by

# ``id(context)``. The reconciler consults this when a

# Provider's value changes so consumers re-render even when a

# memoized ancestor skipped (reactive context).

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

self.owner: Optional[RenderOwner] = None

self.vnode: Any = None

self._pending_effects: List[Tuple[int, Callable, Any]] = []

self._pending_layout_effects: List[Tuple[int, Callable, Any]] = []

# Cleared by the reconciler after each successful render.

# ``use_state`` / ``use_reducer`` setters flip it to ``True``

# whenever they actually mutate state, so a memoized component

# still re-renders even when its props didn't change.

self._dirty: bool = False

# Dev-mode hook-order guard: the sequence of hook kinds called

# during the in-flight render, and the signature captured from

# the first successful render.

self._hook_log: Optional[List[str]] = None

self._hook_signature: Optional[List[str]] = None

self._component_name: str = ""

# For ``async def`` components: the asyncio task running the

# in-flight body, cancelled when a newer render supersedes it.

self._async_task: Any = None

self._async_inputs: Any = None

self.task_scope = TaskScope("component")

# Effect-queue lengths at ``begin_render``, so a suspended

# render can be rolled back without double-queueing effects.

self._pending_effects_mark: int = 0

self._pending_layout_effects_mark: int = 0

# ------------------------------------------------------------------

# Render lifecycle

# ------------------------------------------------------------------

def begin_render(self, component_name: str = "") -> None:

"""Prepare for a render pass: reset cursors and the dev-mode hook log."""

self.state_index = 0

self.effect_index = 0

self.layout_effect_index = 0

self.memo_index = 0

self.ref_index = 0

self.resource_index = 0

self.context_deps = {}

if component_name:

self._component_name = component_name

self._hook_log = [] if diagnostics.is_dev() else None

self._pending_effects_mark = len(self._pending_effects)

self._pending_layout_effects_mark = len(self._pending_layout_effects)

def abort_render(self) -> None:

"""Roll back a suspended render's effect queue.

A suspended body re-runs from the top on retry, so any effects

it queued before suspending would otherwise be queued twice.

"""

del self._pending_effects[self._pending_effects_mark :]

del self._pending_layout_effects[self._pending_layout_effects_mark :]

self._hook_log = None

def finish_render(self) -> None:

"""Finalize a successful render: lock in / verify the hook signature.

Raises:

HookOrderError: In dev mode, when this render called fewer

hooks than the previous one.

"""

log = self._hook_log

self._hook_log = None

if log is None:

return

if self._hook_signature is None:

self._hook_signature = log

return

if len(log) < len(self._hook_signature):

missing = self._hook_signature[len(log)]

raise diagnostics.HookOrderError(

f"{self._component_name or 'Component'} rendered fewer hooks than the previous "

f"render (expected {missing!r} at position {len(log) + 1}). Hooks must be called "

"unconditionally, in the same order, on every render."

)

def record_hook(self, kind: str) -> None:

"""Record a hook call for the dev-mode order guard.

Raises:

HookOrderError: In dev mode, when the hook at this position

differs from (or extends past) the previous render.

"""

log = self._hook_log

if log is None:

return

position = len(log)

log.append(kind)

signature = self._hook_signature

if signature is None:

return

if position >= len(signature):

raise diagnostics.HookOrderError(

f"{self._component_name or 'Component'} rendered more hooks than the previous "

f"render ({kind!r} at position {position + 1}). Hooks must be called "

"unconditionally, in the same order, on every render."

)

if signature[position] != kind:

raise diagnostics.HookOrderError(

f"{self._component_name or 'Component'} called {kind!r} at position "

f"{position + 1}, but the previous render called {signature[position]!r} there. "

"Hooks must be called unconditionally, in the same order, on every render."

)

def reset_hook_signature(self) -> None:

"""Forget the recorded hook signature (used by Fast Refresh)."""

self._hook_signature = None

# ------------------------------------------------------------------

# Effects

# ------------------------------------------------------------------

def flush_layout_effects(self) -> None:

"""Run layout effects queued during render (commit phase, pre-paint)."""

pending = self._pending_layout_effects

self._pending_layout_effects = []

self._pending_layout_effects_mark = 0

for idx, effect_fn, deps in pending:

_, prev_cleanup = self.layout_effects[idx]

_run_cleanup(prev_cleanup)

token = _scope.set(self.task_scope)

try:

cleanup = _activate_effect(effect_fn)

finally:

_scope.reset(token)

self.layout_effects[idx] = (list(deps) if deps is not None else None, cleanup)

def flush_pending_effects(self) -> None:

"""Run passive effects queued during render, after native commit.

For each pending effect, the previous cleanup is invoked first

(if any), then the new effect callback. The new return value

becomes the next cleanup. Effects that are ``async def`` (or

that return an awaitable) run as tasks on the framework loop;

their cleanup cancels the task, and a callable returned by the

coroutine runs as an additional cleanup once it completed.

"""

pending = self._pending_effects

self._pending_effects = []

self._pending_effects_mark = 0

for idx, effect_fn, deps in pending:

_, prev_cleanup = self.effects[idx]

_run_cleanup(prev_cleanup)

token = _scope.set(self.task_scope)

try:

cleanup = _activate_effect(effect_fn)

finally:

_scope.reset(token)

self.effects[idx] = (list(deps) if deps is not None else None, cleanup)

def cleanup_all_effects(self) -> None:

"""Run every outstanding cleanup function, then clear state.

Layout-effect cleanups run before passive-effect cleanups,

matching the mount order in reverse. Also cancels in-flight

resources and any pending ``async def`` body. Called when the

component instance is unmounted by the reconciler.

"""

for i, (_deps, cleanup) in enumerate(self.layout_effects):

_run_cleanup(cleanup)

self.layout_effects[i] = (_SENTINEL, None)

for i, (_deps, cleanup) in enumerate(self.effects):

_run_cleanup(cleanup)

self.effects[i] = (_SENTINEL, None)

self._pending_effects = []

self._pending_layout_effects = []

self._pending_effects_mark = 0

self._pending_layout_effects_mark = 0

for _deps, resource in self.resources:

try:

resource.cancel()

except Exception:

pass

self.resources = []

self.task_scope.close()

driver = self._async_task

self._async_task = None

if driver is not None:

driver.cancel()

def detach(self) -> None:

"""Break the back-references to the reconciler (on unmount).

Lets the unmounted component's hook state (and the closures it

captured) be freed by plain refcounting, which matters on iOS

where the cyclic GC is disabled.

"""

self.owner = None

self.vnode = None

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

# Context helpers (framework-internal)

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

def current_hook_state() -> Optional[HookState]:

"""Return the active ``HookState``, or ``None`` if no render is in flight."""

return _hook_context.get()

def install_hook_state(state: Optional[HookState]) -> Token[Optional[HookState]]:

"""Install ``state`` as the active ``HookState``; returns the reset token."""

return _hook_context.set(state)

def restore_hook_state(token: Token[Optional[HookState]]) -> None:

"""Restore the hook state that was active before ``install_hook_state``."""

_hook_context.reset(token)

def _require_hook_state(hook_name: str) -> HookState:

ctx = _hook_context.get()

if ctx is None:

raise RuntimeError(f"{hook_name} must be called inside a @component function")

return ctx

def _deps_changed(prev: Any, current: Any) -> bool:

"""Return whether the dependency arrays differ enough to re-run an effect."""

if prev is _SENTINEL:

return True

if prev is None or current is None:

return True

if len(prev) != len(current):

return True

from .equality import equal

return any(not equal(p, c) for p, c in zip(prev, current))

def _run_cleanup(cleanup: Any) -> None:

if callable(cleanup):

try:

cleanup()

except Exception as exc:

# Never let a failing cleanup abort an unmount; surface it

# through the RedBox in dev mode, a warning otherwise.

if not diagnostics.report_error(exc, phase="effect cleanup"):

diagnostics.warn(f"Effect cleanup raised {exc!r}")

def _activate_effect(effect_fn: Callable) -> Any:

"""Invoke an effect callback, running coroutine effects as tasks.

Synchronous effects return their cleanup directly. When the effect

is an ``async def`` (or returns an awaitable), the coroutine runs

as a task on the framework loop and the returned cleanup cancels

it; if the coroutine already finished and returned a callable, that

callable runs as the cleanup instead.

"""

result = effect_fn()

if not inspect.isawaitable(result):

return result

from .runtime import run_async

future = run_async(result)

def _observe(fut: Any) -> None:

# Surface unhandled async-effect crashes instead of letting the

# future's exception vanish unobserved: RedBox in dev mode,

# traceback in production.

if fut.cancelled():

return

exc = fut.exception()

if exc is None or isinstance(exc, asyncio.CancelledError):

return

if not diagnostics.report_error(exc, phase="async effect"):

import traceback

traceback.print_exception(type(exc), exc, exc.__traceback__)

future.add_done_callback(_observe)

def _cleanup() -> None:

if future.cancelled():

return

if future.done():

if future.exception() is None:

returned = future.result()

if callable(returned):

try:

returned()

except Exception:

pass

return

future.cancel()

return _cleanup

def _notify_state_changed(ctx: HookState) -> None:

"""Mark ``ctx``'s component dirty and schedule a render after a state change.

Enqueuing the owning node in the reconciler's dirty set is what

makes the subsequent render *local*: the host's trigger flushes

only the components marked here rather than the whole app. The

dirty mark is eager (so several setters coalesce), while the render

request respects [`batch_updates`][pythonnative.scheduler.batch_updates] and

defers to a later loop turn inside a transition (see

[`use_transition`][pythonnative.use_transition]).

"""

ctx._dirty = True

owner = ctx.owner

if owner is None:

return

if ctx.vnode is not None:

owner.mark_dirty(ctx.vnode)

if in_transition():

owner.transitions.defer(owner.request_render)

else:

schedule_trigger(owner.request_render)

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

# State hooks

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

@overload

def use_state() -> Tuple[Optional[Any], StateSetter[Any]]: ...

@overload

def use_state(initial: Callable[[], T]) -> Tuple[T, StateSetter[T]]: ...

@overload

def use_state(initial: T) -> Tuple[T, StateSetter[T]]: ...

def use_state(initial: Any = None) -> Tuple[Any, StateSetter[Any]]:

"""Return ``(value, setter)`` for component-local state.

State persists across re-renders of the same component instance.

The setter accepts a value or a ``current -> new`` callable; calling

it with an unchanged value is a no-op (no re-render).

Args:

initial: Initial state value. If callable, it is invoked once on

the first render (lazy initialization).

Returns:

A 2-tuple ``(value, setter)`` where ``value`` is the current

state and ``setter`` updates it (and triggers a re-render).

Raises:

RuntimeError: If called outside a ``@component`` function.

Example:

```python

import pythonnative as pn

@pn.component

def Counter():

count, set_count = pn.use_state(0)

return pn.Button(

f"Count: {count}",

on_press=lambda: set_count(count + 1),

)

```

"""

ctx = _require_hook_state("use_state")

ctx.record_hook("use_state")

idx = ctx.state_index

ctx.state_index += 1

if idx >= len(ctx.states):

val = initial() if callable(initial) else initial

ctx.states.append(val)

return ctx.states[idx], _state_setter(ctx, idx)

def _state_setter(ctx: HookState, idx: int) -> Callable[[Any], None]:

if idx in ctx._setters:

return ctx._setters[idx]

def setter(update: Any) -> None:

deferred = in_transition() and ctx.owner is not None

def apply() -> None:

if ctx.task_scope.closed:

return

from .equality import equal

if deferred:

if idx not in ctx._state_queues:

ctx._state_queues[idx] = (ctx.states[idx], [])

assert ctx.owner is not None

ctx.owner.transitions.defer(flush)

ctx._state_queues[idx][1].append(update)

return

# An urgent update after skipped work is replayed when that work

# commits, so functional and replacement updates retain their order.

if idx in ctx._state_queues:

ctx._state_queues[idx][1].append(update)

value = update(ctx.states[idx]) if callable(update) else update

if not equal(ctx.states[idx], value):

ctx.states[idx] = value

_notify_state_changed(ctx)

call_on_application_thread(apply)

def flush() -> None:

queued = ctx._state_queues.pop(idx, None)

if queued is None or ctx.task_scope.closed:

return

value, updates = queued

for update in updates:

value = update(value) if callable(update) else update

from .equality import equal

if not equal(ctx.states[idx], value):

ctx.states[idx] = value

_notify_state_changed(ctx)

ctx._setters[idx] = setter

return setter

def use_reducer(

reducer: Callable[[T, Any], T], initial_state: Union[T, Callable[[], T]]

) -> Tuple[T, Callable[[Any], None]]:

"""Return ``(state, dispatch)`` for reducer-based state management.

A reducer is a pure function that takes the current state and an

action and returns the next state. Use it instead of

[`use_state`][pythonnative.use_state] when state transitions are

complex enough that centralizing them in one function aids

readability and testing.

Args:

reducer: ``reducer(current_state, action) -> new_state``.

The component re-renders only when ``reducer`` returns a

value different from the current state.

initial_state: Initial state value, or a callable invoked once

on the first render.

Returns:

A 2-tuple ``(state, dispatch)`` where ``dispatch`` runs the

reducer with the supplied action.

Raises:

RuntimeError: If called outside a ``@component`` function.

"""

ctx = _require_hook_state("use_reducer")

ctx.record_hook("use_reducer")

idx = ctx.state_index

ctx.state_index += 1

if idx >= len(ctx.states):

val = initial_state() if callable(initial_state) else initial_state

ctx.states.append(val)

ctx._reducers[idx] = reducer

if idx not in ctx._dispatchers:

def dispatch(action: Any) -> None:

reduce = ctx._reducers[idx]

_state_setter(ctx, idx)(lambda value: reduce(value, action))

ctx._dispatchers[idx] = dispatch

return ctx.states[idx], ctx._dispatchers[idx]

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

# Effect hooks

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

def use_effect(effect: Callable[[], Any], deps: Optional[list] = None) -> None:

"""Schedule a side effect to run after the native commit.

Effects are queued during the render pass and flushed once the

reconciler has finished applying all native-view mutations, which

means effect callbacks can safely measure layout or interact with

committed native views.

The ``deps`` argument controls when the effect re-runs:

- ``None``: every render.

- ``[]``: mount only.

- ``[a, b]``: when ``a`` or ``b`` change (compared by identity, then ``==``).

A synchronous ``effect`` may return a cleanup callable; the previous

cleanup runs before the next effect (and on unmount).

An **async** ``effect`` (an ``async def``) runs as a task on the

framework loop. When ``deps`` change or the component unmounts, the

in-flight task is cancelled (:class:`asyncio.CancelledError` is

raised at its current ``await``), giving async effects structured

cancellation for free. If the coroutine finishes and returns a

callable, that callable runs as the cleanup instead.

Args:

effect: A zero-arg callable invoked after commit: either a

synchronous function (optionally returning a cleanup

callable) or an ``async def``.

deps: Dependency list, or ``None`` to run on every render.

Raises:

RuntimeError: If called outside a ``@component`` function.

Example:

```python

import asyncio

import time

import pythonnative as pn

@pn.component

def Clock():

now, set_now = pn.use_state("")

async def tick():

while True:

set_now(time.strftime("%H:%M:%S"))

await asyncio.sleep(1)

pn.use_effect(tick, [])

return pn.Text(now)

```

"""

ctx = _require_hook_state("use_effect")

ctx.record_hook("use_effect")

idx = ctx.effect_index

ctx.effect_index += 1

if idx >= len(ctx.effects):

ctx.effects.append((_SENTINEL, None))

ctx._pending_effects.append((idx, effect, deps))

return

prev_deps, _prev_cleanup = ctx.effects[idx]

if _deps_changed(prev_deps, deps):

ctx._pending_effects.append((idx, effect, deps))

def use_layout_effect(effect: Callable[[], Any], deps: Optional[list] = None) -> None:

"""Schedule a side effect that runs synchronously inside the commit.

Like [`use_effect`][pythonnative.use_effect], but the callback

fires *before* passive effects, immediately after native mutations

and the layout pass are applied. Use it when you need to measure a

committed frame (via a [`Ref`][pythonnative.Ref]) or issue an

imperative view command before the user sees the new frame, for

example scrolling a list into position on mount.

Prefer ``use_effect`` for everything else; layout effects block the

commit, so heavy work here delays the frame.

Args:

effect: A zero-arg callable invoked during commit. Optionally

returns a cleanup callable.

deps: Dependency list, or ``None`` to run on every render.

Raises:

RuntimeError: If called outside a ``@component`` function.

"""

ctx = _require_hook_state("use_layout_effect")

ctx.record_hook("use_layout_effect")

idx = ctx.layout_effect_index

ctx.layout_effect_index += 1

if idx >= len(ctx.layout_effects):

ctx.layout_effects.append((_SENTINEL, None))

ctx._pending_layout_effects.append((idx, effect, deps))

return

prev_deps, _prev_cleanup = ctx.layout_effects[idx]

if _deps_changed(prev_deps, deps):

ctx._pending_layout_effects.append((idx, effect, deps))

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

# Memoization hooks

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

def use_memo(factory: Callable[[], T], deps: list) -> T:

"""Return a memoized value that is recomputed only when ``deps`` change.

Use this for expensive computations whose inputs change rarely. For

cheap computations, plain inline code is faster (memoization itself

has overhead).

Args:

factory: Zero-arg callable returning the value.

deps: Dependency list. The value is recomputed when any element

differs from the previous render.

Returns:

The cached or freshly computed value.

Raises:

RuntimeError: If called outside a ``@component`` function.

"""

ctx = _require_hook_state("use_memo")

ctx.record_hook("use_memo")

idx = ctx.memo_index

ctx.memo_index += 1

if idx >= len(ctx.memos):

value = factory()

ctx.memos.append((list(deps), value))

return value

prev_deps, prev_value = ctx.memos[idx]

if not _deps_changed(prev_deps, deps):

return prev_value

value = factory()

ctx.memos[idx] = (list(deps), value)

return value

F = TypeVar("F", bound=Callable[..., Any])

def use_callback(callback: F, deps: list) -> F:

"""Return a stable reference to ``callback``, refreshed when ``deps`` change.

Equivalent to ``use_memo(lambda: callback, deps)``. Useful when

passing a function as a prop to a memoized child component, so the

child doesn't see a fresh function identity on every render.

Args:

callback: The callable to memoize.

deps: Dependency list controlling when the reference refreshes.

Returns:

A callable with stable identity across renders (until ``deps`` change).

"""

return use_memo(lambda: callback, deps)

def use_ref(initial: Optional[T] = None) -> Ref[T]:

"""Return a [`Ref`][pythonnative.Ref] that persists across renders.

Refs are useful for storing values that must survive renders without

triggering them: timers, last-seen values, native handles, and so on.

``ref.current`` is also populated by the reconciler with the

underlying native view when the ref is passed via the ``ref=`` prop

on a built-in element, and cleared to ``None`` when that element

unmounts. Composite components such as

[`FlatList`][pythonnative.FlatList] publish a typed controller

object instead (see

[`use_imperative_handle`][pythonnative.use_imperative_handle]).

Args:

initial: Value placed at ``ref.current`` on first render.

Returns:

A [`Ref`][pythonnative.Ref]. Mutations to ``ref.current`` do

*not* trigger re-renders.

Raises:

RuntimeError: If called outside a ``@component`` function.

"""

ctx = _require_hook_state("use_ref")

ctx.record_hook("use_ref")

idx = ctx.ref_index

ctx.ref_index += 1

if idx >= len(ctx.refs):

ref: Ref[T] = Ref(initial)

ctx.refs.append(ref)

return ref

return ctx.refs[idx]

def use_imperative_handle(

ref: Optional[Ref[Any]],

factory: Callable[[], Any],

deps: Optional[list] = None,

) -> None:

"""Publish a controller object on ``ref.current``.

The composite-component counterpart to passing ``ref=`` to a

built-in element. Call it inside a component that accepts a

``ref`` prop to expose a curated imperative API (rather than the

raw native view) to the parent. The handle is installed during the

commit's layout-effect phase and cleared back to ``None`` on

unmount.

Args:

ref: The [`Ref`][pythonnative.Ref] received via the component's

``ref`` prop. ``None`` is allowed (the parent didn't

request a handle), in which case this is a no-op.

factory: Zero-arg callable returning the handle object.

deps: Dependency list controlling when the handle is rebuilt.

``None`` rebuilds on every render, matching effects.

Raises:

RuntimeError: If called outside a ``@component`` function.

Example:

```python

@pn.component

def VideoPlayer(source: str, ref: pn.Ref | None = None):

pn.use_imperative_handle(ref, lambda: PlayerController(...), [source])

return pn.View(...)

```

"""

def _install() -> Optional[Callable[[], None]]:

if ref is None:

return None

ref.current = factory()

def _clear() -> None:

ref.current = None

return _clear

use_layout_effect(_install, deps)

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

# Async hooks

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

def use_resource(fetcher: Callable[[], Any], deps: Optional[list] = None) -> Resource[Any]:

"""Start an async fetch and cache it across renders.

The fetch starts immediately (during render, not after commit) and

the resulting [`Resource`][pythonnative.Resource] is cached until

``deps`` change, at which point the old fetch is cancelled and a

new one starts. Because results are cached, re-renders resolve

instantly; only genuinely new data suspends.

Consume the resource with ``resource.read()`` (suspends the render

while pending; pair with a [`Suspense`][pythonnative.Suspense]

boundary) or ``await resource`` inside an ``async def`` component.

Errors raised by the fetcher re-raise at the read site, so an

enclosing [`ErrorBoundary`][pythonnative.ErrorBoundary] catches

failures declaratively.

View remainder of file in raw view

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

Repository navigation

Expand file tree

Breadcrumbs

hooks.py

Latest commit

History

Breadcrumbs

hooks.py

File metadata and controls