cjm-fasthtml-interactions

Install

pip install cjm_fasthtml_interactions

Project Structure

nbs/
├── core/ (3)
│   ├── context.ipynb      # Context management for interaction patterns providing access to state, request, and custom data
│   ├── html_ids.ipynb     # Centralized HTML ID constants for interaction pattern components
│   └── state_store.ipynb  # Server-side workflow state storage implementations
└── patterns/ (2)
    ├── async_loading.ipynb  # Pattern for asynchronous content loading with skeleton loaders and loading indicators
    └── step_flow.ipynb      # Multi-step wizard pattern with state management, navigation, and route generation

Total: 5 notebooks across 2 directories

Module Dependencies

graph LR
    core_context[core.context<br/>Interaction Context]
    core_html_ids[core.html_ids<br/>HTML IDs]
    core_state_store[core.state_store<br/>Workflow State Store]
    patterns_async_loading[patterns.async_loading<br/>Async Loading Container]
    patterns_step_flow[patterns.step_flow<br/>Step Flow]

    patterns_step_flow --> core_context
    patterns_step_flow --> core_state_store
    patterns_step_flow --> core_html_ids

3 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Async Loading Container (async_loading.ipynb)

Pattern for asynchronous content loading with skeleton loaders and loading indicators

Import

from cjm_fasthtml_interactions.patterns.async_loading import (
    LoadingType,
    AsyncLoadingContainer
)

Functions

def AsyncLoadingContainer(
    container_id: str,  # HTML ID for the container
    load_url: str,  # URL to fetch content from
    loading_type: LoadingType = LoadingType.SPINNER,  # Type of loading indicator
    loading_size: str = "lg",  # Size of loading indicator (xs, sm, md, lg)
    loading_message: Optional[str] = None,  # Optional message to display while loading
    skeleton_content: Optional[Any] = None,  # Optional skeleton/placeholder content
    trigger: str = "load",  # HTMX trigger event (default: load on page load)
    swap: str = "outerHTML",  # HTMX swap method (default: replace entire container)
    container_cls: Optional[str] = None,  # Additional CSS classes for container
    **kwargs  # Additional attributes for the container
) -> FT:  # Div element with async loading configured
    "Create a container that asynchronously loads content from a URL."

Classes

class LoadingType(Enum):
    "Types of loading indicators for async content."

Interaction Context (context.ipynb)

Context management for interaction patterns providing access to state, request, and custom data

Import

from cjm_fasthtml_interactions.core.context import (
    InteractionContext
)

Classes

@dataclass
class InteractionContext:
    "Context for interaction patterns providing access to state, request, and custom data."
    
    state: Dict[str, Any] = field(...)  # Workflow state
    request: Optional[Any]  # FastHTML request object
    session: Optional[Any]  # FastHTML session object
    data: Dict[str, Any] = field(...)  # Custom data from data loaders
    metadata: Dict[str, Any] = field(...)  # Additional metadata
    
    def get(self,
                key: str,  # Key to retrieve from state
                default: Any = None  # Default value if key not found
               ) -> Any:  # Value from state or default
        "Get value from workflow state."
    
    def get_data(self,
                     key: str,  # Key to retrieve from data
                     default: Any = None  # Default value if key not found
                    ) -> Any:  # Value from data or default
        "Get value from custom data."
    
    def has(self,
                key: str  # Key to check in state
               ) -> bool:  # True if key exists in state
        "Check if key exists in workflow state."
    
    def set(self,
                key: str,  # Key to set in state
                value: Any  # Value to store
               ) -> None
        "Set value in workflow state."
    
    def get_all_state(self) -> Dict[str, Any]:  # All workflow state
            """Get all workflow state as dictionary."""
            return self.state.copy()
        
        def update_state(self, 
                         updates: Dict[str, Any]  # State updates to apply
                        ) -> None
        "Get all workflow state as dictionary."
    
    def update_state(self,
                         updates: Dict[str, Any]  # State updates to apply
                        ) -> None
        "Update multiple state values at once."

HTML IDs (html_ids.ipynb)

Centralized HTML ID constants for interaction pattern components

Import

from cjm_fasthtml_interactions.core.html_ids import (
    InteractionHtmlIds
)

Classes

class InteractionHtmlIds(AppHtmlIds):
    """
    HTML ID constants for interaction pattern components.
    
    Inherits from AppHtmlIds:
        - MAIN_CONTENT = "main-content"
        - as_selector(id_str) - static method
    """
    
    def step_content(step_id: str  # Step identifier
                        ) -> str:  # HTML ID for step content
        "Generate HTML ID for a specific step's content."
    
    def step_indicator(step_id: str  # Step identifier
                          ) -> str:  # HTML ID for step indicator
        "Generate HTML ID for a specific step's progress indicator."

Workflow State Store (state_store.ipynb)

Server-side workflow state storage implementations

Import

from cjm_fasthtml_interactions.core.state_store import (
    WorkflowStateStore,
    get_session_id,
    set_session_id,
    InMemoryWorkflowStateStore
)

