Skip to content

feature: Session-Scoped Singleton Pattern - FtState #5960

@iBuitron

Description

@iBuitron

Duplicate Check

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

No one assigned

    Labels

    feature requestSuggestion/Request for additional feature

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions