-
Notifications
You must be signed in to change notification settings - Fork 602
Open
Labels
feature requestSuggestion/Request for additional featureSuggestion/Request for additional feature
Description
Duplicate Check
- I have searched the opened issues and there are no duplicates
Describe the requested feature
from typing import Any, ClassVar, Optional, Type, TypeVar
import flet as ft
from pydantic import BaseModel, ConfigDict
T = TypeVar("T")
class StateRegistry:
"""
Internal registry implementing the Session-Scoped Singleton pattern.
**Design Pattern: Session-Scoped Singleton**
This registry ensures that each state class has exactly ONE instance per user session,
while allowing multiple concurrent sessions (multi-user support).
**How it works:**
1. State classes are registered at import time (class-level registry)
2. State instances are created and stored in session.store (one per session)
3. Each session gets its own isolated instance of each state class
**Key Characteristics:**
- **Singleton per Session**: One instance per state class per session
- **Multi-Session Safe**: Different users get different instances
- **Lazy Initialization**: Instances created only when initialize() is called
- **Session Storage**: Uses ft.Page.session.store as the backing store
**Internal Infrastructure:**
This is the infrastructure layer - not meant to be inherited from.
Developers should use FtState as their base class instead.
"""
STATE_KEY_PREFIX = "_state_manager_"
_registered_state_classes: ClassVar[set[Type["FtState"]]] = set()
@classmethod
def _get_session(cls):
"""Get current session from context"""
return ft.context.page.session
@classmethod
def _get_storage_key(cls, state_type: Type[T]) -> str:
"""Generate storage key for state type"""
return f"{cls.STATE_KEY_PREFIX}{state_type.__name__}"
@classmethod
def register_class(cls, state_class: Type["FtState"]):
"""Register a state class for auto-initialization"""
cls._registered_state_classes.add(state_class)
@classmethod
def register_instance(cls, state_type: Type[T], instance: T):
"""Register a state instance in current session"""
session = cls._get_session()
storage_key = cls._get_storage_key(state_type)
session.store.set(storage_key, instance)
@classmethod
def get_instance(cls, state_type: Type[T]) -> Optional[T]:
"""Get a state instance from current session"""
try:
session = cls._get_session()
storage_key = cls._get_storage_key(state_type)
return session.store.get(storage_key)
except RuntimeError:
return None
@classmethod
def initialize(cls):
"""Initialize all registered state classes for current session"""
initialized_states = {}
for state_class in cls._registered_state_classes:
instance = state_class()
state, _ = ft.use_state(instance)
cls.register_instance(state_class, state)
initialized_states[state_class.__name__] = state
return initialized_states
class FtState(BaseModel):
"""
Base class for global application state management in Flet.
═══════════════════════════════════════════════════════════════════════════
DESIGN PATTERN: Session-Scoped Singleton
═══════════════════════════════════════════════════════════════════════════
**What is a Session-Scoped Singleton?**
Traditional Singleton Pattern:
- One instance per application (globally shared)
- Problem: In multi-user web apps, all users would share the same state
Session-Scoped Singleton Pattern:
- One instance per state class PER USER SESSION
- Each user gets their own isolated state
- Perfect for web applications with multiple concurrent users
**Implementation Details:**
1. **Class Registration** (Import Time):
- When you define a class inheriting from FtState, it auto-registers via __init_subclass__
- Registered classes are stored in StateRegistry._registered_state_classes
2. **Instance Creation** (Runtime):
- Call FtState.initialize() in your root component
- Creates ONE instance per registered state class for the CURRENT SESSION
- Stores instances in ft.Page.session.store with unique keys
3. **Instance Retrieval**:
- use_state(): Gets instance + subscribes component to updates (reactive)
- get_state(): Gets instance without subscription (non-reactive)
**Architecture Benefits:**
This class provides a robust state management solution by combining:
1. **Pydantic Models**: For strict data validation, type safety, and serialization.
2. **Flet Reactivity**: Seamless integration with Flet's @ft.observable and use_state.
3. **Session Isolation**: Each user's state is completely isolated from others.
4. **Global Access**: Access state from any component without prop drilling.
═══════════════════════════════════════════════════════════════════════════
IMPLEMENTATION GUIDE FOR DEVELOPERS
═══════════════════════════════════════════════════════════════════════════
**Step 1: Define Your State Class**
@ft.observable # Required for reactivity
class AppState(FtState):
count: int = 0
user_name: str = "Guest"
def increment(self):
self.count += 1
**Step 2: Initialize in Root Component**
def main(page: ft.Page):
# Initialize ONCE at app start
FtState.initialize()
page.add(Counter(), Profile())
**Step 3: Use in Components (Reactive)**
@ft.component
def Counter():
# Subscribes to state changes - component re-renders on updates
state = AppState.use_state()
return ft.Column([
ft.Text(f"Count: {state.count}"),
ft.ElevatedButton("Increment", on_click=lambda _: state.increment())
])
**Step 4: Use in Event Handlers (Non-Reactive)**
class UserController:
def update_profile(self, new_name: str):
# Get state without subscribing
state = AppState.get_state()
state.user_name = new_name
═══════════════════════════════════════════════════════════════════════════
ADVANCED: Multiple State Classes
═══════════════════════════════════════════════════════════════════════════
You can have multiple state classes, each as a singleton per session:
@ft.observable
class AuthState(FtState):
is_authenticated: bool = False
user_id: Optional[str] = None
@ft.observable
class CartState(FtState):
items: list[str] = []
total: float = 0.0
# Both are initialized automatically
FtState.initialize()
# Access from different components
@ft.component
def Header():
auth = AuthState.use_state()
cart = CartState.use_state()
return ft.Text(f"User: {auth.user_id}, Cart: {len(cart.items)}")
═══════════════════════════════════════════════════════════════════════════
TECHNICAL NOTES
═══════════════════════════════════════════════════════════════════════════
**Thread Safety:**
- Each session runs in its own context
- Session.store is thread-safe by design
- No shared mutable state between sessions
**Memory Management:**
- State instances are garbage collected when session ends
- Session.store is cleared automatically by Flet
**Testing:**
- Mock ft.Page.session.store for unit tests
- Each test can create isolated sessions
**Potential Flet Core Integration:**
This pattern is a strong candidate for integration into Flet core, possibly as
`flet.state` or `flet.data`. The StateRegistry logic could be internalized into
ft.Page or ft.Session for native session-scoped state management.
"""
model_config = ConfigDict(
arbitrary_types_allowed=True,
validate_assignment=True,
extra="allow",
)
def __init_subclass__(cls, **kwargs):
"""Auto-register when inherited"""
super().__init_subclass__(**kwargs)
StateRegistry.register_class(cls)
@classmethod
def use_state(cls: Type[T]) -> T:
"""
Returns the reactive state instance for the current session.
When called inside a @ft.component, it subscribes the component to state changes.
Usage Example:
@ft.component
def MyComponent():
# Subscribes this component to updates
state = AppState.use_state()
return ft.Text(f"Count: {state.count}")
"""
state = StateRegistry.get_instance(cls)
if state is None:
raise ValueError(
f"State {cls.__name__} not registered. "
f"Did you call State.initialize() in your root component?"
)
_state, _ = ft.use_state(state)
# return _state
return _state
@classmethod
def get_state(cls: Type[T]) -> T:
"""
Returns the state instance without subscribing to updates.
Ideal for use inside event handlers, other classes, or methods where reactivity is not required.
Usage Example:
class UserController:
def update_name(self, new_name):
# Access state without subscribing
state = AppState.get_state()
state.name = new_name
"""
state = StateRegistry.get_instance(cls)
if state is None:
raise RuntimeError(
f"State {cls.__name__} not initialized. "
f"Make sure State.initialize() was called in your root component."
)
return state
@classmethod
def initialize(cls):
"""
Initializes all registered state classes for the current session.
Must be called once in the root component of the application.
"""
return StateRegistry.initialize()Suggest a solution
No response
Screenshots
No response
Additional details
No response
Metadata
Metadata
Assignees
Labels
feature requestSuggestion/Request for additional featureSuggestion/Request for additional feature