Pulse

Hooks

Render-time hooks and runtime helpers

state

Get or create a State instance persisted across renders.

def state(arg: S | Callable[[], S], *, key: str | None = None) -> S

Parameters

ParameterTypeDefaultDescription
argS | Callable[[], S]requiredState instance or factory function
keystr | NoneNoneOptional key for loops or manual invalidation

Returns

The same State instance on subsequent renders at the same location (or with the same key).

Example

# Simple usage - Pulse identifies by code location
def counter():
    s = ps.state(CounterState())
    return m.Button(f"Count: {s.count}", on_click=lambda: s.increment())

# With factory function
def counter():
    s = ps.state(lambda: CounterState())
    return m.Button(f"Count: {s.count}", on_click=lambda: s.increment())

# In a loop - use key to disambiguate
def user_list(user_ids):
    items = []
    for uid in user_ids:
        user = ps.state(lambda uid=uid: UserState(uid), key=uid)
        items.append(m.Text(user.name, key=uid))
    return m.Stack(items)

# Dynamic key - state recreated when key changes
def user_profile(user_id):
    user = ps.state(lambda: UserState(user_id), key=f"user-{user_id}")
    return m.Text(user.name)

Notes

  • Without a key, each call is identified by its code location (file + line number)
  • With a key, the key is used for identification instead
  • Key must be non-empty string if provided
  • Can only be called once per render at the same location or with the same key
  • Factory is only called on first render; subsequent renders return cached instance
  • State is disposed when component unmounts or when key changes

init

Context manager for one-time initialization in components. Variables assigned inside the block persist across re-renders.

def init() -> InitContext

Usage

def my_component():
    with ps.init():
        counter = 0
        api = ApiClient()
        data = fetch_initial_data()

    # counter, api, data retain their values across renders
    return m.Text(f"Counter: {counter}")

Rules

  • Can only be used once per component
  • Must be at the top level of the component function (not inside conditionals, loops, or nested functions)
  • Cannot contain control flow (if, for, while, try, with, match)
  • Cannot use as binding (with ps.init() as ctx: not allowed)
  • Variables are restored from first render on subsequent renders

How It Works

ps.init() uses AST rewriting to transform your code at decoration time. When you use the @ps.component decorator, Pulse parses your function's source code and rewrites with ps.init(): blocks into conditional logic.

On first render, the code inside the block runs normally and variables are captured. On subsequent renders, the block is skipped entirely and variables are restored from storage.

Equivalent to ps.setup()

Under the hood, ps.init() is syntactic sugar. This code:

@ps.component
def Counter():
    with ps.init():
        state = CounterState()
        value = expensive_calculation()

    return ps.div(f"Count: {state.count}")

Is equivalent to:

@ps.component
def Counter():
    def init():
        return CounterState(), expensive_calculation()

    state, value = ps.setup(init)

    return ps.div(f"Count: {state.count}")

Troubleshooting

If you encounter issues with ps.init() (e.g., source code not available in some deployment environments, or unexpected behavior), use ps.setup() instead. It provides the same functionality without AST rewriting:

# If ps.init() causes issues, use ps.setup() instead
state = ps.setup(lambda: CounterState())

effect (inline)

When @ps.effect is used inside a component, effects are automatically registered and cached by code location.

@ps.effect
def my_effect():
    print("Effect runs when dependencies change")

@ps.effect(key="unique-key", immediate=True)
def keyed_effect():
    print("Effect with explicit key")

Parameters (decorator)

ParameterTypeDefaultDescription
keystr | NoneNoneOptional key for loops or manual invalidation
immediateboolFalseRun synchronously instead of batched
lazyboolFalseDon't run on creation
on_errorCallable[[Exception], None] | NoneNoneError handler
intervalfloat | NoneNonePolling interval in seconds

Example

@ps.component
def UserProfile():
    with ps.init():
        state = UserState()

    # Simple inline effect - auto-registered
    @ps.effect
    def log_changes():
        print(f"User changed: {state.name}")

    return ps.div(state.name)

# In a loop - use key parameter
@ps.component
def ItemList():
    with ps.init():
        items = ["a", "b", "c"]

    for item in items:
        @ps.effect(key=item)
        def track_item(item=item):  # Capture via default arg
            print(f"Tracking: {item}")

    return ps.ul(...)

# Dynamic key - effect recreated when key changes
@ps.component
def DynamicEffect():
    with ps.init():
        version = ps.Signal("v1")

    @ps.effect(key=version())
    def versioned_effect():
        print(f"Version: {version()}")
        def cleanup():
            print("Cleaning up old version")
        return cleanup

    return ps.div(...)

Notes

  • Effects are cached by code location when no key is provided
  • In loops, the key parameter is required to disambiguate each effect
  • When key changes, old effect is disposed and new one is created
  • Effects with cleanup functions have their cleanup called on key change or unmount

stable

Return a stable wrapper that always calls the latest value.

@overload
def stable(key: str) -> Any: ...

@overload
def stable(key: str, value: TCallable) -> TCallable: ...

@overload
def stable(key: str, value: T) -> Callable[[], T]: ...

Parameters

