Async and Data Fetching

Most apps need to load data from somewhere—an API, a database, or another async source. Pulse has built-in support for async patterns and provides queries and mutations to handle the boilerplate: loading states, error handling, caching, and automatic refetching.

Async fundamentals

Event handlers can be async functions. Pulse handles this automatically:

class SearchState(ps.State):
    query: str = ""
    results: list = []
    loading: bool = False

    async def search(self):
        self.loading = True
        self.results = await fetch_from_api(self.query)
        self.loading = False

Automatic batching

State updates are automatically batched between await calls. Your UI only rerenders when it makes sense:

async def process(self):
    # These two updates are batched - one rerender
    self.status = "loading"
    self.progress = 0

    await asyncio.sleep(1)

    # These are also batched - one rerender
    self.status = "done"
    self.progress = 100

Each group of synchronous updates triggers a single rerender, not one per update.

Queries

You could manage data fetching with plain async methods and manual loading flags, but that gets tedious. Queries handle this for you.

Think of it this way:

Queries are for reading data (fetching a user profile, loading a list of products)
Mutations are for writing data (saving a form, deleting a record)

Your first query

A query is an async method on a State class decorated with @ps.query:

import pulse as ps

class UserState(ps.State):
    user_id: int = 1

    @ps.query
    async def user(self):
        await asyncio.sleep(0.5)  # simulate network delay
        return {"id": self.user_id, "name": f"User {self.user_id}"}

In your component, access the query like a property. Pulse tracks loading and error states automatically:

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

    if state.user.is_loading:
        return ps.div("Loading...")

    return ps.div(
        ps.h1(f"Hello, {state.user.data['name']}!"),
        ps.button("Refresh", onClick=state.user.refetch),
    )

Notice you didn't have to:

Create a loading boolean and toggle it manually
Catch exceptions and store them somewhere
Figure out when to trigger the fetch

The query runs when the component first renders, and you get reactive properties to check its status.

Checking query status

Every query gives you these properties:

Property	What it tells you
`data`	The fetched result, or `None` if not yet loaded
`is_loading`	`True` during the first fetch (use for initial loading spinners)
`is_fetching`	`True` during any fetch, including refetches (use for refresh indicators)
`is_success`	`True` if the last fetch worked
`is_error`	`True` if the last fetch failed
`error`	The exception object if it failed, otherwise `None`

When should you use is_loading vs is_fetching? Use is_loading for a full loading state on the first load (like a skeleton screen). Use is_fetching to show any fetch activity, like a subtle spinner during background refetches.

Controlling queries manually

Method	When to use it
`refetch()`	User clicks a "refresh" button, or you need fresh data after some action
`invalidate()`	Mark data as stale so it refetches on next access (lazy refresh)
`set_data(...)`	Optimistic updates—show a change immediately before the server confirms
`wait()`	In an async callback when you need to await the query result

When do queries refetch?

By default, queries automatically track their dependencies. If your query reads self.user_id, Pulse notices that and refetches whenever user_id changes:

@ps.query
async def user(self):
    # Pulse sees you're reading self.user_id
    # When user_id changes, this query refetches automatically
    return await fetch_user(self.user_id)

This "just works" for most cases.

Keyed queries

Sometimes automatic tracking isn't what you want. Maybe you're reading multiple state properties but only want to refetch when one specific value changes:

@ps.query
async def user(self):
    return await fetch_user(self.user_id)

@user.key
def _user_key(self):
    # Only refetch when user_id changes, ignore other state
    return ("user", self.user_id)

Situation	Use
Simple case, refetch whenever any dependency changes	Unkeyed (default)
Need explicit control over refetch triggers	Keyed query
Want to share cached data across components	Keyed query (same key = shared cache)

Fine-tuning query behavior

The @ps.query decorator accepts options for common scenarios:

@ps.query(
    keep_previous_data=True,  # Show old data while refetching (avoids flicker)
    stale_time=60.0,          # Data is "fresh" for 60 seconds
    retries=3,                # Retry failed requests up to 3 times
    retry_delay=1.0,          # Wait 1 second between retries
    gc_time=300.0,            # Keep cached data around for 5 minutes
)
async def user(self):
    return await fetch_user(self.user_id)

If you want to...	Use this option
Avoid UI flicker when switching between items	`keep_previous_data=True`
Reduce API calls for data that doesn't change often	`stale_time=60.0` (or longer)
Handle flaky network connections	`retries=3, retry_delay=1.0`
Keep data in memory longer for fast back-navigation	`gc_time=300.0`

Reacting to query results

Use callback decorators when a query succeeds or fails:

@ps.query
async def user(self):
    return await fetch_user(self.user_id)

@user.on_success
def _on_success(self):
    print("User loaded successfully!")

@user.on_error
def _on_error(self):
    print("Failed to load user")

Providing initial data

Show placeholder data immediately while the real fetch happens:

@ps.query
async def user(self):
    return await fetch_user(self.user_id)

@user.initial_data
def _initial_user(self):
    return {"id": 0, "name": "Loading..."}

Mutations

While queries are for reading, mutations are for writing. They don't run automatically—you call them explicitly:

class UserState(ps.State):
    user_id: int = 1

    @ps.query
    async def user(self):
        return await fetch_user(self.user_id)

    @ps.mutation
    async def update_name(self, new_name: str):
        result = await save_user_name(self.user_id, new_name)
        self.user.invalidate()  # Refresh the query after saving
        return result

Use mutations in event handlers:

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

    async def save():
        await state.update_name("New Name")

    return ps.button("Save", onClick=save)

Handling mutation results

Mutations support success and error callbacks:

@ps.mutation
async def update_name(self, new_name: str):
    return await save_user_name(self.user_id, new_name)

@update_name.on_success
def _on_save_success(self, result):
    print(f"Saved: {result}")

@update_name.on_error
def _on_save_error(self, error):
    print(f"Save failed: {error}")

Optimistic updates

Update the display immediately before the server responds for the snappiest UI:

async def rename_user():
    # Show the change immediately (optimistic)
    state.user.set_data({"id": state.user_id, "name": "New Name"})

    # Actually save to server
    await state.update_name("New Name")

    # Refetch to make sure we're in sync
    await state.user.refetch()

Async effects

Effects can be async, useful for background tasks or periodic updates:

class PollingState(ps.State):
    data: dict = {}

    @ps.effect(lazy=True)
    async def poll(self):
        while True:
            self.data = await fetch_latest()
            await asyncio.sleep(30)

    def start_polling(self):
        self.poll.schedule()

    def stop_polling(self):
        self.poll.cancel()

The lazy=True option means the effect doesn't run automatically. Use .schedule() to start it and .cancel() to stop it.

Controlling tracking in async code

Use ps.Untrack() when you need to read state without creating a dependency:

async def save(self):
    with ps.Untrack():
        # Reading these won't cause the effect to rerun
        data = {"name": self.name, "email": self.email}

    await api.save(data)

Quick reference

I want to...	Use
Fetch data that updates automatically	`@ps.query`
Send data to the server	`@ps.mutation`
Show loading state on first load	`query.is_loading`
Show loading state on any fetch	`query.is_fetching`
Manually refresh data	`query.refetch()`
Update UI immediately before server confirms	`query.set_data(...)`
Mark data as stale for lazy refresh	`query.invalidate()`
Run background async tasks	`@ps.effect(lazy=True)` with `.schedule()`