Data & Queries
Query Invalidation
Force queries to refetch after mutations that change related data. This keeps your UI in sync when one action affects multiple parts of your app.
Recipe
import asyncio
from typing import TypedDict
import pulse as ps
class User(TypedDict):
id: int
name: str
email: str
class Profile(TypedDict):
user_id: int
bio: str
avatar_url: str
class UserState(ps.State):
user_id: int = 1
@ps.query
async def user(self) -> User:
await asyncio.sleep(0.3)
return {"id": self.user_id, "name": "Alice", "email": "alice@example.com"}
@user.key
def _user_key(self):
return ("user", self.user_id)
@ps.query
async def profile(self) -> Profile:
await asyncio.sleep(0.3)
return {"user_id": self.user_id, "bio": "Hello!", "avatar_url": "/avatar.png"}
@profile.key
def _profile_key(self):
return ("profile", self.user_id)
@ps.mutation
async def update_user(self, name: str, bio: str):
# Call your API
await asyncio.sleep(0.5)
# Invalidate both queries so they refetch with new data
self.user.invalidate()
self.profile.invalidate()
@ps.component
def EditProfile():
with ps.init():
state = UserState()
async def handle_save():
await state.update_user("Alice Updated", "New bio")
return ps.div(
ps.p(f"Name: {state.user.data['name']}" if state.user.data else "Loading..."),
ps.p(f"Bio: {state.profile.data['bio']}" if state.profile.data else ""),
ps.button("Save Changes", onClick=handle_save),
)How it works
invalidate()marks the query as stale and triggers a refetch- The query automatically fetches fresh data in the background
- The UI updates when the new data arrives
When to use invalidation vs. set_data
| Approach | Use when |
|---|---|
invalidate() | Server is the source of truth, you want fresh data |
set_data() | You know exactly what the new data should be |
| Both (optimistic) | Instant UI feedback, then reconcile with server |
Invalidating from outside the state
If you need to invalidate queries from a different component, pass the state reference or use a shared state:
class AppState(ps.State):
# ... queries defined here
# Create a global instance for shared access
app_state = ps.global_state(AppState)
@ps.component
def SomeOtherComponent():
state = app_state()
async def refresh_data():
state.user.invalidate()