ParameterTypeDefaultDescription
keystrrequiredUnique identifier
valueAnyoptionalValue or callable to wrap

Returns

A stable wrapper function that delegates to the current value.

Example

def my_component():
    s = ps.state("data", lambda: DataState())

    # Without stable, this would create a new function each render
    handle_click = ps.stable("click", lambda: s.increment())

    return m.Button("Click", on_click=handle_click)

Use Cases

  • Event handlers passed to child components to prevent unnecessary re-renders
  • Callbacks registered with external systems
  • Any function reference that needs to stay stable across renders

route

Get the current route context.

def route() -> RouteContext

Returns

RouteContext with access to route parameters, path, and query.

Example

def user_page():
    r = ps.route()
    user_id = r.params.get("user_id")  # From /users/:user_id
    page = r.query.get("page", "1")    # From ?page=2
    return m.Text(f"User {user_id}, Page {page}")

Navigate to a new URL.

def navigate(path: str, *, replace: bool = False, hard: bool = False) -> None

Parameters

ParameterTypeDefaultDescription
pathstrrequiredDestination URL
replaceboolFalseReplace current history entry
hardboolFalseFull page reload instead of client-side navigation

Example

async def handle_login():
    await api.login(username, password)
    ps.navigate("/dashboard")

redirect

Redirect during render (throws exception to interrupt render).

def redirect(path: str, *, replace: bool = False) -> NoReturn

Parameters

ParameterTypeDefaultDescription
pathstrrequiredDestination URL
replaceboolFalseReplace current history entry

Example

def protected_page():
    user = get_current_user()
    if not user:
        ps.redirect("/login")  # Interrupts render

    return m.Text(f"Welcome, {user.name}")

not_found

Trigger 404 during render (throws exception to interrupt render).

def not_found() -> NoReturn

Example

def user_page():
    r = ps.route()
    user = get_user(r.params["id"])
    if not user:
        ps.not_found()  # Shows 404 page

    return m.Text(user.name)

session

Get the current user session data.

def session() -> ReactiveDict[str, Any]

Returns

ReactiveDict of session data that persists across page navigations.

Example

def my_component():
    sess = ps.session()
    sess["last_visited"] = datetime.now()
    return m.Text(f"Visits: {sess.get('visit_count', 0)}")

session_id

Get the current session identifier.

def session_id() -> str

websocket_id

Get the current WebSocket connection identifier.

def websocket_id() -> str

Set a cookie on the client.

async def set_cookie(
    name: str,
    value: str,
    domain: str | None = None,
    secure: bool = True,
    samesite: Literal["lax", "strict", "none"] = "lax",
    max_age_seconds: int = 7 * 24 * 3600,
) -> None

call_api

Make an API call through the client browser.

async def call_api(
    path: str,
    *,
    method: str = "POST",
    headers: Mapping[str, str] | None = None,
    body: Any | None = None,
    credentials: str = "include",
) -> dict[str, Any]

Useful for calling third-party APIs that require browser cookies/credentials.


server_address

Get the server's public address.

def server_address() -> str

client_address

Get the client's IP address.

def client_address() -> str

global_state

Create a globally shared state accessor.

def global_state(
    factory: Callable[P, S] | type[S],
    key: str | None = None,
) -> GlobalStateAccessor[P, S]

Parameters

ParameterTypeDefaultDescription
factoryCallable | typerequiredState class or factory function
keystr | NoneNoneOptional custom key

Returns

GlobalStateAccessor - callable that returns the shared state instance.

Example

@ps.global_state
class AppSettings(ps.State):
    theme: str = "light"
    language: str = "en"

def settings_panel():
    settings = AppSettings()  # Same instance across all components
    return m.Select(
        value=settings.theme,
        data=["light", "dark"],
        on_change=lambda v: setattr(settings, "theme", v),
    )

# With instance ID for per-entity global state
@ps.global_state
class UserCache(ps.State):
    data: dict = {}

def user_profile(user_id: str):
    cache = UserCache(id=user_id)  # Shared per user_id
    return m.Text(cache.data.get("name", "Loading..."))

hooks API

Low-level API for creating custom hooks.

hooks.create

Register a new hook.

def create(
    name: str,
    factory: HookFactory[T] = _default_factory,
    *,
    metadata: HookMetadata | None = None,
) -> Hook[T]

HookState

Base class for hook state. Override lifecycle methods:

class HookState(Disposable):
    def on_render_start(self, render_cycle: int) -> None: ...
    def on_render_end(self, render_cycle: int) -> None: ...
    def dispose(self) -> None: ...

HookMetadata

@dataclass
class HookMetadata:
    description: str | None = None
    owner: str | None = None
    version: str | None = None
    extra: Mapping[str, Any] | None = None

Example Custom Hook

class TimerHookState(ps.hooks.State):
    def __init__(self):
        self.start_time = time.time()

    def elapsed(self) -> float:
        return time.time() - self.start_time

    def dispose(self) -> None:
        pass

_timer_hook = ps.hooks.create("my_app:timer", lambda: TimerHookState())

def use_timer() -> TimerHookState:
    return _timer_hook()

On this page