Functions

def get_session_id(
    sess: Any,  # FastHTML session object
    key: str = "_workflow_session_id"  # Session key for storing the ID
) -> str:  # Stable session identifier
    "Get or create a stable session identifier."

def set_session_id(
    sess: Any,  # FastHTML session object
    session_id: str,  # Workflow session ID to set as active
    key: str = "_workflow_session_id"  # Session key for storing the ID
) -> None
    "Set the active workflow session ID, switching all subsequent state reads/writes to that session."

Classes

@runtime_checkable
class WorkflowStateStore(Protocol):
    "Protocol for workflow state storage backends."
    
    def get_current_step(self,
                             flow_id: str,  # Workflow identifier
                             sess: Any  # FastHTML session object
                            ) -> Optional[str]:  # Current step ID or None
        "Get current step ID for a workflow."
    
    def set_current_step(self,
                             flow_id: str,  # Workflow identifier
                             sess: Any,  # FastHTML session object
                             step_id: str  # Step ID to set as current
                            ) -> None
        "Set current step ID for a workflow."
    
    def get_state(self,
                      flow_id: str,  # Workflow identifier
                      sess: Any  # FastHTML session object
                     ) -> Dict[str, Any]:  # Workflow state dictionary
        "Get all workflow state."
    
    def update_state(self,
                         flow_id: str,  # Workflow identifier
                         sess: Any,  # FastHTML session object
                         updates: Dict[str, Any]  # State updates to apply
                        ) -> None
        "Update workflow state with new values."
    
    def clear_state(self,
                        flow_id: str,  # Workflow identifier
                        sess: Any  # FastHTML session object
                       ) -> None
        "Clear all workflow state."

class InMemoryWorkflowStateStore:
    def __init__(self):
        """Initialize empty state storage."""
        self._current_steps: Dict[str, str] = {}  # {flow_id:session_id -> step_id}
    "In-memory workflow state storage for development and testing."
    
    def __init__(self):
            """Initialize empty state storage."""
            self._current_steps: Dict[str, str] = {}  # {flow_id:session_id -> step_id}
        "Initialize empty state storage."
    
    def get_current_step(self,
                             flow_id: str,  # Workflow identifier
                             sess: Any  # FastHTML session object
                            ) -> Optional[str]:  # Current step ID or None
        "Get current step ID for a workflow."
    
    def set_current_step(self,
                             flow_id: str,  # Workflow identifier
                             sess: Any,  # FastHTML session object
                             step_id: str  # Step ID to set as current
                            ) -> None
        "Set current step ID for a workflow."
    
    def get_state(self,
                      flow_id: str,  # Workflow identifier
                      sess: Any  # FastHTML session object
                     ) -> Dict[str, Any]:  # Workflow state dictionary
        "Get all workflow state."
    
    def update_state(self,
                         flow_id: str,  # Workflow identifier
                         sess: Any,  # FastHTML session object
                         updates: Dict[str, Any]  # State updates to apply
                        ) -> None
        "Update workflow state with new values."
    
    def clear_state(self,
                        flow_id: str,  # Workflow identifier
                        sess: Any  # FastHTML session object
                       ) -> None
        "Clear all workflow state."

Step Flow (step_flow.ipynb)

Multi-step wizard pattern with state management, navigation, and route generation

Import

from cjm_fasthtml_interactions.patterns.step_flow import (
    Step,
    StepFlow
)

Functions

@patch
def get_step(self:StepFlow, 
             step_id: str  # Step identifier
            ) -> Optional[Step]:  # Step object or None
    "Get step by ID."

@patch
def get_step_index(self:StepFlow, 
                   step_id: str  # Step identifier
                  ) -> Optional[int]:  # Step index or None
    "Get step index by ID."

@patch
def get_current_step_id(self:StepFlow, 
                        sess: Any  # FastHTML session object
                       ) -> str:  # Current step ID
    "Get current step ID from state store."

@patch
def set_current_step(self:StepFlow, 
                     sess: Any,  # FastHTML session object
                     step_id: str  # Step ID to set as current
                    ) -> None
    "Set current step in state store."

@patch
def get_next_step_id(self:StepFlow, 
                     current_step_id: str  # Current step ID
                    ) -> Optional[str]:  # Next step ID or None if last step
    "Get the ID of the next step."

@patch
def get_previous_step_id(self:StepFlow, 
                         current_step_id: str  # Current step ID
                        ) -> Optional[str]:  # Previous step ID or None if first step
    "Get the ID of the previous step."

@patch
def is_last_step(self:StepFlow, 
                 step_id: str  # Step ID to check
                ) -> bool:  # True if this is the last step
    "Check if step is the last step."

@patch
def is_first_step(self:StepFlow, 
                  step_id: str  # Step ID to check
                 ) -> bool:  # True if this is the first step
    "Check if step is the first step."

@patch
def get_workflow_state(self:StepFlow, 
                       sess: Any  # FastHTML session object
                      ) -> Dict[str, Any]:  # All workflow state
    "Get all workflow state from state store."

@patch
def update_workflow_state(self:StepFlow, 
                          sess: Any,  # FastHTML session object
                          updates: Dict[str, Any]  # State updates
                         ) -> None
    "Update workflow state with new values."

@patch
def clear_workflow(self:StepFlow, 
                   sess: Any  # FastHTML session object
                  ) -> None
    "Clear all workflow state."

@patch
def _summarize_state(self:StepFlow, 
                     state: Dict[str, Any]  # State dictionary to summarize
                    ) -> str:  # Human-readable summary string
    "Create a concise summary of state for debug output."

@patch
def create_context(self:StepFlow, 
                   request: Any,  # FastHTML request object
                   sess: Any,  # FastHTML session object
                   step: Step  # Current step
                  ) -> InteractionContext:  # Interaction context for rendering
    "Create interaction context for a step."

@patch
def render_progress(self:StepFlow, 
                    sess: Any  # FastHTML session object
                   ) -> FT:  # Progress indicator or empty Div
    "Render progress indicator showing all steps."

@patch
def render_step_content(self:StepFlow,
                        step_obj: Step,  # Step to render
                        ctx: InteractionContext,  # Interaction context
                        next_route: str,  # Route for next/submit
                        back_route: Optional[str] = None,  # Route for back
                        cancel_route: Optional[str] = None  # Route for cancel
                       ) -> FT:  # Complete step content with optional progress and navigation
    "Render step content with optional progress indicator and navigation."

@patch
def render_navigation(self:StepFlow,
                      step_id: str,  # Current step ID
                      next_route: str,  # Route for next/submit action
                      back_route: Optional[str] = None,  # Route for back action
                      cancel_route: Optional[str] = None,  # Route for cancel action
                     ) -> FT:  # Navigation button container
    "Render navigation buttons for a step."

@patch
def create_router(self:StepFlow,
                  prefix: str = ""  # URL prefix for routes (e.g., "/transcription")
                 ) -> APIRouter:  # APIRouter with generated routes
    "Create FastHTML router with generated routes for this flow."

Classes

@dataclass
class Step:
    "Definition of a single step in a multi-step workflow."
    
    id: str  # Unique step identifier (used in URLs)
    title: str  # Display title for the step
    render: Callable[[InteractionContext], Any]  # Function to render step UI
    validate: Optional[Callable[[Dict[str, Any]], bool]]  # Validation function
    data_loader: Optional[Callable[[Any], Dict[str, Any]]]  # Data loading function
    data_keys: List[str] = field(...)  # State keys managed by this step
    can_skip: bool = False  # Whether this step can be skipped
    show_back: bool = True  # Whether to show back button
    show_cancel: bool = True  # Whether to show cancel button
    next_button_text: str = 'Continue'  # Text for next/submit button
    on_enter: Optional[Callable[[Dict[str, Any], Any, Any], Any]]  # Called when entering step, before render (state, request, sess) -> None or component
    on_leave: Optional[Callable[[Dict[str, Any], Any, Any], Any]]  # Called after validation, before navigation (state, request, sess) -> None or component
    
    def is_valid(self, state: Dict[str, Any]  # Current workflow state
                    ) -> bool:  # True if step is complete and valid
        "Check if step has valid data in state."

class StepFlow:
    def __init__(
        self,
        flow_id: str,  # Unique identifier for this workflow
        steps: List[Step],  # List of step definitions
        state_store: Optional[WorkflowStateStore] = None,  # Storage backend (defaults to InMemoryWorkflowStateStore)
        container_id: str = InteractionHtmlIds.STEP_FLOW_CONTAINER,  # HTML ID for content container
        on_complete: Optional[Callable[[Dict[str, Any], Any], Any]] = None,  # Completion handler
        show_progress: bool = False,  # Whether to show progress indicator
        progress_renderer: Optional[Callable] = None,  # Custom progress renderer: (steps, current_index) -> FT
        wrap_in_form: bool = True,  # Whether to wrap content + navigation in a form
        debug: bool = False  # Whether to print debug information
    )
    "Manage multi-step workflows with automatic route generation and state management."
    
    def __init__(
            self,
            flow_id: str,  # Unique identifier for this workflow
            steps: List[Step],  # List of step definitions
            state_store: Optional[WorkflowStateStore] = None,  # Storage backend (defaults to InMemoryWorkflowStateStore)
            container_id: str = InteractionHtmlIds.STEP_FLOW_CONTAINER,  # HTML ID for content container
            on_complete: Optional[Callable[[Dict[str, Any], Any], Any]] = None,  # Completion handler
            show_progress: bool = False,  # Whether to show progress indicator
            progress_renderer: Optional[Callable] = None,  # Custom progress renderer: (steps, current_index) -> FT
            wrap_in_form: bool = True,  # Whether to wrap content + navigation in a form
            debug: bool = False  # Whether to print debug information
        )
        "Initialize step flow manager."